christianandClaude Opus 5.5
246500f965
status_panel.py runs on NA Ingest as a LaunchAgent. Every second it reads Companion's connection statuses and module variables, adds what Companion cannot see (network MIDI peers, Bonjour sessions, pings, launchd services, disk, Resi cloud, the Bible TV app), and writes one JSON custom variable that page 19 renders. Buttons acknowledge alarms, mute them, and restart a stuck Companion connection on a two second hold. If the watchdog stops, the page turns purple instead of freezing on stale green. make_page.py builds the page export from config.json; tests/ runs offline against a fake Companion and a CoreMIDI loopback. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
263 lines
9.3 KiB
Python
263 lines
9.3 KiB
Python
"""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)
|