christian/status-indicators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
80fd89d03bdaa14d4fba868fdac60254b46ea090
status-indicators/coremidi.py
T
christianandClaude Opus 5.5 80fd89d03b Hold MIDI reconnects a dropped peer from the Mac side 
macOS's MIDINetworkSession API only reaches the default session (blank and
disabled on NA Ingest and NA Tracks), so coremidi.connect_peer writes the
named session's "apple.midirtp.session" settings with the peer added, as
Audio MIDI Setup's Connect does. Existing peers are passed back unchanged.
The watchdog remembers each peer's last address (config "address" is the
fallback) and runs it locally for NA Ingest's session or over SSH on the
Tracks Mac for NA Tracks' (FX lights). Checked live: both Macs answer
"already" while the Lighting PC is connected.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-08 15:54:53 -04:00

329 lines
12 KiB
Python
Raw Blame History

"""Minimal CoreMIDI access through ctypes (no PyObjC, stdlib only).
Two things the status panel needs that Companion cannot see:
* sessions() - the live Network MIDI (RTP-MIDI) sessions on this Mac and the
peers currently joined to each, read from the network driver's
"apple.midirtp.session" entity property.
* Listener - timestamps the last message that arrived on a named MIDI
source, so a tile can say "last MIDI 3s ago".
* connect_peer() - re-add a peer to a named session, the way Audio MIDI Setup's
"Connect" does (macOS's MIDINetworkSession API only reaches the
default session, not named ones like "NA Tracks").
"""
import ctypes
import ctypes.util
import plistlib
import threading
import time
_cm = ctypes.CDLL(ctypes.util.find_library("CoreMIDI"))
_cf = ctypes.CDLL(ctypes.util.find_library("CoreFoundation"))
_U32 = ctypes.c_uint32
_UTF8 = 0x08000100
_cm.MIDIGetNumberOfDevices.restype = ctypes.c_ulong
_cm.MIDIGetDevice.restype = _U32
_cm.MIDIGetDevice.argtypes = [ctypes.c_ulong]
_cm.MIDIDeviceGetNumberOfEntities.restype = ctypes.c_ulong
_cm.MIDIDeviceGetNumberOfEntities.argtypes = [_U32]
_cm.MIDIDeviceGetEntity.restype = _U32
_cm.MIDIDeviceGetEntity.argtypes = [_U32, ctypes.c_ulong]
_cm.MIDIGetNumberOfSources.restype = ctypes.c_ulong
_cm.MIDIGetSource.restype = _U32
_cm.MIDIGetSource.argtypes = [ctypes.c_ulong]
_cm.MIDIObjectGetStringProperty.argtypes = [_U32, ctypes.c_void_p, ctypes.POINTER(ctypes.c_void_p)]
_cm.MIDIObjectGetDictionaryProperty.argtypes = [_U32, ctypes.c_void_p, ctypes.POINTER(ctypes.c_void_p)]
_cm.MIDIObjectGetIntegerProperty.argtypes = [_U32, ctypes.c_void_p, ctypes.POINTER(ctypes.c_int32)]
_cm.MIDIObjectSetDictionaryProperty.argtypes = [_U32, ctypes.c_void_p, ctypes.c_void_p]
_cf.CFStringCreateWithCString.restype = ctypes.c_void_p
_cf.CFStringCreateWithCString.argtypes = [ctypes.c_void_p, ctypes.c_char_p, _U32]
_cf.CFStringGetCString.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_long, _U32]
_cf.CFRelease.argtypes = [ctypes.c_void_p]
_cf.CFPropertyListCreateData.restype = ctypes.c_void_p
_cf.CFPropertyListCreateData.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_long, ctypes.c_ulong, ctypes.c_void_p]
_cf.CFDataGetLength.restype = ctypes.c_long
_cf.CFDataGetLength.argtypes = [ctypes.c_void_p]
_cf.CFDataGetBytePtr.restype = ctypes.c_void_p
_cf.CFDataGetBytePtr.argtypes = [ctypes.c_void_p]
_cf.CFDataCreate.restype = ctypes.c_void_p
_cf.CFDataCreate.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_long]
_cf.CFPropertyListCreateWithData.restype = ctypes.c_void_p
_cf.CFPropertyListCreateWithData.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_ulong,
ctypes.c_void_p, ctypes.c_void_p]
# One lock around every CoreMIDI call: concurrent first calls from two threads
# (session reads vs. MIDIClientCreate) intermittently fail with -2.
_lock = threading.RLock()
def _cfstr(s):
return _cf.CFStringCreateWithCString(None, s.encode("utf-8"), _UTF8)
def _string_prop(obj, key):
out = ctypes.c_void_p()
k = _cfstr(key)
err = _cm.MIDIObjectGetStringProperty(obj, k, ctypes.byref(out))
_cf.CFRelease(k)
if err or not out.value:
return None
buf = ctypes.create_string_buffer(1024)
_cf.CFStringGetCString(out, buf, 1024, _UTF8)
_cf.CFRelease(out)
return buf.value.decode("utf-8", "replace")
def _dict_prop(obj, key):
out = ctypes.c_void_p()
k = _cfstr(key)
err = _cm.MIDIObjectGetDictionaryProperty(obj, k, ctypes.byref(out))
_cf.CFRelease(k)
if err or not out.value:
return None
data = _cf.CFPropertyListCreateData(None, out, 100, 0, None) # 100 = XML format
raw = ctypes.string_at(_cf.CFDataGetBytePtr(data), _cf.CFDataGetLength(data))
_cf.CFRelease(data)
_cf.CFRelease(out)
return plistlib.loads(raw)
def _set_dict_prop(obj, key, value):
raw = plistlib.dumps(value)
data = _cf.CFDataCreate(None, raw, len(raw))
plist = _cf.CFPropertyListCreateWithData(None, data, 0, None, None)
_cf.CFRelease(data)
if not plist:
raise OSError("could not build the property list")
k = _cfstr(key)
err = _cm.MIDIObjectSetDictionaryProperty(obj, k, plist)
_cf.CFRelease(k)
_cf.CFRelease(plist)
return err
def _session_entities():
"""[(entity, its "apple.midirtp.session" dict)] for every network MIDI session on this Mac."""
out = []
for i in range(_cm.MIDIGetNumberOfDevices()):
dev = _cm.MIDIGetDevice(i)
if _string_prop(dev, "driver") != "com.apple.AppleMIDINetworkDriver":
continue
for j in range(_cm.MIDIDeviceGetNumberOfEntities(dev)):
ent = _cm.MIDIDeviceGetEntity(dev, j)
sess = _dict_prop(ent, "apple.midirtp.session")
if sess is not None:
out.append((ent, sess))
return out
def with_peer(sess, peer_name, address):
"""The session settings with this peer added, or None if it is already in the session.
Peers already there are passed back exactly as read, so they are left alone."""
peers = list(sess.get("peers", []))
if any(p.get("name") == peer_name or p.get("address") == address for p in peers):
return None
new = dict(sess)
new["peers"] = peers + [{"address": address, "name": peer_name}]
return new
def connect_peer(session_name, peer_name, address):
"""Ask macOS to connect `peer_name` at `address` ("10.1.10.206:5004") to the named session,
like Audio MIDI Setup's Connect. Returns "already" (it is in the session) or "asked"."""
with _lock:
for ent, sess in _session_entities():
if (sess.get("dns-sd-name") or _string_prop(ent, "name")) != session_name:
continue
new = with_peer(sess, peer_name, address)
if new is None:
return "already"
err = _set_dict_prop(ent, "apple.midirtp.session", new)
if err:
raise OSError("MIDIObjectSetDictionaryProperty failed: %d" % err)
return "asked"
raise LookupError("no network MIDI session named %r on this Mac" % session_name)
def sessions():
"""{session name: {"port": int, "peers": [{"name", "address", "state"}]}}"""
found = {}
with _lock:
for i in range(_cm.MIDIGetNumberOfDevices()):
dev = _cm.MIDIGetDevice(i)
if _string_prop(dev, "driver") != "com.apple.AppleMIDINetworkDriver":
continue
for j in range(_cm.MIDIDeviceGetNumberOfEntities(dev)):
ent = _cm.MIDIDeviceGetEntity(dev, j)
sess = _dict_prop(ent, "apple.midirtp.session")
if sess is None:
continue
name = sess.get("dns-sd-name") or _string_prop(ent, "name") or "?"
found[name] = {
"port": sess.get("port"),
"peers": [
{"name": p.get("name", ""), "address": p.get("address", ""), "state": p.get("state", "")}
for p in sess.get("peers", [])
],
}
return found
def source_names():
with _lock:
return [_string_prop(_cm.MIDIGetSource(i), "displayName") or "" for i in range(_cm.MIDIGetNumberOfSources())]
# --- Listener -----------------------------------------------------------
class _MIDIPacket(ctypes.Structure):
_pack_ = 4
_fields_ = [("timeStamp", ctypes.c_uint64), ("length", ctypes.c_uint16), ("data", ctypes.c_ubyte * 256)]
class _MIDIPacketList(ctypes.Structure):
_pack_ = 4
_fields_ = [("numPackets", ctypes.c_uint32), ("packet", _MIDIPacket * 1)]
_READPROC = ctypes.CFUNCTYPE(None, ctypes.POINTER(_MIDIPacketList), ctypes.c_void_p, ctypes.c_void_p)
_cm.MIDIClientCreate.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p, ctypes.POINTER(_U32)]
_cm.MIDIInputPortCreate.argtypes = [_U32, ctypes.c_void_p, _READPROC, ctypes.c_void_p, ctypes.POINTER(_U32)]
_cm.MIDIPortConnectSource.argtypes = [_U32, _U32, ctypes.c_void_p]
_cm.MIDIPortDisconnectSource.argtypes = [_U32, _U32]
_client = None
def _get_client():
global _client
with _lock:
if _client is None:
ref = _U32()
name = _cfstr("status-panel")
err = _cm.MIDIClientCreate(name, None, None, ctypes.byref(ref))
_cf.CFRelease(name)
if err:
raise OSError("MIDIClientCreate failed: %d" % err)
_client = ref.value
return _client
def describe(data):
"""Short human label for a MIDI message: 'NOTE 36', 'CC 7=64', 'PC 5', 'MTC'."""
if not data:
return ""
st = data[0] & 0xF0
if st == 0x90 and len(data) >= 3 and data[2]:
return "NOTE %d" % data[1]
if st in (0x80, 0x90) and len(data) >= 2:
return "OFF %d" % data[1]
if st == 0xB0 and len(data) >= 3:
return "CC %d=%d" % (data[1], data[2])
if st == 0xC0 and len(data) >= 2:
return "PC %d" % data[1]
if data[0] == 0xF1:
return "MTC"
if data[0] == 0xF0:
return "SYSEX"
return "0x%02X" % data[0]
class Listener:
"""Connects an input port to the named source (re-checked on poll())."""
def __init__(self, source_name):
self.source_name = source_name
self.last_time = None
self.last_msg = ""
self.count = 0
self._endpoint = None
self._port = None
self._proc = _READPROC(self._on_packets) # keep a reference or it is collected
def _on_packets(self, pktlist, _ref, _src):
try:
pl = pktlist.contents
if pl.numPackets:
pkt = pl.packet[0]
n = min(pkt.length, 256)
data = bytes(pkt.data[:n])
# Skip active sensing / clock so they do not mask real traffic.
if data and data[0] not in (0xF8, 0xFE):
self.last_msg = describe(data)
self.last_time = time.time()
self.count += 1
except Exception: # never raise into CoreMIDI's thread
pass
def _find(self):
with _lock:
for i in range(_cm.MIDIGetNumberOfSources()):
ep = _cm.MIDIGetSource(i)
if _string_prop(ep, "displayName") == self.source_name or _string_prop(ep, "name") == self.source_name:
return ep
return None
@property
def connected(self):
return self._endpoint is not None
def poll(self):
"""(Re)connect if the source appeared, disappeared or changed."""
with _lock:
self._poll()
def _poll(self):
ep = self._find()
if ep == self._endpoint:
return
if self._port is None:
ref = _U32()
name = _cfstr("status-panel in")
err = _cm.MIDIInputPortCreate(_get_client(), name, self._proc, None, ctypes.byref(ref))
_cf.CFRelease(name)
if err:
raise OSError("MIDIInputPortCreate failed: %d" % err)
self._port = ref.value
if self._endpoint is not None:
_cm.MIDIPortDisconnectSource(self._port, self._endpoint)
self._endpoint = None
if ep is not None and _cm.MIDIPortConnectSource(self._port, ep, None) == 0:
self._endpoint = ep
# --- test helper: a virtual source we can send into ----------------------
_cm.MIDISourceCreate.argtypes = [_U32, ctypes.c_void_p, ctypes.POINTER(_U32)]
_cm.MIDIReceived.argtypes = [_U32, ctypes.POINTER(_MIDIPacketList)]
_cm.MIDIEndpointDispose.argtypes = [_U32]
class VirtualSource:
"""Only used by the tests: a source whose messages a Listener can receive."""
def __init__(self, name):
ref = _U32()
cname = _cfstr(name)
with _lock:
err = _cm.MIDISourceCreate(_get_client(), cname, ctypes.byref(ref))
_cf.CFRelease(cname)
if err:
raise OSError("MIDISourceCreate failed: %d" % err)
self.ref = ref.value
def send(self, data):
pl = _MIDIPacketList()
pl.numPackets = 1
pl.packet[0].timeStamp = 0
pl.packet[0].length = len(data)
for i, b in enumerate(data):
pl.packet[0].data[i] = b
_cm.MIDIReceived(self.ref, ctypes.byref(pl))
def close(self):
_cm.MIDIEndpointDispose(self.ref)
Reference in New Issue View Git Blame Copy Permalink