status-indicators/midi_witness.py

#!/usr/bin/env python3
"""MIDI witness: writes down the program changes (Waves Tune keys) and other channel messages a Mac really
receives on its network MIDI inputs, so the status panel can check that what the tracks sent arrived.

Runs on the receiving Mac as a LaunchAgent (org.waterschurch.midi-witness), next to Pro Tools: CoreMIDI
gives every input port its own copy, so nothing is taken away from the app that uses the input.

  midi_witness.py "Network Tracks Midi" [...]      (Studio B: what the NA Tracks Mac sends)

Writes ~/Library/Application Support/status-panel/midi-witness.json once a second when something changed:
  {"sources": {"Network Tracks Midi": {"connected": true, "last": 1793..., "count": 12}},
   "pcs": [[time, source, channel, program], ...]}    (the last 50 program changes)
Built 2026-10-26 for NA: the key Ableton sends must really change Waves Tune at FOH and in Studio B.
"""
import ctypes
import json
import os
import sys
import time
from collections import deque

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import coremidi  # noqa: E402

OUT = os.path.expanduser("~/Library/Application Support/status-panel/midi-witness.json")
LENGTHS = {0x80: 3, 0x90: 3, 0xA0: 3, 0xB0: 3, 0xC0: 2, 0xD0: 2, 0xE0: 3}


def messages(data):
    """Channel messages in a packet's bytes: [(status, data bytes)]; system real-time and sysex skipped."""
    out, i = [], 0
    while i < len(data):
        b = data[i]
        if b >= 0xF8:                       # clock, active sensing...
            i += 1
        elif b == 0xF0:                     # sysex: to its end
            j = data.find(b"\xf7", i)
            i = len(data) if j < 0 else j + 1
        elif b >= 0xF0:
            i += {0xF1: 2, 0xF2: 3, 0xF3: 2}.get(b, 1)
        elif b & 0x80:
            n = LENGTHS[b & 0xF0]
            out.append((b, data[i + 1:i + n]))
            i += n
        else:
            i += 1                          # a stray data byte
    return out


class Witness:
    def __init__(self, names, out=OUT, now=time.time):
        self.names, self.out, self.now = names, out, now
        self.pcs = deque(maxlen=50)
        self.sources = {n: {"connected": False, "last": None, "count": 0} for n in names}
        self.endpoints, self.port, self.dirty = {}, None, True
        self.proc = coremidi._READPROC(self.on_packets)            # keep a reference
        self.by_ref = {}

    def take(self, name, data):
        for status, body in messages(data):
            src = self.sources[name]
            src["last"], src["count"] = self.now(), src["count"] + 1
            if status & 0xF0 == 0xC0 and body:
                self.pcs.append([round(self.now(), 3), name, (status & 0x0F) + 1, body[0]])
            self.dirty = True

    def on_packets(self, pktlist, _ref, src_ref):
        try:
            name = self.by_ref.get(src_ref or 0)
            if name is None:
                return
            pl = pktlist.contents
            addr = ctypes.addressof(pl.packet[0])
            for _ in range(pl.numPackets):
                pkt = coremidi._MIDIPacket.from_address(addr)
                n = min(pkt.length, 256)
                self.take(name, bytes(pkt.data[:n]))
                addr = (addr + 10 + pkt.length + 3) & ~3                 # MIDIPacketNext
        except Exception:  # noqa: BLE001 - never raise into CoreMIDI's thread
            pass

    def poll(self):
        """(Re)connect each named source; a network session that comes and goes is picked up again."""
        with coremidi._lock:
            if self.port is None:
                ref = coremidi._U32()
                cname = coremidi._cfstr("midi witness")
                err = coremidi._cm.MIDIInputPortCreate(coremidi._get_client(), cname, self.proc, None, ctypes.byref(ref))
                coremidi._cf.CFRelease(cname)
                if err:
                    raise OSError("MIDIInputPortCreate failed: %d" % err)
                self.port = ref.value
            found = {}
            for i in range(coremidi._cm.MIDIGetNumberOfSources()):
                ep = coremidi._cm.MIDIGetSource(i)
                for n in self.names:
                    if n in (coremidi._string_prop(ep, "displayName"), coremidi._string_prop(ep, "name")):
                        found[n] = ep
            for n in self.names:
                ep = found.get(n)
                if ep == self.endpoints.get(n):
                    continue
                if self.endpoints.get(n) is not None:
                    coremidi._cm.MIDIPortDisconnectSource(self.port, self.endpoints[n])
                self.endpoints[n] = None
                if ep is not None:
                    ref = len(self.by_ref) + 1
                    self.by_ref[ref] = n
                    if coremidi._cm.MIDIPortConnectSource(self.port, ep, ctypes.c_void_p(ref)) == 0:
                        self.endpoints[n] = ep
                self.sources[n]["connected"] = self.endpoints[n] is not None
                self.dirty = True

    def save(self):
        if not self.dirty:
            return
        os.makedirs(os.path.dirname(self.out), exist_ok=True)
        with open(self.out + ".tmp", "w") as f:
            json.dump({"time": self.now(), "sources": self.sources, "pcs": list(self.pcs)}, f)
        os.replace(self.out + ".tmp", self.out)
        self.dirty = False


def main():
    names = sys.argv[1:] or ["Network Tracks Midi"]
    w, polled = Witness(names), 0.0
    while True:
        if time.time() - polled > 5:
            try:
                w.poll()
            except OSError:
                pass
            polled = time.time()
        w.save()
        time.sleep(1)


if __name__ == "__main__":
    main()