christian/status-indicators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MIDI alarm when the Lighting PC drops out of NA Tracks' session (FX lights)

Browse Source 
Tracks sends the FX lighting cues to the Lighting PC through NA Tracks'
network MIDI session. A midi_peer check may now name a "host": the watchdog
reads that Mac's sessions over SSH every 5 s, sending this folder's
coremidi.py over stdin (nothing installed there). Unreachable = no verdict;
the Tracks Mac's own ping check covers that.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-08 15:48:25 -04:00
co-authored by Claude Opus 5.5
parent eef3f37fa9
commit 74fbe73a4b
4 changed files with 70 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_status_panel.py
+28 -2
View File
@@ -4,6 +4,7 @@ import json
import logging
import os
import re
import subprocess
import sys
import tempfile
import threading
@@ -244,6 +245,28 @@ class TestChecks(unittest.TestCase):
finally:
stop.set()
def test_fx_lights_midi_on_the_tracks_mac(self):
engine = sp.Engine(CONFIG)
self.assertEqual(engine.step(self.snap, self.now)["midi"]["x"], "MIDI\nOK") # not read yet: no verdict
self.snap.remote_midi["tracks"] = {"NA Tracks": {"port": 5004, "peers": [
{"name": "SuperRack-LiveBox", "address": "", "state": "running"},
{"name": "Lighting PC", "address": "", "state": "running"}]}}
self.assertEqual(engine.step(self.snap, self.now)["midi"]["x"], "MIDI\nOK")
self.snap.remote_midi["tracks"]["NA Tracks"]["peers"].pop()
tiles, _ = run_steps(engine, self.snap, self.now, 4)
self.assertEqual(tiles["midi"]["x"], "MIDI\nFX LIGHTS MIDI LOST")
self.snap.remote_midi["tracks"] = None # Tracks Mac unreachable: its ping says so
tiles, _ = run_steps(engine, self.snap, self.now + 5, 1)
self.assertEqual(tiles["midi"]["x"], "MIDI\nOK")
def test_remote_midi_script_runs_coremidi(self):
r = subprocess.run(["/usr/bin/python3", "-"], input=sp.remote_midi_script(),
capture_output=True, text=True, timeout=30)
self.assertEqual(r.returncode, 0, r.stderr)
self.assertIsInstance(sp.parse_remote_midi(r.stdout), dict)
self.assertIsNone(sp.parse_remote_midi("ssh: connect to host timed out"))
self.assertEqual(sp.remote_midi_hosts(CONFIG), ["tracks"])
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
@@ -932,12 +955,14 @@ class TestAgainstFakeCompanion(unittest.TestCase):
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_remote, sp.poll_dante_map
originals = (sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map,
sp.poll_remote_midi)
sp.poll_dante_map = lambda *a, **k: None # never log in to a switch from a test
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
sp.poll_remote = lambda snap, cfg: None # never SSH anywhere from a test
sp.poll_remote_midi = lambda snap, cfg: None
stop = threading.Event()
loop = threading.Thread(target=sp.run, args=(cfg, data_dir), kwargs={"stop": stop}, daemon=True)
try:
@@ -953,7 +978,8 @@ class TestAgainstFakeCompanion(unittest.TestCase):
finally:
stop.set()
loop.join(timeout=15)
sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map = originals
(sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map,
sp.poll_remote_midi) = originals
store["conns"]["ATEM"]["category"] = "good"
self.assertFalse(loop.is_alive()) # it shuts down cleanly
self.assertTrue(store["custom"]["sp_hb"].isdigit())