"""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". """ 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)] _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] # 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 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)