From 246500f965c3891ac21a602aba9c4480be0f634e Mon Sep 17 00:00:00 2001 From: christian Date: Thu, 8 Oct 2026 09:00:56 -0400 Subject: [PATCH] Add the Companion status page and its watchdog for NA Ingest. 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 --- .gitignore | 3 + README.md | 122 ++++ config.json | 99 +++ coremidi.py | 262 ++++++++ install.sh | 59 ++ make_page.py | 168 +++++ status-page.companionconfig | Bin 0 -> 4785 bytes status_panel.py | 1151 +++++++++++++++++++++++++++++++++++ tests/test_status_panel.py | 617 +++++++++++++++++++ 9 files changed, 2481 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 config.json create mode 100644 coremidi.py create mode 100644 install.sh create mode 100644 make_page.py create mode 100644 status-page.companionconfig create mode 100644 status_panel.py create mode 100644 tests/test_status_panel.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b908d4c --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.pyc +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..a573da2 --- /dev/null +++ b/README.md @@ -0,0 +1,122 @@ +# Status Indicators — Companion STATUS page for NA Ingest + +One Companion page (page 19, **"19 STATUS"**) that is solid green when everything is fine and **blinks red with the actual problem** ("HD2 NO DISK", "LIGHT PC MIDI LOST") when something needs attention. + +There are two parts: + +- **`status_panel.py`** — a watchdog that runs on NA Ingest as the LaunchAgent `org.waterschurch.status-panel`. Every second it reads Companion's connection statuses and module variables. It also checks what Companion can't see: network MIDI peers, Bonjour MIDI sessions, pings, launchd services, disk space, and whether the Resi cloud is reachable. It writes the result into one Companion custom variable. +- **`status-page.companionconfig`** — the page. Every button just draws what the watchdog wrote and blinks on your existing `flash_tick` trigger. + +## Layout (Stream Deck XL; MK.2 and Mini show the top-left corner) + +| | col 1 | col 2 | col 3 | col 4 | col 5 | col 6 | col 7 | +|--------|-------|-------|-------|-------|-------|-------|-------| +| row 0 | **ALL** | REC | AUDIO | RF | SHOW | VIDEO | STREAM | +| row 1 | HD1 | HD2 | HD3 | HD4 | HD5 | HD6 | TASCAM | +| row 2 | X32 | DANTE | TRACKS | LIGHT PC | RACK (SuperRack) | BIBLE TV | PROP (ProPresenter) | +| row 3 | ATEM | OBS | RESI | NET | INGEST | LED | ARENA | + +Column 0 holds page up / page number / page down, plus **SERVICE** at row 3. + +Row 0 holds the group tiles. Each one shows the first problem in its group, plus "+N" if there are more. RF covers all six Shure units: AD4Q ×2, ULXD4, and PSM1000 ×3. + +## Colours + +| Tile | Meaning | +|---|---| +| Green | OK. The second line shows something useful, like `28h` of record time or `10/10` Dante devices. | +| **Blinking red** | Show-stopping, during service mode. | +| Solid dark red | Acknowledged. It stays red until the problem clears, then goes green. A new failure blinks again. | +| Amber | Warning: low disk, battery at 2 bars, RF interference, a device that keeps dropping out ("WING GONE x3"), a network blip in the last 3 minutes ("BLIP 10:21"). | +| Dim red, not blinking | A fault outside service mode. | +| Grey | Expected off outside service: OBS, a HyperDeck with no disk, Bible TV. | +| **Purple** | The watchdog has stopped. The master shows `WATCHDOG DOWN` and Companion's own error count. | + +## Buttons + +- **Press any tile:** acknowledge its alarms (stops the blinking). Press **ALL** to acknowledge everything. +- **Hold a device tile for 2 seconds:** restart that tile's *Companion connection*. This is the fix for a connection stuck after a network blip. It only restarts Companion's side and never reboots the device. +- **SERVICE:** service mode turns on automatically **Sunday 07:00–12:30** (`service_windows` in `config.json`). Press SERVICE to switch it on or off by hand, for example for a Wednesday event. The manual setting lasts until the next scheduled start or end. + +## What each tile checks + +| Tile | Checks | +|---|---| +| HD1–HD6 | Companion connection; **NO DISK**; record time **LOW** (<30 min, amber) / **FULL** (<5 min); **NOT REC** when another deck is recording and this one isn't | +| | HD2 and HD4 are `"optional": true` in `config.json`: with no disk they sit grey with no alarm, and once a disk is in they are watched like the others. Add or remove `"optional": true` on any deck to change that. | +| TASCAM | connection; the recorder's own Error / Caution | +| X32, ATEM, ARENA, LED, OBS | Companion connection | +| DANTE | 10 key devices present (DiGiCo DMIs, WING, DQ rack, AD4Qs, Studio A X32 / Resi AJA / Ingest, Translation) | +| TRACKS | NA-Tracks Dante device; tracks Mac pings; "NA Tracks" MIDI session still on the network; Ableton OSC connection | +| LIGHT PC | Lighting PC still joined to NA Ingest's network MIDI session; its MIDI session still on the network | +| RACK | SuperRack-LiveBox still joined to the MIDI session; its Dante device | +| BIBLE TV | TV pings; OSC connection | +| PROP | ProPresenter API; NA-ProPresenter-Mac Dante device | +| RESI | decoder (RossTalk); resi-studio-cues; Resi cloud reachable | +| NET | NA Ingest's link; gateway 10.1.0.1 pings; **4+ connections from 3+ groups dropping within 10 s** (what happened Sunday 10/5 at 10:21) | +| INGEST | Resi bridge and lobby captions services running; ProdDash running; free disk | +| RF (group) | Shure connections; battery ≤1 bar red, 2 bars amber; RF interference (only while that transmitter is on); PSM RF muted | + +A problem has to last 3 seconds before it shows, so one bad reading doesn't trigger an alarm. Recovery shows immediately. + +## One-time Companion setup + +1. **Variables → Custom Variables:** add three variables. Leave the start values empty and "persist" off: + - `sp_state` + - `sp_hb` + - `sp_cmd` +2. **Import / Export → Import:** choose `status-page.companionconfig` (it's on NA Ingest's Desktop) and import it into **page 19**. Page 19 is empty today. +3. Make sure the **flash_tick_500** trigger stays enabled. It's the blink clock. + +## Changing things + +This folder is the source of truth. Edit `config.json` here, then copy the folder to NA Ingest: + +```bash +scp status_panel.py coremidi.py config.json install.sh checklist-ingest:status-panel-install/ +``` + +Then run the installer there. It backs up the old config and restarts the watchdog: + +```bash +ssh checklist-ingest 'zsh ~/status-panel-install/install.sh' +``` + +If you add or move tiles, also rebuild the page and import it into page 19 again: + +```bash +python3 make_page.py +``` + +Other commands: + +```bash +python3 status_panel.py --once # poll once and print every tile; writes nothing +python3 -m unittest discover -s tests # offline tests, no Companion or gear needed +zsh install.sh --remove # stop the watchdog +``` + +The log is `~/Library/Logs/status-panel.log` on NA Ingest. Every fault and clear is written there with a timestamp, which is handy for a post-service review. + +## Optional: see what the tracks send to the lights + +The TRACKS tile can show when the tracks rig last sent MIDI ("TRK MIDI 3s"). For that, NA Ingest has to *listen* to the "NA Tracks" session. Use a **separate** session so nothing Companion sends can reach Ableton: + +1. On NA Ingest, open **Audio MIDI Setup → Window → Show MIDI Studio → Network** (the globe icon). +2. Under **My Sessions**, press **+**, rename the new session to `NA Ingest Listen`, and tick it to enable it. +3. With `NA Ingest Listen` selected, choose **NA Tracks** in the **Directory** and press **Connect**. + +The watchdog picks it up within about 10 seconds. Until then that line just stays hidden. + +The **LIGHT PC** tile can do the same in the other direction. If the lighting software can send a heartbeat note back to NA Ingest every few seconds, set `"max_silence_seconds": 10` on the LIGHT PC `midi_rx` check. The tile then goes amber if the lighting PC stops answering during service. + +## Files + +| File | Purpose | +|---|---| +| `status_panel.py` | the watchdog | +| `coremidi.py` | reads network MIDI sessions and listens to MIDI sources (ctypes, no extra installs) | +| `config.json` | tiles, positions, checks, service windows | +| `make_page.py` | builds `status-page.companionconfig` from `config.json` | +| `install.sh` | installs or updates the LaunchAgent on NA Ingest | +| `tests/` | offline tests, including a fake Companion and a MIDI loopback | diff --git a/config.json b/config.json new file mode 100644 index 0000000..ef7ce94 --- /dev/null +++ b/config.json @@ -0,0 +1,99 @@ +{ + "companion": { "host": "127.0.0.1", "port": 8000 }, + "page": 19, + "page_name": "19\\nStatus", + "variables": { "state": "sp_state", "heartbeat": "sp_hb", "command": "sp_cmd" }, + + "service_windows": [ + { "days": ["mon", "tue", "wed", "thu", "fri", "sat", "sun"], "start": "00:00", "end": "23:59" } + ], + + "debounce_seconds": 3, + "flap": { "count": 3, "window_seconds": 600 }, + + "network": { + "interface": "en0", + "gateway": "10.1.0.1", + "burst_connections": 4, + "burst_groups": 3, + "burst_window_seconds": 10, + "remember_seconds": 180 + }, + + "alert_webhook": "", + + "tiles": [ + { "id": "all", "label": "ALL", "pos": [0, 1], "kind": "master" }, + { "id": "rec", "label": "REC", "pos": [0, 2], "members": ["hd1", "hd2", "hd3", "hd4", "hd5", "hd6", "tascam"] }, + { "id": "audio", "label": "AUDIO", "pos": [0, 3], "members": ["x32", "dante", "superrack"] }, + { "id": "rf", "label": "RF", "pos": [0, 4], "members": ["axient1", "axient2", "ulxd", "psm1", "psm2", "psm3"] }, + { "id": "show", "label": "SHOW", "pos": [0, 5], "members": ["propres", "lightpc", "tracks", "bibletv"] }, + { "id": "video", "label": "VIDEO", "pos": [0, 6], "members": ["atem", "led", "arena"] }, + { "id": "stream", "label": "STREAM", "pos": [0, 7], "members": ["obs", "resi"] }, + + { "id": "hd1", "label": "HD1", "pos": [1, 1], "checks": [{ "type": "hyperdeck", "conn": "HyperDeck_1" }] }, + { "id": "hd2", "label": "HD2", "pos": [1, 2], "checks": [{ "type": "hyperdeck", "conn": "HyperDeck_2", "optional": true }] }, + { "id": "hd3", "label": "HD3", "pos": [1, 3], "checks": [{ "type": "hyperdeck", "conn": "HyperDeck_3" }] }, + { "id": "hd4", "label": "HD4", "pos": [1, 4], "checks": [{ "type": "hyperdeck", "conn": "HyperDeck_4", "optional": true }] }, + { "id": "hd5", "label": "HD5", "pos": [1, 5], "checks": [{ "type": "hyperdeck", "conn": "HyperDeck_5" }] }, + { "id": "hd6", "label": "HD6", "pos": [1, 6], "checks": [{ "type": "hyperdeck", "conn": "HyperDeck_6" }] }, + { "id": "tascam", "label": "TASCAM", "pos": [1, 7], "checks": [{ "type": "tascam", "conn": "tascam" }] }, + + { "id": "x32", "label": "X32", "pos": [2, 1], "checks": [{ "type": "conn", "conn": "x32" }] }, + { "id": "dante", "label": "DANTE", "pos": [2, 2], "checks": [ + { "type": "dante", "conn": "audinate-dantecontroller", "devices": { + "FOH-DiGiCo-DMI-1": "DMI-1", "FOH-DiGiCo-DMI-2": "DMI-2", "BOH-WING": "WING", "BOH-DQ-Rack": "DQ RACK", + "BOH-AD4Q-A": "AD4Q-A", "BOH-AD4Q-B": "AD4Q-B", "STUDIO-A-X32": "STA X32", + "STUDIO-A-Resi-AJA-SDI": "RESI AJA", "STUDIO-A-Ingest": "INGEST", "NA-Translation": "TRANSL" } } + ] }, + { "id": "tracks", "label": "TRACKS", "pos": [2, 3], "checks": [ + { "type": "dante", "conn": "audinate-dantecontroller", "devices": { "NA-Tracks": "DANTE" } }, + { "type": "ping", "host": "10.1.3.227", "name": "MAC" }, + { "type": "mdns", "instance": "NA Tracks", "name": "MIDI SESS" }, + { "type": "conn", "conn": "ableton" }, + { "type": "midi_rx", "source": "Network NA Ingest Listen", "name": "TRK MIDI" } + ] }, + { "id": "lightpc", "label": "LIGHT PC", "pos": [2, 4], "checks": [ + { "type": "midi_peer", "session": "NA Ingest", "peer": "Lighting PC" }, + { "type": "mdns", "instance": "Lighting PC", "name": "MIDI SESS" }, + { "type": "midi_rx", "source": "Network NA Ingest", "name": "MIDI IN" } + ] }, + { "id": "superrack", "label": "RACK", "pos": [2, 5], "checks": [ + { "type": "midi_peer", "session": "NA Ingest", "peer": "SuperRack-LiveBox" }, + { "type": "dante", "conn": "audinate-dantecontroller", "devices": { "FOH-SuperRack-LiveBox": "DANTE" } } + ] }, + { "id": "bibletv", "label": "BIBLE TV", "pos": [2, 6], "checks": [ + { "type": "http", "url": "http://10.6.7.13:8080/", "msg": "APP DOWN", "interval_seconds": 5 }, + { "type": "conn", "conn": "BibleTV" } + ] }, + { "id": "propres", "label": "PROP", "pos": [2, 7], "checks": [ + { "type": "conn", "conn": "Propresenter-API" }, + { "type": "dante", "conn": "audinate-dantecontroller", "devices": { "NA-ProPresenter-Mac": "DANTE" } } + ] }, + + { "id": "service", "label": "ALARMS", "pos": [3, 0], "kind": "service" }, + { "id": "atem", "label": "ATEM", "pos": [3, 1], "checks": [{ "type": "conn", "conn": "ATEM" }] }, + { "id": "obs", "label": "OBS", "pos": [3, 2], "checks": [{ "type": "conn", "conn": "OBS-Youtube", "service_only": true }] }, + { "id": "resi", "label": "RESI", "pos": [3, 3], "checks": [ + { "type": "conn", "conn": "RESI_Decoder" }, + { "type": "conn", "conn": "resi-studio-cues" }, + { "type": "http", "url": "https://central.resi.io/", "name": "CLOUD" } + ] }, + { "id": "net", "label": "NET", "pos": [3, 4], "kind": "network" }, + { "id": "ingest", "label": "INGEST", "pos": [3, 5], "checks": [ + { "type": "launchd", "label": "org.waterschurch.resi-propresenter", "name": "BRIDGE" }, + { "type": "launchd", "label": "org.waterschurch.lobby-captions", "name": "CAPTIONS" }, + { "type": "process", "match": "/Applications/ProdDash.app/Contents/MacOS/ProdDash", "name": "PRODDASH" }, + { "type": "disk", "path": "/", "warn_gb": 25, "err_gb": 8 } + ] }, + { "id": "led", "label": "LED", "pos": [3, 6], "checks": [{ "type": "conn", "conn": "VX1000" }, { "type": "conn", "conn": "UHDJr" }] }, + { "id": "arena", "label": "ARENA", "pos": [3, 7], "checks": [{ "type": "conn", "conn": "arena" }] }, + + { "id": "axient1", "label": "AXIENT 1", "checks": [{ "type": "shure", "conn": "Axient_1", "channels": 4 }] }, + { "id": "axient2", "label": "AXIENT 2", "checks": [{ "type": "shure", "conn": "Axient_2", "channels": 4 }] }, + { "id": "ulxd", "label": "ULXD", "checks": [{ "type": "shure", "conn": "ULXD-4", "channels": 1 }] }, + { "id": "psm1", "label": "PSM 1", "checks": [{ "type": "psm", "conn": "PSM_1", "channels": 2 }] }, + { "id": "psm2", "label": "PSM 2", "checks": [{ "type": "psm", "conn": "PSM_2", "channels": 2 }] }, + { "id": "psm3", "label": "PSM 3", "checks": [{ "type": "psm", "conn": "PSM_3", "channels": 2 }] } + ] +} diff --git a/coremidi.py b/coremidi.py new file mode 100644 index 0000000..ce70553 --- /dev/null +++ b/coremidi.py @@ -0,0 +1,262 @@ +"""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) diff --git a/install.sh b/install.sh new file mode 100644 index 0000000..5a22cbd --- /dev/null +++ b/install.sh @@ -0,0 +1,59 @@ +#!/bin/zsh +# Install or update the status panel watchdog on this Mac (NA Ingest) as a LaunchAgent. +# Re-run after changing status_panel.py, coremidi.py or config.json in this folder. +# +# zsh install.sh install / update and (re)start +# zsh install.sh --remove stop it and remove the LaunchAgent (files stay) +set -e +SRC=${0:A:h} +LABEL=org.waterschurch.status-panel +# App Support, not ~/Documents: launchd-run python3 cannot read ~/Documents. +DEST="$HOME/Library/Application Support/status-panel" +PLIST="$HOME/Library/LaunchAgents/$LABEL.plist" +DOMAIN="gui/$(id -u)" + +if [[ "$1" == "--remove" ]]; then + launchctl bootout "$DOMAIN/$LABEL" 2>/dev/null || true + rm -f "$PLIST" + echo "removed $LABEL (files left in $DEST)" + exit 0 +fi + +/usr/bin/python3 -c "import json,sys; json.load(open(sys.argv[1]))" "$SRC/config.json" +mkdir -p "$DEST" +if [[ -f "$DEST/config.json" ]] && ! cmp -s "$SRC/config.json" "$DEST/config.json"; then + cp "$DEST/config.json" "$DEST/config.json.bak-$(date +%Y%m%d-%H%M%S)" +fi +cp "$SRC/status_panel.py" "$SRC/coremidi.py" "$SRC/config.json" "$DEST/" + +cat > "$PLIST" < + + + + Label$LABEL + ProgramArguments + + /usr/bin/python3 + $DEST/status_panel.py + --data-dir + $DEST + + RunAtLoad + KeepAlive + ThrottleInterval10 + StandardErrorPath$HOME/Library/Logs/status-panel.err.log + + +EOF + +launchctl bootout "$DOMAIN/$LABEL" 2>/dev/null || true +# bootout returns before the old job is gone; bootstrapping too early fails with "5: Input/output error". +for i in {1..50}; do launchctl print "$DOMAIN/$LABEL" >/dev/null 2>&1 || break; sleep 0.2; done +for attempt in 1 2 3; do + launchctl bootstrap "$DOMAIN" "$PLIST" && break + [[ $attempt == 3 ]] && { echo "launchctl bootstrap failed" >&2; exit 1; } + sleep 1 +done +echo "installed $LABEL -> $DEST" +echo "log: ~/Library/Logs/status-panel.log" diff --git a/make_page.py b/make_page.py new file mode 100644 index 0000000..d18809c --- /dev/null +++ b/make_page.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python3 +"""Build the Companion 5 page export for the STATUS page from config.json. + +Every positioned tile becomes a layered button whose background colour, text +colour and text are expressions over three custom variables: + + sp_state JSON written by status_panel.py: {tile: {x, b, f, b2, f2}} + sp_hb unix time of status_panel.py's last write + flash_tick the existing 1 Hz blink clock (trigger "flash_tick_500") + +If sp_hb is more than 15 s old every tile turns purple, so a stopped +watchdog can never leave a frozen green page behind. + +Buttons write commands into sp_cmd, which status_panel.py reads: + press ack (the master acks everything) + hold 2 seconds restart (restarts that tile's Companion connections) + + python3 make_page.py -> status-page.companionconfig +""" +import gzip +import hashlib +import json +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) +from status_panel import RESTARTABLE, load_config # noqa: E402 + +EXPORT_VERSION = 12 # Companion 5.0.7's export format +COMPANION_BUILD = "5.0.7+9763-stable-cec2f88e6f" +GRID = {"minColumn": 0, "maxColumn": 8, "minRow": 0, "maxRow": 5} +STALE_SECONDS = 15 +PURPLE = 0x6A1B9A +HOLD_MS = 2000 + +ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-" + + +def stable_id(*parts): + """Companion-style 21 character id, the same on every build.""" + digest = hashlib.sha256("/".join(parts).encode()).digest() + return "".join(ALPHABET[b % 64] for b in digest[:21]) + + +def prop(value, expression=False): + return {"value": value, "isExpression": expression} + + +def layers(tid, tile, names, stale_text): + state, hb = names["state"], names["heartbeat"] + stale = "$(internal:time_unix) - $(custom:%s) > %d" % (hb, STALE_SECONDS) + + def pick(on_key, off_key): + return "jsonpath($(custom:%s), '$.%s.%s')" % (state, tid, on_key), \ + "jsonpath($(custom:%s), '$.%s.%s')" % (state, tid, off_key) + + b_on, b_off = pick("b", "b2") + f_on, f_off = pick("f", "f2") + bg = "%s ? %d : ($(custom:flash_tick) == 1 ? %s : %s)" % (stale, PURPLE, b_on, b_off) + fg = "%s ? %d : ($(custom:flash_tick) == 1 ? %s : %s)" % (stale, 0xFFFFFF, f_on, f_off) + text = "%s ? %s : jsonpath($(custom:%s), '$.%s.x')" % (stale, stale_text, state, tid) + # The watchdog picks a size per key so words never break mid-word. + size = "%s ? 20 : jsonpath($(custom:%s), '$.%s.s')" % (stale, state, tid) + geometry = {"enabled": prop(True), "opacity": prop(100), "x": prop(0), "y": prop(0), + "width": prop(100), "height": prop(100), "rotation": prop(0)} + return [ + {"id": "canvas", "name": "Canvas", "usage": "auto", "type": "canvas", + # "border": no top bar (more room for text), still shows a press. + "decoration": prop("border"), "showStatusIcons": prop("default")}, + dict({"id": "box0", "name": "Background", "usage": "auto", "type": "box"}, **geometry, + color=prop(bg, True), borderWidth=prop(0), borderColor=prop(0), borderPosition=prop("inside")), + dict({"id": "image0", "name": "Image", "usage": "auto", "type": "image"}, **geometry, + base64Image=prop(None), halign=prop("center"), valign=prop("center"), fillMode=prop("fit")), + dict({"id": "text0", "name": "Text", "usage": "auto", "type": "text"}, **geometry, + text=prop(text, True), color=prop(fg, True), halign=prop("center"), valign=prop("center"), + fontsize=prop(size, True), fontsizeAllowShrink=prop(True), font=prop("companion-sans"), + outlineColor=prop(4278190080)), + ] + + +def command_action(tid, role, cmd_var, command): + return { + "type": "action", + "id": stable_id(tid, role), + "definitionId": "custom_variable_set_value", + "connectionId": "internal", + "options": { + "name": prop(cmd_var), + # Plain text, not an expression: Companion fills in $(custom:...) as text, + # so this appends (two quick presses are both seen). In a Companion 5 + # expression '+' is numeric and 'a' + 'b' gives NaN. + "value": prop("$(custom:%s);%s" % (cmd_var, command)), + }, + "children": {}, + } + + +def button(tid, tile, cfg): + names = cfg["variables"] + kind = tile.get("kind") + label = tile["label"] + if kind == "master": + # Template string: '+' would add numbers, not join text (see above). + stale_text = "`${'WATCHDOG\\nDOWN\\n'}${$(internal:instance_errors)} ERR`" + press, hold = "ackall", None + note = "Press: acknowledge every alarm." + elif kind == "service": + stale_text = "'%s\\n?'" % label + press, hold = "service toggle", None + note = "Press: turn service mode on/off until the next scheduled window starts or ends." + else: + stale_text = "'%s\\n?'" % label + press = "ack " + tid + restartable = any(c["type"] in RESTARTABLE and c.get("conn") for c in tile.get("checks", [])) + hold = ("restart " + tid) if restartable else None + note = "Press: acknowledge." + (" Hold 2 s: restart this tile's Companion connection(s)." if hold else "") + action_sets = {"down": [], "up": [command_action(tid, "press", names["command"], press)]} + run_while_held = [] + if hold: + action_sets[str(HOLD_MS)] = [command_action(tid, "hold", names["command"], hold)] + run_while_held = [HOLD_MS] + return { + "type": "button-layered", + "style": {"layers": layers(tid, tile, names, stale_text)}, + "options": {"stepProgression": "auto", "stepExpression": "", "rotaryActions": False, + "canModifyStyleInApis": True, + "notes": "STATUS tile '%s' - drawn by status_panel.py on NA Ingest. %s" % (tid, note)}, + "feedbacks": [], + "steps": {"0": {"action_sets": action_sets, "options": {"runWhileHeld": run_while_held}}}, + "localVariables": [], + } + + +def build(cfg): + controls = {"0": {"0": {"type": "pageup"}}, "1": {"0": {"type": "pagenum"}}, "2": {"0": {"type": "pagedown"}}} + for tile in cfg["tiles"]: + if "pos" not in tile: + continue + r, c = (str(n) for n in tile["pos"]) + if controls.get(r, {}).get(c): + raise ValueError("two buttons at row %s column %s" % (r, c)) + controls.setdefault(r, {})[c] = button(tile["id"], tile, cfg) + return { + "version": EXPORT_VERSION, + "type": "page", + "companionBuild": COMPANION_BUILD, + "page": {"id": stable_id("page", cfg["page_name"]), "name": cfg["page_name"], + "controls": controls, "gridSize": GRID}, + "instances": {}, + "connectionCollections": [], + "oldPageNumber": cfg["page"], + "imageLibrary": [], + "imageLibraryCollections": [], + } + + +def main(): + cfg = load_config(os.path.join(HERE, "config.json")) + out = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "status-page.companionconfig") + data = json.dumps(build(cfg), indent=1).encode() + with gzip.open(out, "wb") as f: # Companion's own exports are gzipped JSON + f.write(data) + print("wrote %s (%d buttons)" % (out, sum(1 for t in cfg["tiles"] if "pos" in t))) + + +if __name__ == "__main__": + main() diff --git a/status-page.companionconfig b/status-page.companionconfig new file mode 100644 index 0000000000000000000000000000000000000000..ca61b4b4b122277144acda2ac9ff2ac71a0af3c2 GIT binary patch literal 4785 zcmYLLcRUpS|F^OV(ea@^bheYSldh8)&gPED%$XTkcSeXrW+L0!dxW#k4C#!J*%^_U zb+S&J@x%A=`+om=|M7l5-p}Xj`5MpH?MPZ$+K*oyDb74RtvtOv?!L6Lv$<#O>ip8m z#lh9Z+SSF@!H#Tm#=&FKq>g%S>y%Yk7#>rqbdOei$#C{wzooS#$Dbc7fV)z!D`Vv; zY4s1cj3_PBff%=_$#(~RvW!l3KPV$>Gd6IuIz>3;Ay0KU&ur0ylP{;$;RsH6K-a;Pt-=Ub7VvGS`XcrDm;*1I*?%u@}R>w`S@14X%vA+kK|`b&yiUTGNJq7j=5w%U>EZol28+&e=hfFNV8Usd2nnhE=%Z#w?%uMWTkI?cub!OjX2Paj?hnCjx^A8fuFw2X z7{5Nys#}=ZioqP$`uWPWTVaGR38cGyXLX?cqi?P$KOV2vXRi&{ z*@HmXuB%(Odh)V4)9!a;c3y6?LjUc~%d`lR50xi!WilH6qaRg&TjroN8FO5xZfR4z zLJC@-5`5Bn7m~73Zl@QwRMJcO_V(l$YzGdtK%%`rK$X< z;P=D9!mYxCM&v0*x2^IjR`I0O_H1kKYvI-lPP|1Ib?Hk*)gdYP$Y6l(7u(&(A;!^O zM|J7Tkda1jVYK!L@%>ipV0hPI`rqYO1Yl>4+1dKAsdpiNu60OtS({tn-C0-d#LW(D zg?jI<-&dOzmR5rn&(xv;;iIVnQ)vcc{SfCvS*1eBZ6}e$@Rnb9+_yVt1uH%|1EQ^t_Ks)&V$Ic^iSJSRtR;!Ly-=N@ z=n7b;kL#>ZqMzYt@d^LT4IwKYd&!6dW9PrpJ0pDWg4-Jc$uoaE?t{2uG-8lT|r zz*=wo4&I)0O6_ttFEON1y6b}&mTM_#2CsGd^9+Sr>HnO8}a z0}O{oW+yJ4t*5ZEgJsl*{g=-zHa7w{-`1e|28ZOhrW z>_l=>dRt~vUV<7@Nn34!r=F*O9jtqWzxJXc`A03xF;a#%yXmb}mV$bD`MOK!{>`q! z-nA+dGo+GWzZ1pU9YGdE#{9Y}<9_7f`5-v;f+pvNsB5CP-*w?Rx;$r}uk}lG0iI)x zH8*FrRiFEMLrq#F6C52%_PVchOQf~@dXp!Ro!1J5;-4)FB}hF^xfCjOk4)zXq8~||{($MNK%l0pQaZ@`DmEqico@CE6K@AmlP}Jy_WwAFxe$o|k zzj53^JkZ*lq58c_HFyg{Hhgyl`XzVhJD(@uVMGZzTuIGy^st}hsAvM9NYAYz^d?qO zjXRI_I_-4@T8DK^fR1CHOCpN^zZVT(oSl8_u)jtz_*c{9I>v_`^y|o^!I1_3katVBxG!C^r|wB;yr6V z!9#j?L+bA-Hx##}Vbd0c=A;-NTuIOQC=X&zd{YoAP^mPWTM*fuWiSNl;P?&jq#{1gv>%jLra6sf&!t5-=G30y~qA zEEd*WB_#5S{EYjb(HZ_0h|!`CJhSXceHUGDYF#+qT-Tn}OG9q4bK${?`+aNKv-MD+1K zdBwpq*3*1RyELcYThU+B9Dv|h&U``W;1Ce36}2H1%ZZCQpT*P&8Vv6R>9-qFvX~KW zC1ff)-T~-GhSR3R2W_-o9O`qm#{E*R>u|3}=!~@GkoO>A1F~ipCFGIrhm#U>HC4O) z;V$pyz7@&(!=_}Z4wExR{Hpa6BGm=P@QwWQQi3Q%ZO~mX2)Y(@4abO3y=WYI{EbCi z1}z6h@ZARpIf97{e?uM#Y!bx?zTV*blR6glB#ofmjrNJ)^zh_OnZqDl#Mlvj7 zBVi3*@gyngbGmf3$3k6?1l@G?Nj+}hUrCH$&^g4%3?Ycpn+^5n2px)Z0<;KP#CarO z3MvX^U0epADiMK;g{Vhus9y-1L4~2j`+tDYAq0aa7Ww=W%3?5sZ~MQD$(F9~y&0oA zSC+V#%>$MRms8WMM){=q5MW=Sja%D(=|t$ zTHf@aNrlZUvM3Q_79p0 z;kSAxwdr2(D;N|Jch~Cv?U`$2nA5)m=TA!esQZ}>w=7I|Ko7;8rSFwz+dyKfrl(FI zfBDbEr~)yLtl36M=1fMV5E;xmr;J<+hV)>A3;EuJt=36x<0o_|sJK1$ij8Uv)C;4o}WE>9gF(|-pcbE*Yw+5i&3^5noI?={d+{h>+kVd<*=0L_D}6;7XRB^ zu|V+PluQq|_uB1F2a1>U4^%)w z`aB{iw(sep&xgU0A~)Y?D-_b&UChDXfQv$p1xHGG9Imr%qg9$U`C%xw?NB~KTyc=j zL4X~zYtFC2KL(lL$c9@~g^B9ZUa|Kct#_(|?~RitI=dU)<2P4dPTWPYxyejqbD6f@ znnheiPfC&O6Dyid)FY)$coIo6C^ouRJ&z~xHq=y1!K&&JaZ-F#qF@i1d0K{mv759^ zA*%9}9hAR<{#@D%@KydRH+;cOAN^1Zfjd_2f0vTG(8JPd$F2MUvLjjEy2rK=78|j= z;Qr%RHgN0jCU-KbX&!%y5nkPf1!InKkDjOf`^9aUPlPowqL(5;MNI@QqZ+(?%)pUd z@ToSeu)~0PbguXh%v<;}+Zk)=4g$5KNicrR)FTODx33`#IapUc|CATW!kS*y#h-|{ z555NwDw}~^G{eWM5Z->h-Ml^z(_Hp@T4ASGR#X>z^%~p63c8r>Je)O7H{wmnCc)B zsPKcR65cm4mDt-2Wi6HrXs~m2%lvtlw&{mim~ZF1c|1~{Ua%^T9D&Nto^Z?0lHqhXuEWL$lLxA#Z4)$n@rJ{<5Y3s);Hfb zu=%8LXfj{G1rrzpOPg!B#efta7fYM*{5caA9iHmwNVx;4dUv{lpIgFsax>|*8@RHz zE>Yw}HuRC-K&Fvn&F7>;@@FkN?;}i|hujO0wP8J|ByP z446nn-r)Nm9t7JL$0wfhfi(DWW_V@+^40Y-( z$yt?6k|CAm8)@3JYl^#Rq}j@mzSdh>y2j&b6aK6yM=d2{M$%^Q3be{vB;Uh7NM_U3 z$x#^8{4n3NeFOy0rFuKTG`QmG^0zR}miwnqZ?|Bs!qR7~KIYHlH}j@xVCdv}Plw)g z1s~(PD_uDv3dSrf?`i|tS%k|b0903lRfH%%2U{vsUQ}k#b*HAM{8ygJXwb#CVkudA zB&8bGTW7}5wpn+Eyd<#?rzpd+)hnfC+4oXGDH$uf&cF0WDw=B3_$tTF#A>Q*8Ta{W z#4;h#YceOprqH@5E>44#-?Wh}Cw!hmS5LpYs42dg`n)Nmpcfn(GBtK@^_$vsVms0j zOu(f{&mjuN6&`345aHc6pKyTo4`FNxmnG&3+?0KL4=>%%x+s~S(X$AUwCtV#Tu13Vy6Qo>j zKZ*-~SPdVRZh65>+q$csD3*AmF@>u%S#Jl_o~0Z-{gE_Xu;jlSB*=aD}& z{z-u*<2=Rr+>$gIUSeGzm}yFF0FYR`b_;_$;?<<4p(mY|2-WI`e3^ zxLzYhQP>S~nB#v{jHD!9{gxpgQTj6r_D?qF{ z0Nc_+(1D*cj)P>}N_?(joBY;w@&4Y%fNHnKQ9u21!&0JUk6~}SqpRIPyU}rHJE>>> zu}hztPW<2TNg02H5Qr!fD+@F9aYW!uO2Z^963bK9_TV?R277MQqY6D#-LDz`nl24Q z-S)xjBEvA1M=>woSadNCFf`Saq;!6odqvq@?INnJV)_A_dQy&Y#GcL#nxT*X2KMBU K;OWs6Wd8@k+*5=A literal 0 HcmV?d00001 diff --git a/status_panel.py b/status_panel.py new file mode 100644 index 0000000..54ea983 --- /dev/null +++ b/status_panel.py @@ -0,0 +1,1151 @@ +#!/usr/bin/env python3 +"""Status panel watchdog for the Companion STATUS page on NA Ingest. + +Every second it reads every connection's status and a handful of module +variables from Companion's HTTP API, adds what Companion cannot see (network +MIDI peers, Bonjour sessions, pings, launchd services, disk, Resi cloud), +decides what is wrong, and writes one JSON custom variable (sp_state) that the +STATUS page buttons render. It also writes a heartbeat (sp_hb) so the page can +tell when this script has stopped, and reads button presses from sp_cmd. + +Stdlib only; runs on the macOS /usr/bin/python3 (3.9). + + status_panel.py run (LaunchAgent) + status_panel.py --once poll everything once and print the tiles + status_panel.py --dry-run run, but never write to Companion +""" +import argparse +import http.client +import json +import logging +import logging.handlers +import os +import re +import shutil +import ssl +import subprocess +import sys +import threading +import time +import urllib.parse +import urllib.request +from collections import deque +from datetime import datetime + +HERE = os.path.dirname(os.path.abspath(__file__)) +log = logging.getLogger("status-panel") + +OK, WARN, ERR, OFF = "ok", "warn", "err", "off" + +# Display levels, least to most urgent. "idle" is a service-only fault while +# service mode is off (OBS PC off midweek); "quiet" is any other fault then. +ORDER = ["off", "ok", "idle", "warn", "quiet", "ack", "err"] +RANK = {name: i for i, name in enumerate(ORDER)} + +# (background, text) for the flash-on phase and the flash-off phase. +COLORS = { + "ok": ((0x007A2E, 0xFFFFFF), (0x007A2E, 0xFFFFFF)), + "warn": ((0xFFA000, 0x000000), (0xFFA000, 0x000000)), + "quiet": ((0x3A0000, 0xFF8080), (0x3A0000, 0xFF8080)), + "ack": ((0x8B0000, 0xFFFFFF), (0x8B0000, 0xFFFFFF)), + "err": ((0xFF0000, 0xFFFFFF), (0x000000, 0xFF3030)), + "off": ((0x262626, 0x8C8C8C), (0x262626, 0x8C8C8C)), + "idle": ((0x262626, 0xD0D0D0), (0x262626, 0xD0D0D0)), + "service": ((0x0B57D0, 0xFFFFFF), (0x0B57D0, 0xFFFFFF)), +} + + +class Result: + """One check's verdict. `named` means msg already says which device it is.""" + __slots__ = ("level", "msg", "info", "service_only", "named") + + def __init__(self, level, msg="", info="", service_only=False, named=False): + self.level = level + self.msg = msg + self.info = info + self.service_only = service_only + self.named = named + + def __repr__(self): + return "Result(%s, %r, info=%r)" % (self.level, self.msg, self.info) + + +# --- Companion HTTP API ---------------------------------------------------- + +class Companion: + def __init__(self, host, port, timeout=3.0): + self.host, self.port, self.timeout = host, port, timeout + self._conn = None + + def _request(self, method, path, body=None, headers=None): + for attempt in (1, 2): + try: + if self._conn is None: + self._conn = http.client.HTTPConnection(self.host, self.port, timeout=self.timeout) + self._conn.request(method, path, body=body, headers=headers or {}) + resp = self._conn.getresponse() + return resp.status, resp.read().decode("utf-8", "replace") + except (OSError, http.client.HTTPException): + if self._conn is not None: + self._conn.close() + self._conn = None + if attempt == 2: + raise + + def connections(self): + status, text = self._request("GET", "/api/connections") + if status != 200: + raise OSError("GET /api/connections -> %d" % status) + return json.loads(text) + + def variable(self, label, name): + path = "/api/variable/%s/%s/value" % (urllib.parse.quote(label, safe=""), urllib.parse.quote(name, safe="")) + status, text = self._request("GET", path) + return text if status == 200 else None + + def get_custom(self, name): + status, text = self._request("GET", "/api/custom-variable/%s/value" % urllib.parse.quote(name, safe="")) + return text if status == 200 else None + + def set_custom(self, name, value): + path = "/api/custom-variable/%s/value?value=%s" % ( + urllib.parse.quote(name, safe=""), urllib.parse.quote(value, safe="")) + status, text = self._request("POST", path) + return status == 200, status + + def restart_connection(self, conn_id): + status, _ = self._request("POST", "/api/connections/%s/restart" % urllib.parse.quote(conn_id, safe="")) + return status == 200 + + +# --- Shared snapshot written by the pollers -------------------------------- + +class Snapshot: + def __init__(self): + self.conns = {} # label -> {"id", "enabled", "category", "message"} + self.conns_time = 0.0 + self.vars = {} # (label, name) -> str or None + self.midi = {} # session name -> {"peers": [...]} + self.midi_time = 0.0 + self.mdns = set() # Bonjour _apple-midi._udp instance names + self.mdns_time = 0.0 + self.ping = {} # host -> consecutive failures (0 = answering) + self.link = None # interface active? + self.launchd = {} # label -> pid or None (loaded, not running) + self.launchd_time = 0.0 + self.procs = {} # match -> running + self.http = {} # url -> reachable + self.disk = {} # path -> free GB + self.listeners = {} # MIDI source name -> coremidi.Listener + + def var(self, label, name): + return self.vars.get((label, name)) + + def conns_fresh(self, now): + return now - self.conns_time < 10 + + +# --- Checks ------------------------------------------------------------------ + +def short_reason(message, default): + m = (message or "").upper() + for pattern, text in (("ETIMEDOUT", "NO ANSWER"), ("TIMED OUT", "NO ANSWER"), ("TIMEOUT", "NO ANSWER"), + ("ECONNREFUSED", "REFUSED"), ("EHOSTUNREACH", "UNREACHABLE"), + ("ENETUNREACH", "UNREACHABLE"), ("EHOSTDOWN", "UNREACHABLE"), + ("AUTH", "LOGIN"), ("LOGIN", "LOGIN"), ("PASSWORD", "LOGIN"), + ("401", "LOGIN"), ("403", "LOGIN")): + if pattern in m: + return text + return default + + +def conn_result(snap, label, now, service_only=False): + if not snap.conns_fresh(now): + return Result(WARN, "NO DATA") + c = snap.conns.get(label) + if c is None: + return Result(ERR, "NOT IN COMPANION", service_only=service_only) + if not c.get("enabled"): + return Result(OFF, info="DISABLED") + cat = c.get("category") + if cat == "good": + return Result(OK) + if cat == "warning": + return Result(WARN, short_reason(c.get("message"), "WARNING"), service_only=service_only) + if cat == "error": + return Result(ERR, short_reason(c.get("message"), "OFFLINE"), service_only=service_only) + return Result(WARN, "STARTING", service_only=service_only) + + +def parse_hms_minutes(text): + m = re.match(r"^\s*(\d+):(\d{1,2}):(\d{1,2})\s*$", text or "") + if not m: + return None + return int(m.group(1)) * 60 + int(m.group(2)) + + +def fmt_minutes(mins): + return "%dh" % (mins // 60) if mins >= 60 else "%dm" % mins + + +def fmt_age(seconds): + s = int(seconds) + if s < 60: + return "%ds" % s + if s < 3600: + return "%dm" % (s // 60) + return "%dh" % (s // 3600) + + +def to_int(text): + try: + return int(str(text).strip()) + except (TypeError, ValueError): + return None + + +def check_conn(chk, snap, ctx): + return [("conn", conn_result(snap, chk["conn"], ctx["now"], chk.get("service_only", False)))] + + +def hyperdeck_recording(snap, label): + return (snap.var(label, "status") or "").strip().lower() == "record" + + +def check_hyperdeck(chk, snap, ctx): + label = chk["conn"] + base = conn_result(snap, label, ctx["now"]) + if base.level != OK: + return [("conn", base)] + recording = hyperdeck_recording(snap, label) + slot = (snap.var(label, "slotId") or "").strip() + mins = parse_hms_minutes(snap.var(label, "recordingTime")) + out = [] + if (slot in ("", "-") or mins is None) and chk.get("optional"): + # An optional deck with no disk is simply not in use today: grey, no alarm. + # Insert a disk and it is watched like the others. + return [("conn", Result(OFF, info="NO DISK"))] + if slot in ("", "-") or mins is None: + out.append(("disk", Result(ERR, "NO DISK", info="NO DISK", service_only=chk.get("disk_service_only", True)))) + info = "NO DISK" + else: + left = fmt_minutes(mins) + if mins < chk.get("full_minutes", 5): + out.append(("disk", Result(ERR, "FULL " + left))) + elif mins < chk.get("low_minutes", 30): + out.append(("disk", Result(WARN, "LOW " + left))) + else: + out.append(("disk", Result(OK))) + info = ("REC " + left) if recording else left + if chk.get("rec_sync", True): + if ctx["any_recording"] and not recording: + out.append(("rec", Result(ERR, "NOT REC"))) + else: + out.append(("rec", Result(OK))) + out.insert(0, ("conn", Result(OK, info=info))) + return out + + +def check_dante(chk, snap, ctx): + label = chk["conn"] + base = conn_result(snap, label, ctx["now"]) + if base.level != OK: + return [("conn", base)] + raw = snap.var(label, "devices") + try: + present = set(json.loads(raw)) if raw else None + except ValueError: + present = None + if present is None: + return [("conn", Result(WARN, "NO DATA"))] + devices = chk["devices"] + if isinstance(devices, list): + devices = {d: d for d in devices} + out, seen = [], 0 + for name, alias in devices.items(): + if name in present: + seen += 1 + out.append((name, Result(OK))) + else: + out.append((name, Result(ERR, alias + " GONE", service_only=chk.get("service_only", False)))) + if len(devices) > 1: + out.insert(0, ("conn", Result(OK, info="%d/%d" % (seen, len(devices))))) + return out + + +def channel_name(snap, label, ch): + name = (snap.var(label, "ch_%d_name" % ch) or "").strip().upper() + return name[:9] if name and name != "UNKNOWN" else "CH%d" % ch + + +def check_shure(chk, snap, ctx): + label = chk["conn"] + base = conn_result(snap, label, ctx["now"]) + if base.level != OK: + return [("conn", base)] + out = [] + for ch in range(1, chk.get("channels", 1) + 1): + name = channel_name(snap, label, ch) + bars = to_int(snap.var(label, "ch_%d_battery_bars" % ch)) + res = Result(OK) + if bars is not None and 0 <= bars <= 5: + if bars <= chk.get("battery_err_bars", 1): + res = Result(ERR, name + " BATT", named=True) + elif bars <= chk.get("battery_warn_bars", 2): + res = Result(WARN, name + " BATT", named=True) + out.append(("ch%d_batt" % ch, res)) + # Only while that channel's transmitter is on (battery reads): an idle + # receiver on a noisy frequency flags interference on and off all week. + intf = (snap.var(label, "ch_%d_interference_status" % ch) or "").strip().upper() + if bars is not None and 0 <= bars <= 5 and intf and intf not in ("NONE", "UNKNOWN"): + out.append(("ch%d_rf" % ch, Result(WARN, name + " INTERF", named=True))) + else: + out.append(("ch%d_rf" % ch, Result(OK))) + return out + + +def check_psm(chk, snap, ctx): + label = chk["conn"] + base = conn_result(snap, label, ctx["now"]) + if base.level != OK: + return [("conn", base)] + out = [] + level = ERR if chk.get("muted_is_error") else WARN + for ch in range(1, chk.get("channels", 1) + 1): + muted = (snap.var(label, "ch_%d_rf_mute" % ch) or "").strip().upper() == "MUTED" + res = Result(level, channel_name(snap, label, ch) + " RF MUTE", named=True) if muted else Result(OK) + out.append(("ch%d_mute" % ch, res)) + return out + + +def check_tascam(chk, snap, ctx): + label = chk["conn"] + base = conn_result(snap, label, ctx["now"]) + if base.level != OK: + return [("conn", base)] + out = [("conn", Result(OK))] + error = (snap.var(label, "error") or "").strip() + caution = (snap.var(label, "caution") or "").strip() + if error and error.lower() != "no error": + out.append(("error", Result(ERR, error.upper()[:14]))) + else: + out.append(("error", Result(OK))) + if caution and caution.lower() != "no caution": + out.append(("caution", Result(WARN, caution.upper()[:14]))) + else: + out.append(("caution", Result(OK))) + return out + + +def check_ping(chk, snap, ctx): + fails = snap.ping.get(chk["host"]) + if fails is not None and fails >= chk.get("failures", 2): + msg = chk.get("msg") or chk.get("name", chk["host"]) + " OFFLINE" + return [("ping", Result(ERR, msg, service_only=chk.get("service_only", False)))] + return [("ping", Result(OK))] + + +def check_launchd(chk, snap, ctx): + if not snap.launchd_time: + return [("svc", Result(OK))] + name = chk.get("name", chk["label"]) + if chk["label"] not in snap.launchd: + return [("svc", Result(ERR, name + " NOT LOADED"))] + if snap.launchd[chk["label"]] is None: + return [("svc", Result(ERR, name + " STOPPED"))] + return [("svc", Result(OK))] + + +def check_process(chk, snap, ctx): + running = snap.procs.get(chk["match"]) + if running is False: + return [("proc", Result(ERR, chk.get("name", "PROCESS") + " STOPPED"))] + return [("proc", Result(OK))] + + +def check_http(chk, snap, ctx): + if snap.http.get(chk["url"]) is False: + return [("http", Result(ERR, chk.get("msg") or chk.get("name", "WEB") + " UNREACH"))] + return [("http", Result(OK))] + + +def check_disk(chk, snap, ctx): + free = snap.disk.get(chk["path"]) + if free is None: + return [("disk", Result(OK))] + text = "DISK %dGB" % free + if free < chk.get("err_gb", 5): + return [("disk", Result(ERR, text))] + if free < chk.get("warn_gb", 20): + return [("disk", Result(WARN, text))] + return [("disk", Result(OK, info="%dGB free" % free))] + + +def check_midi_peer(chk, snap, ctx): + if not snap.midi_time: + return [("peer", Result(OK))] + sess = snap.midi.get(chk["session"]) + if sess is None: + return [("peer", Result(ERR, "NO SESSION"))] + for p in sess["peers"]: + if p["name"] == chk["peer"]: + # Measured 2026-10-07: a joining peer reads "connected" for ~6 s until + # clock sync, then "running". A peer that dies is removed from the list: + # at once if it says goodbye, after ~60 s if it crashes or its cable is pulled. + if p["state"] in ("running", "connected"): + return [("peer", Result(OK, info="MIDI OK"))] + return [("peer", Result(ERR, "MIDI " + (p["state"] or "?").upper()[:8]))] + return [("peer", Result(ERR, "MIDI LOST"))] + + +def check_mdns(chk, snap, ctx): + if snap.mdns_time and chk["instance"] not in snap.mdns: + return [("mdns", Result(ERR, chk.get("name", chk["instance"]) + " GONE"))] + return [("mdns", Result(OK))] + + +def check_midi_rx(chk, snap, ctx): + listener = snap.listeners.get(chk["source"]) + name = chk.get("name", "MIDI") + if listener is None or not listener.connected: + return [("rx", Result(OFF))] + if listener.last_time is None: + return [("rx", Result(OK, info=name + " --"))] + age = ctx["now"] - listener.last_time + limit = chk.get("max_silence_seconds") + if limit and ctx["service"] and age > limit: + return [("rx", Result(WARN, name + " SILENT"))] + return [("rx", Result(OK, info="%s %s" % (name, fmt_age(age))))] + + +CHECKS = { + "conn": check_conn, "hyperdeck": check_hyperdeck, "dante": check_dante, "shure": check_shure, + "psm": check_psm, "tascam": check_tascam, "ping": check_ping, "launchd": check_launchd, + "process": check_process, "http": check_http, "disk": check_disk, "midi_peer": check_midi_peer, + "mdns": check_mdns, "midi_rx": check_midi_rx, +} + +# Checks whose connection a long press may restart (Dante is shared, so not that one). +RESTARTABLE = ("conn", "hyperdeck", "shure", "psm", "tascam") + + +def needed_variables(cfg): + """(fast, slow) lists of (connection label, variable name) to poll.""" + fast, slow = [], [] + for tile in cfg["tiles"]: + for chk in tile.get("checks", []): + t, label = chk["type"], chk.get("conn") + if t == "hyperdeck": + fast += [(label, "status"), (label, "slotId"), (label, "recordingTime")] + elif t == "dante": + fast.append((label, "devices")) + elif t == "tascam": + fast += [(label, "error"), (label, "caution")] + elif t == "shure": + for ch in range(1, chk.get("channels", 1) + 1): + slow += [(label, "ch_%d_name" % ch), (label, "ch_%d_battery_bars" % ch), + (label, "ch_%d_interference_status" % ch)] + elif t == "psm": + for ch in range(1, chk.get("channels", 1) + 1): + slow += [(label, "ch_%d_name" % ch), (label, "ch_%d_rf_mute" % ch)] + dedupe = lambda xs: list(dict.fromkeys(xs)) + return dedupe(fast), dedupe(slow) + + +# --- Debounce, flapping and acknowledgement -------------------------------- + +class Tracker: + def __init__(self): + self.bad_since = None + self.confirmed = False + self.acked = False + self.result = Result(OK) + self.since = None # when the current confirmed fault started + self.starts = deque() # confirmed fault start times (for flapping) + self.last_fault_msg = "" + + def update(self, res, now, debounce, flap_window): + """Returns "fault", "clear" or None.""" + event = None + bad = res.level in (WARN, ERR) + if bad: + if self.bad_since is None: + self.bad_since = now + if not self.confirmed and now - self.bad_since >= debounce: + self.confirmed, self.since = True, now + self.starts.append(now) + event = "fault" + self.result = res + self.last_fault_msg = res.msg + else: + if self.confirmed: + event = "clear" + self.bad_since, self.confirmed, self.acked, self.since = None, False, False, None + self.result = res + while self.starts and now - self.starts[0] > flap_window: + self.starts.popleft() + return event + + def effective(self, flap_count): + if self.confirmed: + return self.result + if self.bad_since is not None: # bad but still inside the debounce + return Result(OK) + if len(self.starts) >= flap_count: + return Result(WARN, "%s x%d" % (self.last_fault_msg, len(self.starts)), named=True) + return self.result + + +# --- Service mode ------------------------------------------------------------ + +DAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"] + + +class ServiceMode: + """On inside a scheduled window; a button press overrides until the next window edge.""" + + def __init__(self, windows, path): + self.windows, self.path = windows, path + self.override, self.last_sched = None, None + if not path: + return + try: + with open(path) as f: + saved = json.load(f) + self.override, self.last_sched = saved.get("override"), saved.get("last_sched") + except (OSError, ValueError): + pass + + def scheduled(self, now): + dt = datetime.fromtimestamp(now) + hm = dt.strftime("%H:%M") + day = DAYS[dt.weekday()] + return any(day in w["days"] and w["start"] <= hm < w["end"] for w in self.windows) + + def on(self, now): + sched = self.scheduled(now) + if sched != self.last_sched: + self.last_sched, self.override = sched, None + self._save() + return self.override if self.override is not None else sched + + def set(self, value, now): + sched = self.scheduled(now) + self.last_sched = sched + self.override = None if value == sched else value + self._save() + + def _save(self): + if not self.path: + return + try: + tmp = self.path + ".tmp" + with open(tmp, "w") as f: + json.dump({"override": self.override, "last_sched": self.last_sched}, f) + os.replace(tmp, self.path) + except OSError as e: + log.warning("could not save service mode: %s", e) + + +# --- Engine ------------------------------------------------------------------ + +class Engine: + def __init__(self, cfg, state_path=None): + self.cfg = cfg + self.tiles = {t["id"]: t for t in cfg["tiles"]} + self.debounce = cfg.get("debounce_seconds", 3) + self.flap_count = cfg.get("flap", {}).get("count", 3) + self.flap_window = cfg.get("flap", {}).get("window_seconds", 600) + self.net = cfg.get("network", {}) + self.service = ServiceMode(cfg.get("service_windows", []), state_path) + self.trackers = {} + self.group_of = {} + for t in cfg["tiles"]: + for m in t.get("members", []): + self.group_of[m] = t["id"] + self.detail_ids = [t["id"] for t in cfg["tiles"] if "checks" in t or t.get("kind") == "network"] + self.prev_conn_bad = {} + self.conn_drops = deque() # (time, conn label, group) + self.burst = None # {"start", "conns", "cleared"} + self.events = [] + + # raw results for every check key + def evaluate(self, snap, now, service_on): + ctx = {"now": now, "service": service_on, "any_recording": False} + for t in self.cfg["tiles"]: + for chk in t.get("checks", []): + if chk["type"] == "hyperdeck" and chk.get("rec_sync", True): + if conn_result(snap, chk["conn"], now).level == OK and hyperdeck_recording(snap, chk["conn"]): + ctx["any_recording"] = True + raw = {} + for t in self.cfg["tiles"]: + for i, chk in enumerate(t.get("checks", [])): + try: + results = CHECKS[chk["type"]](chk, snap, ctx) + except Exception as e: # a bad check must not take the panel down + log.exception("check %s/%s failed", t["id"], chk["type"]) + results = [("error", Result(WARN, "CHECK BUG"))] + for sub, res in results: + raw["%s:%d:%s" % (t["id"], i, sub)] = (t["id"], res, chk) + for sub, res in self.network_results(snap, now): + raw["net:%s" % sub] = ("net", res, {"type": "network"}) + return raw + + def network_results(self, snap, now): + out = [] + out.append(("link", Result(ERR, "LINK DOWN") if snap.link is False else Result(OK))) + gw = self.net.get("gateway") + fails = snap.ping.get(gw) if gw else None + out.append(("gateway", Result(ERR, "NO GATEWAY") if fails is not None and fails >= 2 else Result(OK))) + # Burst: several unrelated connections dropping together means the network, not the gear. + bad_now = set() + if snap.conns_fresh(now): + for label, c in snap.conns.items(): + bad = bool(c.get("enabled")) and c.get("category") == "error" + if bad: + bad_now.add(label) + if bad and not self.prev_conn_bad.get(label): + self.conn_drops.append((now, label, self.group_for_conn(label))) + self.prev_conn_bad[label] = bad + window = self.net.get("burst_window_seconds", 10) + while self.conn_drops and now - self.conn_drops[0][0] > window: + self.conn_drops.popleft() + conns = {c for _, c, _ in self.conn_drops} + groups = {g for _, _, g in self.conn_drops} + if (len(conns) >= self.net.get("burst_connections", 4) and len(groups) >= self.net.get("burst_groups", 3) + and (self.burst is None or self.burst["cleared"] is not None)): + self.burst = {"start": now, "conns": set(conns), "cleared": None} + self.events.append(("NETWORK", "burst: %s" % ", ".join(sorted(conns)))) + if self.burst: + if self.burst["cleared"] is None: + self.burst["conns"] |= conns + still = self.burst["conns"] & bad_now + if still: + out.append(("burst", Result(ERR, "%d DROPPED" % len(self.burst["conns"])))) + else: + self.burst["cleared"] = now + if self.burst["cleared"] is not None: + if now - self.burst["cleared"] < self.net.get("remember_seconds", 180): + stamp = datetime.fromtimestamp(self.burst["start"]).strftime("%H:%M") + out.append(("burst", Result(WARN, "BLIP " + stamp))) + else: + self.burst = None + if not any(sub == "burst" for sub, _ in out): + out.append(("burst", Result(OK))) + return out + + def group_for_conn(self, label): + for t in self.cfg["tiles"]: + for chk in t.get("checks", []): + if chk.get("conn") == label: + return self.group_of.get(t["id"], t["id"]) + return label + + def step(self, snap, now, debounce=None): + service_on = self.service.on(now) + raw = self.evaluate(snap, now, service_on) + deb = self.debounce if debounce is None else debounce + for key, (tile_id, res, chk) in raw.items(): + tr = self.trackers.setdefault(key, Tracker()) + event = tr.update(res, now, 0 if chk["type"] == "network" else chk.get("debounce_seconds", deb), + self.flap_window) + if event == "fault": + self.events.append(("FAULT", "%s %s" % (self.tiles[tile_id]["label"], res.msg))) + elif event == "clear": + self.events.append(("CLEAR", "%s %s" % (self.tiles[tile_id]["label"], tr.last_fault_msg))) + self.items = {} + for key, (tile_id, res, chk) in raw.items(): + tr = self.trackers[key] + eff = tr.effective(self.flap_count) + self.items.setdefault(tile_id, []).append( + (display_level(eff, tr.acked, service_on), eff, tr.since or now)) + return render(self, service_on) + + def keys_for(self, tile_id): + tile = self.tiles.get(tile_id, {}) + if tile.get("kind") == "master": + ids = self.detail_ids + elif "members" in tile: + ids = tile["members"] + else: + ids = [tile_id] + prefixes = tuple(i + ":" for i in ids) + return [k for k in self.trackers if k.startswith(prefixes)] + + def ack(self, tile_id): + n = 0 + for key in self.keys_for(tile_id): + tr = self.trackers[key] + if tr.confirmed and not tr.acked: + tr.acked = True + n += 1 + return n + + def restart_targets(self, tile_id): + tile = self.tiles.get(tile_id, {}) + return [c["conn"] for c in tile.get("checks", []) if c["type"] in RESTARTABLE and c.get("conn")] + + +def display_level(res, acked, service_on): + if res.level == OFF: + return "off" + if res.level == OK: + return "ok" + if res.level == WARN: + return "warn" + if res.service_only and not service_on: + return "idle" + if not service_on: + return "quiet" + return "ack" if acked else "err" + + +def worst(items): + """items: [(level, Result, since)] -> (level, ordered faults).""" + if not items: + return "off", [] + level = max((lv for lv, _, _ in items), key=lambda lv: RANK[lv]) + if all(lv == "off" for lv, _, _ in items): + level = "off" + faults = [it for it in items if it[0] in ("idle", "warn", "quiet", "ack", "err")] + faults.sort(key=lambda it: (-RANK[it[0]], it[2])) + return level, faults + + +def fault_text(tile, res): + return res.msg if res.named else "%s %s" % (tile["label"], res.msg) + + +# Characters x text size (in % of key height) that fit across one key, measured +# on Companion's web buttons: "VIDEO" fits at 29.4%, "TASCAM" does not. +LINE_CAPACITY = 145 +LINE_HEIGHT = 1.15 + + +def fit_size(text, largest=30, smallest=12): + """Largest text size at which no word breaks in the middle and all lines fit.""" + for size in range(largest, smallest - 1, -1): + cap = int(LINE_CAPACITY / size) + lines, fits = 0, True + for para in text.split("\n"): + width = 0 + lines += 1 + for word in para.split(): + if len(word) > cap: + fits = False + break + if width and width + 1 + len(word) > cap: + lines += 1 + width = len(word) + else: + width += (1 if width else 0) + len(word) + if not fits: + break + if fits and lines * size * LINE_HEIGHT <= 95: + return size + return smallest + + +def colored(text, level): + (b, f), (b2, f2) = COLORS[level] + return {"x": text, "b": b, "f": f, "b2": b2, "f2": f2, "s": fit_size(text)} + + +def render(engine, service_on): + out = {} + for tile in engine.cfg["tiles"]: + tid, kind = tile["id"], tile.get("kind") + if kind == "service": + manual = engine.service.override is not None + text = tile["label"] + "\n" + ("ON" if service_on else "OFF") + ("\nmanual" if manual else "") + out[tid] = colored(text, "service" if service_on else "off") + continue + if kind == "master": + # Idle faults are expected outside service; they do not stop "ALL OK". + items = [("ok" if lv == "idle" else lv, res, since, engine.tiles[t]) for t in engine.detail_ids + for lv, res, since in engine.items.get(t, [])] + level, faults = worst([(lv, res, since) for lv, res, since, _ in items]) + if not faults: + text, level = "ALL OK", "ok" + else: + first = next(fault_text(t, r) for lv, r, s, t in items if (lv, r, s) == faults[0]) + n_bad = sum(1 for f in faults if f[0] != "warn") + text = ("%d DOWN" % n_bad if n_bad else "%d WARN" % len(faults)) + "\n" + first + out[tid] = colored(text, level) + continue + if "members" in tile: + items = [(lv, res, since, engine.tiles[m]) for m in tile["members"] + for lv, res, since in engine.items.get(m, [])] + level, faults = worst([(lv, res, since) for lv, res, since, _ in items]) + if faults: + first = next(fault_text(t, r) for lv, r, s, t in items if (lv, r, s) == faults[0]) + text = tile["label"] + "\n" + first + ("\n+%d" % (len(faults) - 1) if len(faults) > 1 else "") + else: + text = tile["label"] + "\n" + ("OK" if level == "ok" else "OFF") + out[tid] = colored(text, level) + continue + items = engine.items.get(tid, []) + level, faults = worst(items) + if faults: + text = tile["label"] + "\n" + faults[0][1].msg + ("\n+%d" % (len(faults) - 1) if len(faults) > 1 else "") + else: + infos = [res.info for lv, res, _ in items if res.info] + if level == "off": + text = tile["label"] + "\n" + (infos[0] if infos else "OFF") + else: + text = tile["label"] + "\n" + (infos[0] if infos else "OK") + out[tid] = colored(text, level) + return out + + +# --- Pollers ----------------------------------------------------------------- + +def poll_connections(comp, snap): + data = comp.connections() + conns = {} + for c in data: + st = c.get("status") or {} + conns[c["label"]] = {"id": c["id"], "enabled": c.get("enabled", False), + "category": st.get("category"), "message": st.get("message")} + snap.conns, snap.conns_time = conns, time.time() + + +def poll_variables(comp, snap, pairs): + values = dict(snap.vars) + for label, name in pairs: + values[(label, name)] = comp.variable(label, name) + snap.vars = values + + +def ping_once(host): + try: + return subprocess.run(["/sbin/ping", "-c", "1", "-t", "1", "-q", host], stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, timeout=3).returncode == 0 + except subprocess.TimeoutExpired: + return False + + +def poll_pings(snap, hosts, interface): + results = {} + + def one(h): + results[h] = ping_once(h) + threads = [threading.Thread(target=one, args=(h,)) for h in hosts] + for t in threads: + t.start() + for t in threads: + t.join() + pings = dict(snap.ping) + for h, ok in results.items(): + pings[h] = 0 if ok else pings.get(h, 0) + 1 + snap.ping = pings + if interface: + try: + out = subprocess.run(["/sbin/ifconfig", interface], capture_output=True, text=True, timeout=3).stdout + snap.link = "status: active" in out + except (OSError, subprocess.TimeoutExpired): + snap.link = None + + +def poll_system(snap, cfg): + checks = [c for t in cfg["tiles"] for c in t.get("checks", [])] + if any(c["type"] == "launchd" for c in checks): + out = subprocess.run(["/bin/launchctl", "list"], capture_output=True, text=True, timeout=5).stdout + agents = {} + for line in out.splitlines()[1:]: + parts = line.split("\t") + if len(parts) == 3: + agents[parts[2]] = int(parts[0]) if parts[0].isdigit() else None + snap.launchd, snap.launchd_time = agents, time.time() + procs = {} + for c in checks: + if c["type"] == "process": + r = subprocess.run(["/usr/bin/pgrep", "-f", c["match"]], stdout=subprocess.DEVNULL, timeout=5) + procs[c["match"]] = r.returncode == 0 + elif c["type"] == "disk": + usage = shutil.disk_usage(c["path"]) + snap.disk = dict(snap.disk, **{c["path"]: int(usage.free / 1e9)}) + snap.procs = procs + + +def poll_mdns(snap, seconds=3): + proc = subprocess.Popen(["/usr/bin/dns-sd", "-B", "_apple-midi._udp", "local."], + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True) + try: + out, _ = proc.communicate(timeout=seconds) + except subprocess.TimeoutExpired: + proc.kill() + out, _ = proc.communicate() + names = set() + for line in out.splitlines(): + m = re.match(r"^\S+\s+(Add|Rmv)\s+\d+\s+\d+\s+\S+\s+_apple-midi\._udp\.\s+(.+?)\s*$", line) + if m: + (names.add if m.group(1) == "Add" else names.discard)(m.group(2)) + snap.mdns, snap.mdns_time = names, time.time() + + +_UNVERIFIED = ssl._create_unverified_context() # reachability only; nothing is sent + + +_http_last = {} + + +def poll_http(snap, checks): + """checks: [(url, interval seconds)]; each URL is fetched when its interval is due.""" + result = dict(snap.http) + now = time.time() + due = [u for u, every in checks if now - _http_last.get(u, 0) >= every] + for url in due: + _http_last[url] = now + try: + urllib.request.urlopen(urllib.request.Request(url, method="HEAD"), timeout=6, context=_UNVERIFIED) + result[url] = True + except urllib.error.HTTPError: + result[url] = True # it answered; that is all we ask + except (OSError, ValueError): + result[url] = False + snap.http = result + + +def poll_midi(snap, cfg): + import coremidi + snap.midi, snap.midi_time = coremidi.sessions(), time.time() + for t in cfg["tiles"]: + for c in t.get("checks", []): + if c["type"] == "midi_rx": + lis = snap.listeners.get(c["source"]) + if lis is None: + lis = snap.listeners[c["source"]] = coremidi.Listener(c["source"]) + lis.poll() + + +class Periodic(threading.Thread): + def __init__(self, name, interval, fn, stop): + super().__init__(name=name, daemon=True) + self.interval, self.fn, self.stop = interval, fn, stop + self.failing = False + + def run(self): + while not self.stop.is_set(): + start = time.time() + try: + self.fn() + if self.failing: + log.info("%s recovered", self.name) + self.failing = False + except Exception as e: + if not self.failing: + log.warning("%s failing: %s", self.name, e) + self.failing = True + self.stop.wait(max(0.05, self.interval - (time.time() - start))) + + +# --- Main -------------------------------------------------------------------- + +def load_config(path): + with open(path) as f: + cfg = json.load(f) + ids = set() + for t in cfg["tiles"]: + if t["id"] in ids: + raise ValueError("duplicate tile id %s" % t["id"]) + ids.add(t["id"]) + for c in t.get("checks", []): + if c["type"] not in CHECKS: + raise ValueError("tile %s: unknown check type %s" % (t["id"], c["type"])) + for t in cfg["tiles"]: + for m in t.get("members", []): + if m not in ids: + raise ValueError("tile %s: unknown member %s" % (t["id"], m)) + return cfg + + +def ping_hosts(cfg): + hosts = [c["host"] for t in cfg["tiles"] for c in t.get("checks", []) if c["type"] == "ping"] + gw = cfg.get("network", {}).get("gateway") + return list(dict.fromkeys(hosts + ([gw] if gw else []))) + + +def http_checks(cfg): + """[(url, interval)]: the Resi cloud every 30 s, a LAN app as often as its check asks.""" + found = {} + for t in cfg["tiles"]: + for c in t.get("checks", []): + if c["type"] == "http": + found[c["url"]] = min(found.get(c["url"], 999), c.get("interval_seconds", 30)) + return list(found.items()) + + +def make_companion(cfg): + return Companion(cfg["companion"].get("host", "127.0.0.1"), cfg["companion"].get("port", 8000)) + + +def start_pollers(cfg, snap, stop): + fast, slow = needed_variables(cfg) + c1, c2, c3 = make_companion(cfg), make_companion(cfg), make_companion(cfg) + iface = cfg.get("network", {}).get("interface") + threads = [ + Periodic("connections", 1.0, lambda: poll_connections(c1, snap), stop), + Periodic("variables", 1.0, lambda: poll_variables(c2, snap, fast), stop), + Periodic("slow-variables", 3.0, lambda: poll_variables(c3, snap, slow), stop), + Periodic("pings", 3.0, lambda: poll_pings(snap, ping_hosts(cfg), iface), stop), + Periodic("midi", 2.0, lambda: poll_midi(snap, cfg), stop), + Periodic("system", 10.0, lambda: poll_system(snap, cfg), stop), + Periodic("bonjour", 15.0, lambda: poll_mdns(snap), stop), + Periodic("http", 5.0, lambda: poll_http(snap, http_checks(cfg)), stop), + ] + for t in threads: + t.start() + return threads + + +def post_webhook(url, text): + if not url: + return + try: + req = urllib.request.Request(url, data=json.dumps({"text": text}).encode(), + headers={"Content-Type": "application/json"}) + urllib.request.urlopen(req, timeout=8) + except (OSError, ValueError) as e: + log.warning("webhook failed: %s", e) + + +def handle_commands(text, engine, comp, snap, now): + for part in text.split(";"): + words = part.strip().split() + if not words: + continue + verb, arg = words[0].lower(), (words[1] if len(words) > 1 else "") + if verb == "ack": + log.info("ACK %s (%d faults)", arg, engine.ack(arg)) + elif verb == "ackall": + log.info("ACK all (%d faults)", engine.ack("all")) + elif verb == "service": + target = {"on": True, "off": False}.get(arg, not engine.service.on(now)) + engine.service.set(target, now) + log.info("SERVICE %s (button)", "ON" if target else "OFF") + elif verb == "restart": + for label in engine.restart_targets(arg): + c = snap.conns.get(label) + if c and c.get("enabled"): + ok = comp.restart_connection(c["id"]) + log.info("RESTART %s -> %s", label, "ok" if ok else "failed") + else: + log.warning("unknown command %r", part) + + +def run(cfg, data_dir, dry_run=False, stop=None): + stop = stop or threading.Event() + names = cfg["variables"] + comp = make_companion(cfg) + snap = Snapshot() + engine = Engine(cfg, os.path.join(data_dir, "service-mode.json")) + threads = start_pollers(cfg, snap, stop) + stop.wait(2.5) # let the first polls land before judging anything + last_json, last_write, last_eval = None, 0.0, 0.0 + down_since, alerted, warned_missing = None, False, set() + log.info("status panel running (%s)", "dry run" if dry_run else "live") + while not stop.is_set(): + now = time.time() + try: + if not dry_run: + cmd = comp.get_custom(names["command"]) + if cmd and cmd.strip(): + comp.set_custom(names["command"], "") + handle_commands(cmd, engine, comp, snap, now) + if now - last_eval >= 1.0: + last_eval = now + tiles = engine.step(snap, now) + for kind, text in engine.events: + log.info("%s %s", kind, text) + engine.events = [] + js = json.dumps(tiles, separators=(",", ":"), sort_keys=True) + if dry_run: + if js != last_json: + log.info("tiles: %s", " | ".join(tiles[t]["x"].replace("\n", " ") for t in sorted(tiles))) + last_json = js + else: + if js != last_json or now - last_write > 10: + ok, status = comp.set_custom(names["state"], js) + if ok: + last_json, last_write = js, now + elif names["state"] not in warned_missing: + log.warning("cannot set custom:%s (HTTP %s) - create it in Companion", names["state"], status) + warned_missing.add(names["state"]) + ok, status = comp.set_custom(names["heartbeat"], str(int(now))) + if not ok and names["heartbeat"] not in warned_missing: + log.warning("cannot set custom:%s (HTTP %s) - create it in Companion", names["heartbeat"], status) + warned_missing.add(names["heartbeat"]) + if down_since is not None: + log.info("Companion reachable again") + if alerted: + post_webhook(cfg.get("alert_webhook"), "NA Ingest: Companion is answering again.") + down_since, alerted = None, False + except (OSError, http.client.HTTPException) as e: + if down_since is None: + down_since = now + log.warning("Companion unreachable: %s", e) + elif not alerted and now - down_since > 30: + post_webhook(cfg.get("alert_webhook"), "NA Ingest: Companion has not answered for 30 seconds.") + alerted = True + stop.wait(0.25) + for t in threads: + t.join(timeout=10) + + +def run_once(cfg): + """Poll everything synchronously, judge with no debounce, print the tiles.""" + snap = Snapshot() + comp = make_companion(cfg) + fast, slow = needed_variables(cfg) + poll_connections(comp, snap) + poll_variables(comp, snap, fast + slow) + poll_pings(snap, ping_hosts(cfg), cfg.get("network", {}).get("interface")) + poll_pings(snap, ping_hosts(cfg), None) # pings need two misses to count + poll_system(snap, cfg) + poll_mdns(snap) + poll_http(snap, http_checks(cfg)) + try: + poll_midi(snap, cfg) + except Exception as e: + print("midi: %s" % e) + engine = Engine(cfg, None) + tiles = engine.step(snap, time.time(), debounce=0) + rows = {} + for t in cfg["tiles"]: + if "pos" in t: + rows.setdefault(t["pos"][0], {})[t["pos"][1]] = t["id"] + for r in sorted(rows): + for c in sorted(rows[r]): + tid = rows[r][c] + tile = tiles[tid] + lv = next((k for k, ((b, f), _) in COLORS.items() if b == tile["b"] and f == tile["f"]), "?") + print("r%dc%d %-8s %-6s %s" % (r, c, tid, lv, tile["x"].replace("\n", " | "))) + print("hidden:", ", ".join("%s=%s" % (t["id"], tiles[t["id"]]["x"].replace("\n", " | ")) + for t in cfg["tiles"] if "pos" not in t)) + return tiles + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--config", default=os.path.join(HERE, "config.json")) + ap.add_argument("--data-dir", default=HERE) + ap.add_argument("--once", action="store_true", help="poll once, print the tiles, write nothing") + ap.add_argument("--dry-run", action="store_true", help="run but never write to Companion") + ap.add_argument("--log", default=os.path.expanduser("~/Library/Logs/status-panel.log")) + args = ap.parse_args(argv) + cfg = load_config(args.config) + if args.once: + run_once(cfg) + return + handler = logging.handlers.RotatingFileHandler(args.log, maxBytes=1_000_000, backupCount=3) + handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) + log.addHandler(handler) + if sys.stderr.isatty(): + log.addHandler(logging.StreamHandler()) + log.setLevel(logging.INFO) + run(cfg, args.data_dir, dry_run=args.dry_run) + + +if __name__ == "__main__": + main() diff --git a/tests/test_status_panel.py b/tests/test_status_panel.py new file mode 100644 index 0000000..8fef413 --- /dev/null +++ b/tests/test_status_panel.py @@ -0,0 +1,617 @@ +"""Offline tests: no Companion, no gear. Run: python3 -m unittest discover -s tests""" +import gzip +import json +import logging +import os +import re +import sys +import tempfile +import threading +import time +import unittest +import urllib.parse +from datetime import datetime +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, ROOT) + +import make_page # noqa: E402 +import status_panel as sp # noqa: E402 + +sp.log.addHandler(logging.NullHandler()) +sp.log.propagate = False + +CONFIG = sp.load_config(os.path.join(ROOT, "config.json")) +# The tests pin their own schedule so they do not depend on the site's choice. +CONFIG["service_windows"] = [{"days": ["sun"], "start": "07:00", "end": "12:30"}] +CONFIG["tiles"] = [dict(t, label="SERVICE") if t["id"] == "service" else t for t in CONFIG["tiles"]] +# A Sunday 08:00 (inside the service window) and a Wednesday 15:00 (outside it). +SUNDAY = datetime(2026, 10, 11, 8, 0).timestamp() +WEDNESDAY = datetime(2026, 10, 7, 15, 0).timestamp() + + +def healthy_snapshot(now): + """Everything in config.json present and fine.""" + snap = sp.Snapshot() + labels = {c["conn"] for t in CONFIG["tiles"] for c in t.get("checks", []) if c.get("conn")} + snap.conns = {l: {"id": "id-" + l, "enabled": True, "category": "good", "message": None} for l in labels} + snap.conns_time = now + for i in range(1, 7): + snap.vars[("HyperDeck_%d" % i, "status")] = "Preview" + snap.vars[("HyperDeck_%d" % i, "slotId")] = "1" + snap.vars[("HyperDeck_%d" % i, "recordingTime")] = "28:00:00" + devices = set() + for t in CONFIG["tiles"]: + for c in t.get("checks", []): + if c["type"] == "dante": + devices |= set(c["devices"]) + snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(sorted(devices)) + snap.vars[("tascam", "error")] = "No Error" + snap.vars[("tascam", "caution")] = "No Caution" + for label, n in (("Axient_1", 4), ("Axient_2", 4), ("ULXD-4", 1)): + for ch in range(1, n + 1): + snap.vars[(label, "ch_%d_name" % ch)] = "VOX %d" % ch + snap.vars[(label, "ch_%d_battery_bars" % ch)] = "5" + snap.vars[(label, "ch_%d_interference_status" % ch)] = "NONE" + for label in ("PSM_1", "PSM_2", "PSM_3"): + for ch in (1, 2): + snap.vars[(label, "ch_%d_name" % ch)] = "PACK" + snap.vars[(label, "ch_%d_rf_mute" % ch)] = "UNMUTED" + snap.midi = {"NA Ingest": {"peers": [{"name": "Lighting PC", "address": "", "state": "running"}, + {"name": "SuperRack-LiveBox", "address": "", "state": "running"}]}} + snap.midi_time = now + snap.mdns, snap.mdns_time = {"NA Tracks", "Lighting PC", "NA Ingest"}, now + snap.ping = {"10.1.3.227": 0, "10.6.7.13": 0, "10.1.0.1": 0} + snap.link = True + snap.launchd = {"org.waterschurch.resi-propresenter": 884, "org.waterschurch.lobby-captions": 900} + snap.launchd_time = now + snap.procs = {"/Applications/ProdDash.app/Contents/MacOS/ProdDash": True} + snap.http = {"https://central.resi.io/": True, "http://10.6.7.13:8080/": True} + snap.disk = {"/": 97} + return snap + + +def run_steps(engine, snap, start, seconds, step=1.0): + tiles, t = None, start + while t <= start + seconds: + snap.conns_time = t + tiles = engine.step(snap, t) + t += step + return tiles, t + + +def level(tile): + for name, ((b, f), _) in sp.COLORS.items(): + if (tile["b"], tile["f"]) == (b, f): + return name + return "?" + + +class TestChecks(unittest.TestCase): + def setUp(self): + self.now = SUNDAY + self.snap = healthy_snapshot(self.now) + self.ctx = {"now": self.now, "service": True, "any_recording": False} + + def test_all_green_when_healthy(self): + tiles = sp.Engine(CONFIG).step(self.snap, self.now) + bad = {k: v["x"] for k, v in tiles.items() if level(v) not in ("ok", "service", "off")} + self.assertEqual(bad, {}) + self.assertEqual(tiles["all"]["x"], "ALL OK") + self.assertEqual(tiles["hd1"]["x"], "HD1\n28h") + self.assertEqual(tiles["dante"]["x"], "DANTE\n10/10") + + def test_connection_states(self): + c = self.snap.conns["ATEM"] + c["category"], c["message"] = "error", "Network error: connect ETIMEDOUT 10.1.10.2:9910" + self.assertEqual(sp.conn_result(self.snap, "ATEM", self.now).msg, "NO ANSWER") + c["message"] = "connect EHOSTUNREACH" + self.assertEqual(sp.conn_result(self.snap, "ATEM", self.now).msg, "UNREACHABLE") + c["enabled"] = False + self.assertEqual(sp.conn_result(self.snap, "ATEM", self.now).level, sp.OFF) + self.assertEqual(sp.conn_result(self.snap, "nope", self.now).msg, "NOT IN COMPANION") + self.assertEqual(sp.conn_result(self.snap, "x32", self.now + 60).msg, "NO DATA") + + def test_hyperdeck_disk_and_record_sync(self): + s = self.snap + s.vars[("HyperDeck_2", "slotId")] = "-" + s.vars[("HyperDeck_2", "recordingTime")] = "--:--:--" + out = dict(sp.check_hyperdeck({"conn": "HyperDeck_2"}, s, self.ctx)) + self.assertEqual(out["disk"].msg, "NO DISK") + self.assertTrue(out["disk"].service_only) + s.vars[("HyperDeck_3", "recordingTime")] = "0:20:00" + self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_3"}, s, self.ctx))["disk"].msg, "LOW 20m") + s.vars[("HyperDeck_3", "recordingTime")] = "0:04:00" + self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_3"}, s, self.ctx))["disk"].level, sp.ERR) + ctx = dict(self.ctx, any_recording=True) + self.assertEqual(dict(sp.check_hyperdeck({"conn": "HyperDeck_1"}, s, ctx))["rec"].msg, "NOT REC") + s.vars[("HyperDeck_1", "status")] = "Record" + out = dict(sp.check_hyperdeck({"conn": "HyperDeck_1"}, s, ctx)) + self.assertEqual(out["rec"].level, sp.OK) + self.assertEqual(out["conn"].info, "REC 28h") + + def test_optional_deck(self): + for i in (2, 4): + self.snap.vars[("HyperDeck_%d" % i, "slotId")] = "-" + self.snap.vars[("HyperDeck_%d" % i, "recordingTime")] = "--:--:--" + for i in (1, 3, 5, 6): + self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record" + tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) + self.assertEqual((level(tiles["hd2"]), tiles["hd2"]["x"]), ("off", "HD2\nNO DISK")) + self.assertEqual(level(tiles["rec"]), "ok") + self.assertEqual(tiles["all"]["x"], "ALL OK") + self.snap.vars[("HyperDeck_2", "slotId")] = "1" # disk goes in: now it is watched + self.snap.vars[("HyperDeck_2", "recordingTime")] = "20:00:00" + tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) + self.assertEqual(tiles["hd2"]["x"], "HD2\nNOT REC") + self.assertEqual(level(tiles["hd2"]), "err") + + def test_one_deck_not_recording_blinks(self): + engine = sp.Engine(CONFIG) + for i in (1, 2, 3, 5, 6): + self.snap.vars[("HyperDeck_%d" % i, "status")] = "Record" + tiles, _ = run_steps(engine, self.snap, self.now, 4) + self.assertEqual(tiles["hd4"]["x"], "HD4\nNOT REC") + self.assertEqual(level(tiles["hd4"]), "err") + self.assertEqual(tiles["all"]["x"], "1 DOWN\nHD4 NOT REC") + self.assertEqual(tiles["rec"]["x"], "REC\nHD4 NOT REC") + + def test_dante(self): + self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(["NA-Tracks"]) + out = dict(sp.check_dante({"conn": "audinate-dantecontroller", + "devices": {"NA-Tracks": "DANTE", "BOH-WING": "WING"}}, self.snap, self.ctx)) + self.assertEqual(out["BOH-WING"].msg, "WING GONE") + self.assertEqual(out["NA-Tracks"].level, sp.OK) + self.snap.vars[("audinate-dantecontroller", "devices")] = None + out = dict(sp.check_dante({"conn": "audinate-dantecontroller", "devices": ["BOH-WING"]}, self.snap, self.ctx)) + self.assertEqual(out["conn"].msg, "NO DATA") + + def test_shure_psm_tascam(self): + s = self.snap + s.vars[("Axient_1", "ch_1_battery_bars")] = "1" + s.vars[("Axient_1", "ch_2_battery_bars")] = "2" + s.vars[("Axient_1", "ch_3_battery_bars")] = "Unknown" + s.vars[("Axient_1", "ch_4_interference_status")] = "DETECTED" + s.vars[("Axient_1", "ch_3_interference_status")] = "DETECTED" # transmitter off: ignored + out = dict(sp.check_shure({"conn": "Axient_1", "channels": 4}, s, self.ctx)) + self.assertEqual((out["ch1_batt"].level, out["ch1_batt"].msg), (sp.ERR, "VOX 1 BATT")) + self.assertEqual(out["ch2_batt"].level, sp.WARN) + self.assertEqual(out["ch3_batt"].level, sp.OK) + self.assertEqual(out["ch3_rf"].level, sp.OK) + self.assertEqual(out["ch4_rf"].msg, "VOX 4 INTERF") + s.vars[("PSM_2", "ch_1_rf_mute")] = "MUTED" + self.assertEqual(dict(sp.check_psm({"conn": "PSM_2", "channels": 2}, s, self.ctx))["ch1_mute"].msg, + "PACK RF MUTE") + s.vars[("tascam", "error")] = "Media Error" + self.assertEqual(dict(sp.check_tascam({"conn": "tascam"}, s, self.ctx))["error"].msg, "MEDIA ERROR") + + def test_rf_group_names_the_mic(self): + self.snap.vars[("Axient_2", "ch_3_battery_bars")] = "1" + tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) + self.assertEqual(tiles["rf"]["x"], "RF\nVOX 3 BATT") + self.snap.vars[("Axient_2", "ch_3_battery_bars")] = "5" + self.snap.conns["PSM_1"]["category"] = "error" + tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) + self.assertIn("PSM 1 OFFLINE", tiles["rf"]["x"]) + + def test_midi_peer_joining_is_not_an_error(self): + for p in self.snap.midi["NA Ingest"]["peers"]: + p["state"] = "connected" # joined, clock sync not finished yet + tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 8) + self.assertEqual(tiles["lightpc"]["x"], "LIGHT PC\nMIDI OK") + + def test_midi_peer_and_bonjour(self): + self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"] + if p["name"] != "Lighting PC"] + self.snap.mdns.discard("NA Tracks") + tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) + self.assertEqual(tiles["lightpc"]["x"], "LIGHT PC\nMIDI LOST") + self.assertEqual(tiles["tracks"]["x"], "TRACKS\nMIDI SESS GONE") + + def test_bible_tv_app_down(self): + self.snap.http["http://10.6.7.13:8080/"] = False + tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) + self.assertEqual(tiles["bibletv"]["x"], "BIBLE TV\nAPP DOWN") + + def test_http_intervals(self): + checks = dict(sp.http_checks(CONFIG)) + self.assertEqual(checks["http://10.6.7.13:8080/"], 5) + self.assertEqual(checks["https://central.resi.io/"], 30) + + def test_ingest_services(self): + del self.snap.launchd["org.waterschurch.resi-propresenter"] + self.snap.launchd["org.waterschurch.lobby-captions"] = None + self.snap.disk["/"] = 6 + tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4) + self.assertEqual(tiles["ingest"]["x"], "INGEST\nBRIDGE NOT LOADED\n+2") + + +class TestTextSize(unittest.TestCase): + def test_no_word_is_split(self): + for text in ("TASCAM\nOK", "SERVICE\nON\nmanual", "OBS\nOFFLINE", "INGEST\n97GB free", + "4 DOWN\nHD2 NO DISK", "REC\nHD4 NO DISK\n+1", "TRACKS\nMIDI SESS GONE"): + size = sp.fit_size(text) + longest = max(len(w) for w in text.split()) + self.assertLessEqual(longest, int(sp.LINE_CAPACITY / size), text) + + def test_short_text_stays_big(self): + self.assertGreaterEqual(sp.fit_size("HD1\n28h"), 29) + self.assertGreater(sp.fit_size("ATEM\nOK"), sp.fit_size("TASCAM\nOK")) + + def test_every_tile_has_a_size(self): + tiles = sp.Engine(CONFIG).step(healthy_snapshot(SUNDAY), SUNDAY) + for t in tiles.values(): + self.assertTrue(12 <= t["s"] <= 30, t) + + +class TestBehaviour(unittest.TestCase): + def setUp(self): + self.snap = healthy_snapshot(SUNDAY) + + def test_debounce(self): + engine = sp.Engine(CONFIG) + self.snap.conns["x32"]["category"] = "error" + tiles = engine.step(self.snap, SUNDAY) + self.assertEqual(level(tiles["x32"]), "ok") # one bad reading is not enough + tiles, t = run_steps(engine, self.snap, SUNDAY + 1, 3) + self.assertEqual(level(tiles["x32"]), "err") + self.snap.conns["x32"]["category"] = "good" + self.snap.conns_time = t + self.assertEqual(level(engine.step(self.snap, t)["x32"]), "ok") # recovery is immediate + + def test_blip_shorter_than_debounce_is_ignored(self): + engine = sp.Engine(CONFIG) + self.snap.conns["x32"]["category"] = "error" + run_steps(engine, self.snap, SUNDAY, 1) + self.snap.conns["x32"]["category"] = "good" + tiles, _ = run_steps(engine, self.snap, SUNDAY + 2, 5) + self.assertEqual(level(tiles["x32"]), "ok") + + def test_flapping_shows_amber(self): + engine = sp.Engine(CONFIG) + t = SUNDAY + for _ in range(3): + self.snap.conns["x32"]["category"] = "error" + _, t = run_steps(engine, self.snap, t, 4) + self.snap.conns["x32"]["category"] = "good" + tiles, t = run_steps(engine, self.snap, t, 2) + self.assertEqual(level(tiles["x32"]), "warn") + self.assertEqual(tiles["x32"]["x"], "X32\nOFFLINE x3") + tiles, _ = run_steps(engine, self.snap, t + 700, 1) # quiet for 10 minutes -> green again + self.assertEqual(level(tiles["x32"]), "ok") + + def test_ack_then_recover_then_fail_again(self): + engine = sp.Engine(CONFIG) + self.snap.conns["ATEM"]["category"] = "error" + tiles, t = run_steps(engine, self.snap, SUNDAY, 4) + self.assertEqual(level(tiles["atem"]), "err") + self.assertEqual(engine.ack("atem"), 1) + tiles, t = run_steps(engine, self.snap, t, 1) + self.assertEqual(level(tiles["atem"]), "ack") + self.assertEqual(level(tiles["all"]), "ack") + self.snap.conns["ATEM"]["category"] = "good" + tiles, t = run_steps(engine, self.snap, t, 1) + self.assertEqual(level(tiles["atem"]), "ok") + self.snap.conns["ATEM"]["category"] = "error" + tiles, t = run_steps(engine, self.snap, t, 4) + self.assertEqual(level(tiles["atem"]), "err") # a new failure blinks again + + def test_group_and_master_ack(self): + engine = sp.Engine(CONFIG) + self.snap.conns["VX1000"]["category"] = "error" + self.snap.conns["x32"]["category"] = "error" + tiles, t = run_steps(engine, self.snap, SUNDAY, 4) + self.assertEqual(tiles["all"]["x"].split("\n")[0], "2 DOWN") + engine.ack("video") + tiles, t = run_steps(engine, self.snap, t, 1) + self.assertEqual(level(tiles["video"]), "ack") + self.assertEqual(level(tiles["x32"]), "err") + engine.ack("all") + tiles, t = run_steps(engine, self.snap, t, 1) + self.assertEqual(level(tiles["all"]), "ack") + + def test_outside_service_nothing_blinks(self): + snap = healthy_snapshot(WEDNESDAY) + snap.conns["OBS-Youtube"]["category"] = "error" # OBS PC is off midweek + snap.conns["x32"]["category"] = "error" + snap.vars[("HyperDeck_3", "slotId")] = "-" # a deck that is always used + tiles, _ = run_steps(sp.Engine(CONFIG), snap, WEDNESDAY, 4) + self.assertEqual(level(tiles["obs"]), "idle") + self.assertEqual(level(tiles["hd3"]), "idle") + self.assertEqual(tiles["hd3"]["x"], "HD3\nNO DISK") + self.assertEqual(level(tiles["x32"]), "quiet") + self.assertEqual(tiles["all"]["x"], "1 DOWN\nX32 OFFLINE") + self.assertEqual(level(tiles["all"]), "quiet") + self.assertEqual(tiles["service"]["x"], "SERVICE\nOFF") + for t in tiles.values(): + self.assertEqual((t["b"], t["f"]), (t["b2"], t["f2"]), t["x"]) # no blinking anywhere + + def test_service_schedule_and_override(self): + path = os.path.join(tempfile.mkdtemp(), "service.json") + mode = sp.ServiceMode(CONFIG["service_windows"], path) + self.assertTrue(mode.on(SUNDAY)) + self.assertFalse(mode.on(WEDNESDAY)) + mode.set(True, WEDNESDAY) # button: on for a Wednesday event + self.assertTrue(mode.on(WEDNESDAY + 3600)) + self.assertTrue(sp.ServiceMode(CONFIG["service_windows"], path).on(WEDNESDAY + 3600)) # survives restart + sunday_start = datetime(2026, 10, 11, 7, 0).timestamp() + self.assertTrue(mode.on(sunday_start)) # window edge clears the override + self.assertIsNone(mode.override) + mode.set(False, sunday_start + 60) # turned off early on Sunday + self.assertFalse(mode.on(sunday_start + 120)) + self.assertFalse(mode.on(datetime(2026, 10, 11, 12, 31).timestamp())) + self.assertIsNone(mode.override) # end of window clears it too + + def test_network_burst(self): + engine = sp.Engine(CONFIG) + t = SUNDAY + run_steps(engine, self.snap, t, 2) + for label in ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam"): # the Oct 5 10:21 drop + self.snap.conns[label]["category"] = "error" + tiles, t = run_steps(engine, self.snap, t + 3, 4) + self.assertEqual(tiles["net"]["x"], "NET\n5 DROPPED") + self.assertEqual(level(tiles["net"]), "err") + for label in ("x32", "VX1000", "UHDJr", "RESI_Decoder", "tascam"): + self.snap.conns[label]["category"] = "good" + tiles, t = run_steps(engine, self.snap, t, 1) + self.assertTrue(tiles["net"]["x"].startswith("NET\nBLIP ")) + self.assertEqual(level(tiles["net"]), "warn") + tiles, _ = run_steps(engine, self.snap, t + 200, 1) + self.assertEqual(tiles["net"]["x"], "NET\nOK") + + def test_racks_powering_off_is_not_a_network_event(self): + engine = sp.Engine(CONFIG) + run_steps(engine, self.snap, SUNDAY, 2) + for label in ("Axient_1", "Axient_2", "ULXD-4", "PSM_1", "PSM_2", "PSM_3"): + self.snap.conns[label]["category"] = "error" + tiles, _ = run_steps(engine, self.snap, SUNDAY + 3, 4) + self.assertEqual(tiles["net"]["x"], "NET\nOK") + self.assertEqual(tiles["rf"]["x"].split("\n")[-1], "+5") + + def test_gateway_and_link(self): + self.snap.ping["10.1.0.1"] = 2 + tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1) + self.assertEqual(tiles["net"]["x"], "NET\nNO GATEWAY") + self.snap.link = False + tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, SUNDAY, 1) + self.assertEqual(tiles["net"]["x"], "NET\nLINK DOWN\n+1") + + def test_restart_targets(self): + engine = sp.Engine(CONFIG) + self.assertEqual(engine.restart_targets("hd3"), ["HyperDeck_3"]) + self.assertEqual(engine.restart_targets("led"), ["VX1000", "UHDJr"]) + self.assertEqual(engine.restart_targets("tracks"), ["ableton"]) # not the shared Dante one + self.assertEqual(engine.restart_targets("ingest"), []) + + +class FakeCompanion(BaseHTTPRequestHandler): + """Just enough of Companion 5's HTTP API.""" + store = {} + + def log_message(self, *a): + pass + + def _send(self, status, body): + data = body.encode() if isinstance(body, str) else body + self.send_response(status) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_GET(self): + s = self.store + path = urllib.parse.urlparse(self.path).path + if path == "/api/connections": + return self._send(200, json.dumps([{"id": c["id"], "label": l, "enabled": c["enabled"], + "status": {"category": c["category"], "message": c["message"]}} + for l, c in s["conns"].items()])) + m = re.match(r"^/api/variable/([^/]+)/([^/]+)/value$", path) + if m: + key = (urllib.parse.unquote(m.group(1)), urllib.parse.unquote(m.group(2))) + v = s["vars"].get(key) + return self._send(404, "Not found") if v is None else self._send(200, str(v)) + m = re.match(r"^/api/custom-variable/([^/]+)/value$", path) + if m: + v = s["custom"].get(m.group(1)) + return self._send(404, "Not found") if v is None else self._send(200, v) + self._send(404, "") + + def do_POST(self): + s = self.store + u = urllib.parse.urlparse(self.path) + m = re.match(r"^/api/custom-variable/([^/]+)/value$", u.path) + if m: + if m.group(1) not in s["custom"]: + return self._send(404, "Not found") + s["custom"][m.group(1)] = urllib.parse.parse_qs(u.query, keep_blank_values=True)["value"][0] + return self._send(200, "ok") + m = re.match(r"^/api/connections/([^/]+)/restart$", u.path) + if m: + s["restarts"].append(m.group(1)) + return self._send(200, "{}") + self._send(404, "") + + +class TestAgainstFakeCompanion(unittest.TestCase): + @classmethod + def setUpClass(cls): + snap = healthy_snapshot(time.time()) + FakeCompanion.store = { + "conns": snap.conns, + "vars": dict(snap.vars), + "custom": {"sp_state": "", "sp_hb": "", "sp_cmd": "", "flash_tick": "0"}, + "restarts": [], + } + cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeCompanion) + threading.Thread(target=cls.server.serve_forever, daemon=True).start() + cls.port = cls.server.server_address[1] + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + + def test_client_round_trip_and_long_state(self): + comp = sp.Companion("127.0.0.1", self.port) + self.assertEqual(len(comp.connections()), len(FakeCompanion.store["conns"])) + self.assertEqual(comp.variable("HyperDeck_1", "recordingTime"), "28:00:00") + self.assertIsNone(comp.variable("HyperDeck_1", "nope")) + snap = healthy_snapshot(time.time()) + tiles = sp.Engine(CONFIG).step(snap, time.time()) + js = json.dumps(tiles, separators=(",", ":"), sort_keys=True) + self.assertEqual(comp.set_custom("sp_state", js), (True, 200)) + self.assertEqual(json.loads(comp.get_custom("sp_state"))["all"]["x"], "ALL OK") + self.assertEqual(comp.set_custom("missing", "x"), (False, 404)) + print("\n sp_state is %d bytes, %d in the URL" % (len(js), len(urllib.parse.quote(js, safe="")))) + + def test_commands(self): + comp = sp.Companion("127.0.0.1", self.port) + snap = sp.Snapshot() + sp.poll_connections(comp, snap) + engine = sp.Engine(CONFIG) + now = SUNDAY + sp.handle_commands(";restart hd3;restart led;service off;bogus", engine, comp, snap, now) + self.assertEqual(FakeCompanion.store["restarts"][-3:], ["id-HyperDeck_3", "id-VX1000", "id-UHDJr"]) + self.assertFalse(engine.service.on(now)) + sp.handle_commands("service toggle", engine, comp, snap, now) + self.assertTrue(engine.service.on(now)) + + def test_end_to_end_loop(self): + """The real run() loop: it writes state and heartbeat, and consumes button commands.""" + cfg = json.loads(json.dumps(CONFIG)) + cfg["companion"]["port"] = self.port + # keep it offline: no pings, no web, no launchd, no bonjour lookups + cfg["network"]["gateway"] = None + keep = ("conn", "hyperdeck", "dante", "tascam", "shure", "psm") + for t in cfg["tiles"]: + if "checks" in t: + t["checks"] = [c for c in t["checks"] if c["type"] in keep] or [{"type": "conn", "conn": "x32"}] + store = FakeCompanion.store + store["conns"]["ATEM"]["category"] = "error" + store["custom"]["sp_cmd"] = ";ack atem" + data_dir = tempfile.mkdtemp() + originals = sp.poll_mdns, sp.poll_system, sp.poll_midi + sp.poll_mdns = lambda snap, seconds=3: None + sp.poll_system = lambda snap, cfg: None + sp.poll_midi = lambda snap, cfg: None # CoreMIDI has its own test + stop = threading.Event() + loop = threading.Thread(target=sp.run, args=(cfg, data_dir), kwargs={"stop": stop}, daemon=True) + try: + loop.start() + deadline = time.time() + 12 + state = {} + while time.time() < deadline: + raw = store["custom"]["sp_state"] + state = json.loads(raw) if raw else {} + if state.get("atem", {}).get("x") == "ATEM\nOFFLINE": + break + time.sleep(0.25) + finally: + stop.set() + loop.join(timeout=15) + sp.poll_mdns, sp.poll_system, sp.poll_midi = originals + store["conns"]["ATEM"]["category"] = "good" + self.assertFalse(loop.is_alive()) # it shuts down cleanly + self.assertTrue(store["custom"]["sp_hb"].isdigit()) + self.assertLess(abs(int(store["custom"]["sp_hb"]) - time.time()), 5) + self.assertEqual(store["custom"]["sp_cmd"], "") # the command was consumed + self.assertIn("atem", state) + self.assertIn("ATEM", state["atem"]["x"]) + + +class TestPage(unittest.TestCase): + def setUp(self): + self.export = make_page.build(CONFIG) + self.controls = self.export["page"]["controls"] + + def test_layout(self): + self.assertEqual(self.export["type"], "page") + self.assertEqual(self.export["oldPageNumber"], 19) + self.assertEqual(self.controls["0"]["0"], {"type": "pageup"}) + self.assertEqual(self.controls["2"]["0"], {"type": "pagedown"}) + positioned = [t for t in CONFIG["tiles"] if "pos" in t] + n = sum(1 for r in self.controls.values() for c in r.values() if c["type"] == "button-layered") + self.assertEqual(n, len(positioned)) + # The master is where the smallest Stream Deck can see it. + self.assertEqual(CONFIG["tiles"][0]["pos"], [0, 1]) + + def test_each_button_reads_its_own_tile(self): + for t in CONFIG["tiles"]: + if "pos" not in t: + continue + r, c = (str(x) for x in t["pos"]) + layers = {l["id"]: l for l in self.controls[r][c]["style"]["layers"]} + for lid, key in (("box0", "color"), ("text0", "color"), ("text0", "text"), ("text0", "fontsize")): + expr = layers[lid][key] + self.assertTrue(expr["isExpression"]) + self.assertIn("'$.%s." % t["id"], expr["value"]) + self.assertIn("$(custom:sp_hb) > 15", expr["value"]) + self.assertIn("$(custom:flash_tick) == 1", layers["box0"]["color"]["value"]) + + def test_actions(self): + hd1 = self.controls["1"]["1"]["steps"]["0"] + up = hd1["action_sets"]["up"][0]["options"]["value"] + self.assertEqual(up, {"value": "$(custom:sp_cmd);ack hd1", "isExpression": False}) + self.assertEqual(hd1["action_sets"]["2000"][0]["options"]["value"]["value"], "$(custom:sp_cmd);restart hd1") + self.assertEqual(hd1["options"]["runWhileHeld"], [2000]) + master = self.controls["0"]["1"]["steps"]["0"] + self.assertIn("ackall", master["action_sets"]["up"][0]["options"]["value"]["value"]) + self.assertNotIn("2000", master["action_sets"]) + ingest = self.controls["3"]["5"]["steps"]["0"] + self.assertNotIn("2000", ingest["action_sets"]) # nothing to restart there + ids = [a["id"] for r in self.controls.values() for c in r.values() + for s in c.get("steps", {}).values() for al in s["action_sets"].values() for a in al] + self.assertEqual(len(ids), len(set(ids))) + + def test_stale_text_and_file(self): + master = {l["id"]: l for l in self.controls["0"]["1"]["style"]["layers"]} + self.assertIn("`${'WATCHDOG\\nDOWN\\n'}${$(internal:instance_errors)} ERR`", master["text0"]["text"]["value"]) + # No string '+' anywhere: in Companion 5 expressions it adds numbers. + for r in self.controls.values(): + for c in r.values(): + for layer in c.get("style", {}).get("layers", []): + for p in layer.values(): + if isinstance(p, dict) and p.get("isExpression"): + self.assertNotRegex(p["value"], r"'\s*\+|\+\s*'") + path = os.path.join(tempfile.mkdtemp(), "p.companionconfig") + sys.argv = ["make_page.py", path] + make_page.main() + on_disk = make_page.build(sp.load_config(os.path.join(ROOT, "config.json"))) + with gzip.open(path) as f: + self.assertEqual(json.load(f), json.loads(json.dumps(on_disk))) + + +class TestCoreMIDI(unittest.TestCase): + def test_listener_loopback(self): + import coremidi + src = coremidi.VirtualSource("status-panel-test") + try: + time.sleep(0.3) + lis = coremidi.Listener("status-panel-test") + lis.poll() + self.assertTrue(lis.connected) + src.send([0x90, 36, 100]) + src.send([0xF8]) # clock is ignored + time.sleep(0.3) + self.assertEqual((lis.last_msg, lis.count), ("NOTE 36", 1)) + snap = sp.Snapshot() + snap.listeners["status-panel-test"] = lis + res = sp.check_midi_rx({"source": "status-panel-test", "name": "IN"}, snap, + {"now": time.time(), "service": True})[0][1] + self.assertEqual(res.info, "IN 0s") + finally: + src.close() + time.sleep(0.3) + lis.poll() + self.assertFalse(lis.connected) + + def test_describe(self): + import coremidi + self.assertEqual(coremidi.describe([0xB0, 7, 64]), "CC 7=64") + self.assertEqual(coremidi.describe([0xC2, 5]), "PC 5") + self.assertEqual(coremidi.describe([0x90, 60, 0]), "OFF 60") + self.assertEqual(coremidi.describe([0xF1, 0x20]), "MTC") + + +if __name__ == "__main__": + unittest.main()