NA: NA Ingest out of the Tracks Mac's MIDI session in auto-fix hours (rehearsals, services); back after
User 2026-10-26: Ingest's notes did odd things in Ableton; the link is needed for programming only. An automatic switch (bypass auto autofix_hours, names_only so ProdDash Lighting on the same Mac stays) takes it out and keeps it out, and puts it back when the hours end. bypass_tick now takes turns over every switch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
+17
@@ -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": {
|
||||
|
||||
@@ -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"])
|
||||
|
||||
+22
-4
@@ -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:
|
||||
|
||||
@@ -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))
|
||||
|
||||
Reference in New Issue
Block a user