diff --git a/config.json b/config.json index 84975ef..06c3a99 100644 --- a/config.json +++ b/config.json @@ -2731,6 +2731,23 @@ "on_text": "AUTO-OPEN\nOFF", "off_text": "AUTO-OPEN\nON" } + }, + { + "id": "trk_ingest_off", + "label": "TRK INGEST", + "kind": "bypass", + "bypass": { + "peers": [ + "NA Ingest" + ], + "hosts": [ + "tracks" + ], + "auto": "autofix_hours", + "on_text": "OFF IN\nAUTO-FIX HOURS", + "off_text": "ALLOWED", + "names_only": true + } } ], "timing": { diff --git a/schema.py b/schema.py index 4225997..cc7413c 100644 --- a/schema.py +++ b/schema.py @@ -276,7 +276,8 @@ field("tiles[].kind", "enum", None, "A special key: master (ALL), service (ALARM field("tiles[].bypass", "dict", None, "A bypass key: peers (MIDI peer or session names to take out, e.g. Lighting), " "hosts (the Macs whose sessions to look in; \"\" is this Mac) and conns (Companion connections to switch off); or " "fixes (fix ids it holds off: their checks stay green and nothing runs them, e.g. ableset), with on_text / off_text " - "for the key's face.", + "for the key's face. auto \"autofix_hours\": on by itself exactly in auto-fix's hours (NA: Ingest out of Tracks' MIDI); names_only: take " + "out only peers of those names, not others at the same address.", "Keys", advanced=True) field("tiles[].members", "list", None, "A group: the keys it sums up.", "Keys") field("tiles[].checks", "list", [], "What the key watches.", "Keys") @@ -744,7 +745,7 @@ def validate(cfg, live=None): continue toggled = {k.get("toggle") for k in (cfg.get("restart_page") or {}).get("keys", [])} if "pos" not in t and "members" not in t and t.get("id") not in in_group and t.get("kind") != "master" \ - and t.get("id") not in toggled: + and t.get("id") not in toggled and not ((t.get("bypass") or {}).get("auto")): # an automatic switch needs no key warnings.append("key %s has no place on the deck and is in no group: nobody will see it" % t.get("id")) if kinds["master"] != 1: errors.append("there must be exactly one ALL key (kind master); found %d" % kinds["master"]) diff --git a/status_panel.py b/status_panel.py index 0953eeb..25003c2 100644 --- a/status_panel.py +++ b/status_panel.py @@ -602,13 +602,15 @@ def bypass_apply(cfg, tile, on, entry, comp, names_ips, lock=None): if not b.get("peers") and not b.get("conns"): return # a key that only holds off fixes: nothing to hand over names, ips = set(b.get("peers") or []), set(names_ips[1]) if on else set() + if b.get("names_only"): + ips = set() # NA Ingest only, not ProdDash Lighting on the same Mac (2026-10-26) hosts = b.get("hosts") or [""] + sorted({c["host"] for t in cfg["tiles"] for c in t.get("checks", []) if c.get("type") == "midi_peer" and c.get("peer") in names and c.get("host")}) if not on: for c in [x for t in cfg["tiles"] for x in t.get("checks", []) if x.get("type") == "midi_peer" and x.get("peer") in names]: ip = peer_ip(c.get("address")) - if ip: + if ip and not b.get("names_only"): ips.add(str(ip)) conns = {c["label"]: c for c in comp.connections()} if b.get("conns") else {} for label in b.get("conns") or []: @@ -1803,13 +1805,29 @@ class Engine: threading.Thread(target=job, name="bypass-" + tile_id, daemon=True).start() def bypass_tick(self, comp, now): - """While an operator has it, every 10 s: a link that came back (rtpMIDI asking again) goes out again.""" - if not self.bypass or self.bypass_busy or self.stopping or now - self.bypass_last < 10: + """While an operator has it, every 10 s: a link that came back (rtpMIDI asking again) goes out again. + Several switches take turns; one that only holds off fixes (ABLESET) has nothing to re-take.""" + self.auto_bypass(comp, now) + held = [t for t in self.bypass if (((self.tiles.get(t) or {}).get("bypass") or {}).get("peers") + or ((self.tiles.get(t) or {}).get("bypass") or {}).get("conns"))] + if not held or self.bypass_busy or self.stopping or now - self.bypass_last < 10: return self.bypass_last = now - tid = next(iter(self.bypass)) + self.bypass_turn = (getattr(self, "bypass_turn", -1) + 1) % len(held) + tid = held[self.bypass_turn] self.start_bypass_job(tid, True, self.bypass[tid], comp) + def auto_bypass(self, comp, now): + """A switch with "auto": "autofix_hours" is on exactly in auto-fix's hours (NA, 2026-10-26: NA Ingest out of + the Tracks Mac's MIDI session then - its notes were doing odd things in Ableton - and back after them).""" + for tid, t in self.tiles.items(): + if t.get("kind") != "bypass" or ((t.get("bypass") or {}).get("auto") != "autofix_hours"): + continue + want = self.autofix.scheduled(now) + if want != (tid in self.bypass) and not self.bypass_busy: + log.info("BYPASS %s: auto-fix hours %s", t["label"], "began" if want else "ended") + self.toggle_bypass(tid, comp, now) + def run_fixes(self, tile_id, snap, comp, now, cooldown=None): cooldown = self.cooldown if cooldown is None else cooldown if self.stopping: diff --git a/tests/test_status_panel.py b/tests/test_status_panel.py index 6dc83ce..50dc90c 100644 --- a/tests/test_status_panel.py +++ b/tests/test_status_panel.py @@ -2674,6 +2674,36 @@ class TestSwitches(unittest.TestCase): self.assertNotIn("ableset", [w for _, k, w, _ in engine.fix_plan("apps", snap)]) sp.bypass_apply(CONFIG, engine.tiles["ableset"], True, engine.bypass["ableset"], None, (set(), set())) # nothing to hand over + def test_ingest_out_of_tracks_midi_in_auto_fix_hours(self): + """NA, 2026-10-26: NA Ingest's notes did odd things in Ableton - in auto-fix's hours it is taken out of + the Tracks Mac's MIDI session (and kept out), and put back when they end.""" + engine = sp.Engine(CONFIG) + flips = [] + engine.start_bypass_job = lambda tid, on, entry, comp: flips.append((tid, on)) + sunday_8am, monday = datetime(2026, 10, 11, 8, 0).timestamp(), datetime(2026, 10, 12, 9, 0).timestamp() + engine.bypass_tick(None, monday) + self.assertNotIn("trk_ingest_off", engine.bypass) # outside the hours: left as it is + engine.bypass_tick(None, sunday_8am) + self.assertIn("trk_ingest_off", engine.bypass) + self.assertEqual(flips[0], ("trk_ingest_off", True)) + self.assertIn("NA Ingest", engine.bypass_names()[0]) # its MIDI check is green meanwhile + engine.bypass_last = 0 + engine.bypass_tick(None, sunday_8am + 30) # re-taken every 10 s + self.assertEqual(flips[-1], ("trk_ingest_off", True)) + dropped = [] + orig = sp.midi_sessions, sp.midi_peer + sp.midi_sessions = lambda cfg, host: {"NA Tracks": {"peers": [{"name": "NA Ingest", "address": "10.1.1.1:5009"}, + {"name": "ProdDash Lighting", "address": "10.1.1.1:5006"}]}} + sp.midi_peer = lambda cfg, host, verb, sess, peer, *a, **k: dropped.append((verb, peer)) or ("asked", None) + try: + sp.bypass_apply(CONFIG, engine.tiles["trk_ingest_off"], True, {"dropped": []}, None, engine.bypass_names()) + finally: + sp.midi_sessions, sp.midi_peer = orig + self.assertEqual(dropped, [("disconnect", "NA Ingest")]) # ProdDash Lighting, same Mac, stays + engine.bypass_tick(None, sunday_8am + 7 * 3600) # 15:00: the hours are over + self.assertNotIn("trk_ingest_off", engine.bypass) + self.assertEqual(flips[-1], ("trk_ingest_off", False)) + def test_a_bypass_key_needs_its_peers(self): import schema bad = json.loads(json.dumps(self.cfg))