status-indicators/coremidi.py

"""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)