ALARMS and AUTO-FIX as two keys; LIGHTS at Southcoast hands the lights to an operator
- kind autofix: its own switch on the auto-fix hours; ALARMS then only silences (NA keeps the old combined key until it has room for one) - kind bypass: hold takes every network MIDI link to a peer out (here and on listed Macs), checks stop calling it lost, re-taken every 10 s; hold again puts back exactly what was taken plus every watched link - a MIDI peer with no address is never called (it connected SC Tracks to itself) - holds from NA's CAMPUS page flip switch keys; a rebuild keeps them; new campuses get AUTO-FIX - key kind changes restart the watchdog; settings page has separate alarm / auto-fix buttons - WS and SC configs: AUTO-FIX at 3,1; SC LIGHTS at 1,3; NA config synced from live Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -531,3 +531,18 @@ class TestStandalone(unittest.TestCase):
|
||||
self.assertIn("--standalone", inst)
|
||||
self.assertNotIn("10.1.1.1", inst)
|
||||
self.assertNotIn("authorized_keys", inst) # no one else's key is let in
|
||||
|
||||
|
||||
class TestSwitchKeys(unittest.TestCase):
|
||||
def test_a_new_campus_gets_auto_fix_by_alarms_and_a_rebuild_keeps_its_own_switches(self):
|
||||
new, _ = ac.build(BASE, small(), {"id": "sc", "name": "Southcoast"}, "1234")
|
||||
af = [t for t in new["tiles"] if t.get("kind") == "autofix"]
|
||||
self.assertEqual([t["pos"] for t in af], [[3, 1]]) # beside ALARMS
|
||||
self.assertEqual(schema.validate(new)[0], [])
|
||||
old = json.loads(json.dumps(new))
|
||||
old["tiles"].append({"id": "lights", "label": "LIGHTS", "kind": "bypass", "pos": [2, 7], "bypass": {"peers": ["Lighting"]}})
|
||||
again, _ = ac.build(BASE, small(), {"id": "sc", "name": "Southcoast"}, "1234")
|
||||
kept_cfg, kept = ac.keep_from(again, old)
|
||||
lights = [t for t in kept_cfg["tiles"] if t["id"] == "lights"]
|
||||
self.assertEqual((len(lights), lights[0]["pos"], "LIGHTS" in kept), (1, [2, 7], True))
|
||||
self.assertEqual(len([t for t in kept_cfg["tiles"] if t.get("kind") == "autofix"]), 1)
|
||||
|
||||
+1
-1
@@ -85,7 +85,7 @@ class TestSites(unittest.TestCase):
|
||||
errors, _ = schema.validate(cfg)
|
||||
self.assertEqual(errors, [])
|
||||
make_page.build(cfg)
|
||||
self.assertEqual(sorted(t["kind"] for t in cfg["tiles"]), ["master", "network", "service"])
|
||||
self.assertEqual(sorted(t["kind"] for t in cfg["tiles"]), ["autofix", "master", "network", "service"])
|
||||
self.assertEqual([a for a, h in cfg["hosts"].items() if h.get("self")], ["ingest"])
|
||||
self.assertEqual(cfg["hosts"]["ingest"]["networks"], [])
|
||||
self.assertNotIn("Attleboro", json.dumps(cfg))
|
||||
|
||||
@@ -2451,3 +2451,129 @@ class TestDanteNewcomers(unittest.TestCase):
|
||||
self.assertFalse(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now + 100))
|
||||
self.engine.any_recording = False
|
||||
self.assertTrue(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now + 200))
|
||||
|
||||
|
||||
class TestSwitches(unittest.TestCase):
|
||||
"""ALARMS and AUTO-FIX as two keys (2026-10-21), and a key that hands a device to an operator
|
||||
(Southcoast's LIGHTS: a lighting op runs the desk, so nothing may send it MIDI)."""
|
||||
|
||||
def setUp(self):
|
||||
self.snap = healthy_snapshot(SUNDAY)
|
||||
self.cfg = json.loads(json.dumps(CONFIG))
|
||||
arena = next(t for t in self.cfg["tiles"] if t.get("pos") == [3, 7])
|
||||
arena.pop("pos") # NA's page is full: make room
|
||||
self.cfg["tiles"].append({"id": "autofix", "label": "AUTO-FIX", "kind": "autofix", "pos": [3, 7]})
|
||||
self.midi = next(c for t in self.cfg["tiles"] for c in t.get("checks", []) if c["type"] == "midi_peer")
|
||||
self.cfg["tiles"].append({"id": "lights", "label": "LIGHTS", "kind": "bypass",
|
||||
"bypass": {"peers": [self.midi["peer"]], "hosts": [""], "conns": ["x32"]}})
|
||||
self.state = os.path.join(tempfile.mkdtemp(), "state.json")
|
||||
self.engine = self.new_engine()
|
||||
|
||||
def new_engine(self):
|
||||
e = sp.Engine(self.cfg, self.state)
|
||||
e.executor = e.poe_executor = lambda *a: True
|
||||
e.autolog = sp.AutoLog(tempfile.mkdtemp())
|
||||
return e
|
||||
|
||||
def test_alarms_silence_only_and_auto_fix_has_its_own_key(self):
|
||||
import schema
|
||||
self.assertEqual(schema.validate(self.cfg)[0], [])
|
||||
ran = []
|
||||
self.engine.executor = lambda cfg, fix_id, chk: ran.append(fix_id) or True
|
||||
self.snap.remote["tracks\0pgrep -xq Live"] = False # Ableton closed
|
||||
tiles, _ = run_steps(self.engine, self.snap, SUNDAY, 4)
|
||||
self.assertEqual(tiles["service"]["x"], "SERVICE\nON")
|
||||
self.assertEqual(tiles["autofix"]["x"], "AUTO-FIX\nON\nTIL 1:30PM") # Sunday's hours end then
|
||||
self.engine.service.set(False, SUNDAY + 5) # ALARMS: silence only
|
||||
run_steps(self.engine, self.snap, SUNDAY + 5, 25)
|
||||
self.engine.auto_fix(self.snap, mock_comp(), SUNDAY + 30)
|
||||
time.sleep(0.2)
|
||||
self.assertEqual(ran, ["ableton"]) # auto-fix carried on
|
||||
self.engine.autofix.set(False, SUNDAY + 40) # AUTO-FIX pressed: paused
|
||||
tiles = self.engine.step(self.snap, SUNDAY + 41)
|
||||
self.assertEqual(tiles["autofix"]["x"], "AUTO-FIX\nOFF\nTIL WED 4PM") # back on with Wednesday's hours
|
||||
self.assertEqual(level(tiles["autofix"]), "off")
|
||||
self.assertEqual(self.engine.auto_fix(self.snap, mock_comp(), SUNDAY + 200), [])
|
||||
self.assertFalse(self.new_engine().autofix.on(SUNDAY + 300)) # survives a restart
|
||||
self.assertTrue(self.engine.autofix.on(datetime(2026, 10, 14, 16, 30).timestamp())) # Wednesday's hours: back by itself
|
||||
weekday = SUNDAY - 3 * 86400 # Thursday: pressed = runs anyway
|
||||
self.engine.autofix.set(True, weekday)
|
||||
self.assertTrue(self.engine.autofix.on(weekday + 60))
|
||||
page = make_page.build(self.cfg)["page"]["controls"]
|
||||
notes = {(ctl.get("options") or {}).get("notes", "") for row in page.values() for ctl in row.values() if ctl}
|
||||
self.assertTrue(any("Auto-fix is the AUTO-FIX key" in n for n in notes))
|
||||
self.assertTrue(any("pause auto-fix" in n for n in notes))
|
||||
|
||||
def test_a_hold_from_north_attleboro_flips_a_switch(self):
|
||||
self.engine.run_fixes("autofix", self.snap, mock_comp(), SUNDAY)
|
||||
self.assertFalse(self.engine.autofix.on(SUNDAY + 1))
|
||||
self.engine.run_fixes("autofix", self.snap, mock_comp(), SUNDAY + 2)
|
||||
self.assertTrue(self.engine.autofix.on(SUNDAY + 3))
|
||||
|
||||
def test_the_operator_key_takes_the_midi_out_and_puts_it_back(self):
|
||||
calls = []
|
||||
here = {self.midi["session"]: {"port": 5004, "peers": [{"name": self.midi["peer"], "address": self.midi["address"]},
|
||||
{"name": "Other", "address": "10.9.9.9:5004"}]}}
|
||||
real = (sp.midi_sessions, sp.midi_peer)
|
||||
sp.midi_sessions = lambda cfg, host: here
|
||||
sp.midi_peer = lambda cfg, host, action, session, peer, address=None, repoint=False: calls.append((action, host, session, peer)) or ("asked", None)
|
||||
|
||||
class Comp:
|
||||
def __init__(self):
|
||||
self.on = True
|
||||
self.log = []
|
||||
def connections(self):
|
||||
return [{"label": "x32", "id": "id-x32", "enabled": self.on}]
|
||||
def disable_connection(self, cid):
|
||||
self.on = False; self.log.append("off"); return True
|
||||
def enable_connection(self, cid):
|
||||
self.on = True; self.log.append("on"); return True
|
||||
comp = Comp()
|
||||
try:
|
||||
if self.midi.get("host"):
|
||||
self.snap.remote_midi[self.midi["host"]] = {} # its Mac reports: no peers
|
||||
self.snap.remote_midi_time[self.midi["host"]] = SUNDAY + 100
|
||||
else:
|
||||
self.snap.midi, self.snap.midi_time = {self.midi["session"]: {"port": 5004, "peers": []}}, SUNDAY + 100
|
||||
tiles, _ = run_steps(self.engine, self.snap, SUNDAY + 100, 3)
|
||||
self.assertEqual(level(tiles["lights"]), "ok")
|
||||
self.engine.run_fixes("lights", self.snap, comp, SUNDAY + 104) # hold: an operator has it
|
||||
for _ in range(50):
|
||||
if not self.engine.bypass_busy:
|
||||
break
|
||||
time.sleep(0.05)
|
||||
self.assertEqual(calls, [("disconnect", "", self.midi["session"], self.midi["peer"])]) # never "Other"
|
||||
self.assertEqual(comp.log, ["off"])
|
||||
tiles, _ = run_steps(self.engine, self.snap, SUNDAY + 105, 3)
|
||||
midi_tile = next(t["id"] for t in self.cfg["tiles"] if self.midi in t.get("checks", []))
|
||||
tag = self.midi.get("tag") or self.midi["peer"]
|
||||
self.assertNotIn(tag + " MIDI", tiles[midi_tile]["x"]) # taken out on purpose: no alarm
|
||||
self.assertEqual(tiles["lights"]["x"], "LIGHTS\nTAKING\nMIDI OUT")
|
||||
tiles = self.engine.step(self.snap, SUNDAY + 120)
|
||||
self.assertEqual(tiles["lights"]["x"], "LIGHTS\nOPERATOR\nMIDI OFF")
|
||||
self.assertIn("lights", self.new_engine().bypass) # survives a restart
|
||||
calls.clear()
|
||||
self.engine.run_fixes("lights", self.snap, comp, SUNDAY + 130) # hold again: put back
|
||||
for _ in range(50):
|
||||
if not self.engine.bypass_busy:
|
||||
break
|
||||
time.sleep(0.05)
|
||||
self.assertEqual(calls[0], ("connect", "", self.midi["session"], self.midi["peer"])) # what was taken out
|
||||
self.assertEqual({c[0] for c in calls}, {"connect"}) # and every link the checks watch
|
||||
self.assertEqual(comp.log, ["off", "on"])
|
||||
self.assertNotIn("lights", self.engine.bypass)
|
||||
finally:
|
||||
sp.midi_sessions, sp.midi_peer = real
|
||||
|
||||
def test_a_peer_without_an_address_is_never_called(self):
|
||||
res, why = sp.midi_peer(self.cfg, "", "connect", "SC Tracks NEW", "Lighting", ":5004")
|
||||
self.assertIsNone(res)
|
||||
self.assertIn("calls in", why)
|
||||
|
||||
def test_a_bypass_key_needs_its_peers(self):
|
||||
import schema
|
||||
bad = json.loads(json.dumps(self.cfg))
|
||||
next(t for t in bad["tiles"] if t["id"] == "lights")["bypass"] = {"hosts": ["nowhere"]}
|
||||
errors = schema.validate(bad)[0]
|
||||
self.assertTrue(any("bypass.peers" in e for e in errors))
|
||||
self.assertTrue(any("nowhere" in e for e in errors))
|
||||
|
||||
Reference in New Issue
Block a user