diff --git a/README.md b/README.md
index 62135fa..a1b6e81 100644
--- a/README.md
+++ b/README.md
@@ -68,7 +68,9 @@ Row 0 holds the group buttons. Each one shows the first problem in its group, pl
**The Dante map is automatic.** Every 10 minutes the watchdog maps each Dante device: its name and IP from Companion, its MAC from NA Ingest's Dante port, and its switch port and PoE from the Netgear switches listed in `"dante_switches"`. It saves the map to `dante-map.json` in its data folder. Re-patching, adding or renaming a device needs no config change. A device that has been dead for a while ages out of the switch tables, so its last known port is used, but only if no other device has appeared there. A port carrying more than 2 devices (an uplink) is never power-cycled. The switch login is in NA Ingest's login keychain (service `netgear-status-panel`).
It checks over SSH every 10 s that ProPresenter, Ableton and Church Translate are running, using NA Ingest's key `~/.ssh/id_status_panel`. That key was authorized on the Macs with `remote-access/mac-authorize.sh`.
-- **ALARMS:** alarms are armed all the time (`service_windows` in `config.json`: every day, 00:00–23:59). Press ALARMS to mute the blinking **and stop every auto-fix**, for example during maintenance; press it again to turn both back on. Both come back by themselves at midnight. The key says both states (`ALARMS OFF` / `AUTO-FIX OFF` / `TIL 12AM`). Holds still fix while muted; the Sunday stream-audio failover still switches by itself. The settings page's Overview has the same switch.
+- **ALARMS:** alarms are armed all the time (`service_windows` in `config.json`: every day, 00:00–23:59). Press ALARMS to mute the blinking; press it again to turn them back on; they come back by themselves at midnight. Holds still fix while muted; the Sunday stream-audio failover still switches by itself. The settings page's Overview has the same switch. On a page with no AUTO-FIX key (NA today), ALARMS also stops every auto-fix and the key says both (`ALARMS OFF` / `AUTO-FIX OFF` / `TIL 12AM`).
+- **AUTO-FIX** (beside ALARMS on the campuses; `kind: autofix`): its own switch since 2026-10-21. On in the auto-fix hours. Press: pause it (`AUTO-FIX OFF / TIL SUN 5AM`: until the hours next start), or run it outside the hours until they next start or end; it goes back to its hours by itself. Holds always fix. With this key, ALARMS only silences. From NA's CAMPUS page a hold is the press. The Overview has a Pause auto-fix button.
+- **LIGHTS** (Southcoast; `kind: bypass`, `"bypass": {"peers": ["Lighting"], "hosts": ["", "sc_tracks_new"]}`): hold 2 s to hand the lights to a lighting operator. Every network MIDI link to Lighting (ProPresenter's session, Tracks' session, and Companion's 56 Midi-Lighting buttons, which go out through ProPresenter's session) is taken out, the MIDI checks stop calling it lost and auto-fix stops putting it back; the key says `LIGHTS / OPERATOR / MIDI OFF` (steady blue). A link that comes back while the operator has it is taken out again within 10 s. Hold again: everything taken out is put back as it was, and every watched link to Lighting is reconnected (`LIGHTS / AUTO`). A peer that had called in (no address, ":5004") is not called back: it rejoins when Lighting calls again. Survives a restart; a hold from NA's CAMPUS page works too. `"conns"` can also switch Companion connections off for the operator.
## Auto-fix on Sundays (06:30–13:30)
@@ -270,7 +272,7 @@ once. Everything it makes is ordinary settings, changed on the page like North A
| Apollo Beach (standalone) | their own | their Companion Mac | 19 / 20 or the first free |
Every campus: alarms armed all day; auto-fix on NA's hours (Tue 14:00–22:00, Wed 09:00–21:00, Sun 05:00–14:00);
-a hold on any key fixes it at any time, on that campus's own deck or from NA's CAMPUS page; ALARMS stops auto-fix.
+a hold on any key fixes it at any time, on that campus's own deck or from NA's CAMPUS page; ALARMS silences and AUTO-FIX pauses auto-fix, separately.
`python3 -m settings.campus rebuild` (on a campus's Companion Mac, in the panel folder) makes its setup again from
what is found now, keeping its name, PIN and pages. `python3 -m settings.campus upgrade-companion` brings an older
Companion to 5.0.7 from NA's copy (settings come along; the old version stays installed).
diff --git a/coremidi.py b/coremidi.py
index df8a268..c7282dd 100644
--- a/coremidi.py
+++ b/coremidi.py
@@ -266,6 +266,39 @@ if (mine.length && (stuck || (REPOINT && elsewhere))) {
"""
+# Disconnect, for a Mac without python: what disconnect_peer does. Prints SPMIDI"asked" / "not connected".
+JXA_DISCONNECT = r"""
+ObjC.import('CoreMIDI'); ObjC.import('Foundation');
+var SESSION = __SESSION__, PEER = __PEER__;
+function str(o, k) { var r = Ref(); if ($.MIDIObjectGetStringProperty(o, k, r) !== 0) return null; return ObjC.unwrap(ObjC.castRefToObject(r[0])); }
+function dict(o, k) { var r = Ref(); if ($.MIDIObjectGetDictionaryProperty(o, k, r) !== 0) return null; return ObjC.deepUnwrap(ObjC.castRefToObject(r[0])); }
+var found = null, n = $.MIDIGetNumberOfDevices();
+for (var i = 0; i < n && !found; i++) {
+ var dev = $.MIDIGetDevice(i), drv = str(dev, $('driver')) || '';
+ if (drv.indexOf('Network') < 0 && drv.toLowerCase().indexOf('rtp') < 0) continue;
+ for (var j = 0; j < $.MIDIDeviceGetNumberOfEntities(dev); j++) {
+ var ent = $.MIDIDeviceGetEntity(dev, j), s = dict(ent, $('apple.midirtp.session'));
+ if (s && (s['dns-sd-name'] || str(ent, $('name'))) === SESSION) { found = [ent, s]; break; }
+ }
+}
+if (!found) throw new Error('no network MIDI session named ' + SESSION + ' on this Mac');
+var ent = found[0], s = found[1], peers = s.peers || [], result = 'not connected';
+var left = peers.filter(function (p) { return p.name !== PEER; });
+if (left.length !== peers.length) {
+ s.peers = left;
+ var e = $.MIDIObjectSetDictionaryProperty(ent, $('apple.midirtp.session'), $(s));
+ if (e !== 0) throw new Error('MIDIObjectSetDictionaryProperty failed: ' + e);
+ result = 'asked';
+}
+'SPMIDI' + JSON.stringify(result);
+"""
+
+
+def jxa_disconnect(session_name, peer_name):
+ import json as _json
+ return JXA_DISCONNECT.replace("__SESSION__", _json.dumps(session_name)).replace("__PEER__", _json.dumps(peer_name))
+
+
def jxa_connect(session_name, peer_name, address, repoint=False):
"""The JXA_CONNECT script for one peer (values go in as JSON, never as code)."""
import json as _json
diff --git a/make_page.py b/make_page.py
index 831466d..5a2f01b 100644
--- a/make_page.py
+++ b/make_page.py
@@ -198,9 +198,25 @@ def button(tid, tile, cfg):
press, hold = "service toggle", None
sw = cfg.get("service_windows") or []
always = len(sw) == 1 and set(sw[0].get("days", [])) == set(DAYS) and sw[0].get("start") == "00:00" and sw[0].get("end") in ("23:59", "00:00")
- note = ("Press: silence the alarms AND stop every auto-fix; press again to turn both back on. They come back by "
- "themselves %s. Holds still fix. The stream-audio failover still switches by itself on Sundays."
- % ("at midnight" if always else "when the alarm schedule next starts or ends"))
+ if any(t.get("kind") == "autofix" for t in cfg.get("tiles", [])):
+ note = ("Press: silence the alarms; press again to turn them back on. They come back by themselves %s. "
+ "Auto-fix is the AUTO-FIX key." % ("at midnight" if always else "when the alarm schedule next starts or ends"))
+ else:
+ note = ("Press: silence the alarms AND stop every auto-fix; press again to turn both back on. They come back by "
+ "themselves %s. Holds still fix. The stream-audio failover still switches by itself on Sundays."
+ % ("at midnight" if always else "when the alarm schedule next starts or ends"))
+ elif kind == "autofix":
+ stale_text = "'%s\\n?'" % label
+ press, hold = "autofix toggle", None
+ note = ("Press: pause auto-fix (alarms stay on); press again to turn it back on - or on outside its hours. It goes "
+ "back to its hours by itself when they next start or end. Holds always fix.")
+ elif kind == "bypass":
+ stale_text = "'%s\\n?'" % label
+ press, hold = "ack " + tid, "restart " + tid
+ b = tile.get("bypass") or {}
+ note = ("Hold %s s: hand %s to an operator - its MIDI links (%s) are taken out%s and the panel stops putting them "
+ "back. Hold again: everything is put back." % (secs, label, ", ".join(b.get("peers") or []),
+ " and Companion's %s switched off" % ", ".join(b["conns"]) if b.get("conns") else ""))
else:
stale_text = "'%s\\n?'" % label
press = "ack " + tid
diff --git a/schema.py b/schema.py
index 61634e8..7dfe06c 100644
--- a/schema.py
+++ b/schema.py
@@ -258,8 +258,12 @@ field("tiles[].label", "str", None, "The first line on the key.", "Keys", requir
field("tiles[].pos", "list", None, "[row, column]; absent = no key of its own (shown inside a group).", "Keys",
apply="pages")
field("tiles[].kind", "enum", None, "A special key: master (ALL), service (ALARMS), network (NET), ready (READY: "
- "opens the READY page and sums up its checklist).", "Keys", advanced=True,
- options=["master", "service", "network", "ready"], apply="pages")
+ "opens the READY page and sums up its checklist), autofix (AUTO-FIX: its own switch, so ALARMS only "
+ "silences), bypass (hold: hand a device to an operator - its MIDI links out, its Companion connections off).",
+ "Keys", advanced=True, options=["master", "service", "network", "ready", "autofix", "bypass"])
+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).",
+ "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")
field("tiles[].disabled", "bool", False, "Switched off: the key stays on the page (grey, OFF) but checks nothing "
@@ -491,7 +495,10 @@ GLOSSARY = {
"HOLD", "REC", "PLAYING", "OK", "OFF", "ON", "ALL OK", "WARN", "TRS", "GB", "NOT", "REACHABLE", "RECORDING",
"LEFT ALONE", "TO FIX", "TO CLEAR", "NO AUTO", "FIX", "AUTO", "IN", "COMPANION", "POWER", "CYCLE",
"STILL", "OFF IN", "DISABLED IN", "CONNECTING", "CHECK", "NETWORK", "NO", "PRESSING", "NO BUTTON THERE",
- "ON FOH BACKUP",
+ "ON FOH BACKUP", "BUSY", "TAKING MIDI OUT", "PUTTING MIDI BACK", "TAKING", "PUTTING", "MIDI OUT", "MIDI BACK",
+ ],
+ "a switch key (AUTO-FIX, or one like LIGHTS that hands a device to an operator)": [
+ "OPERATOR", "MIDI OFF", "OFF IN SETTINGS", "SETTINGS", "TIL",
],
}
@@ -583,7 +590,7 @@ def validate(cfg, live=None):
if errors:
return errors, warnings
ids, seen_pos = set(), {}
- kinds = {"master": 0, "service": 0, "network": 0}
+ kinds = {"master": 0, "service": 0, "network": 0, "autofix": 0}
fixes, hosts = cfg.get("fixes", {}), cfg.get("hosts", {})
for t in cfg["tiles"]:
tid = t.get("id", "")
@@ -659,6 +666,16 @@ def validate(cfg, live=None):
errors.append("there must be exactly one ALARMS key (kind service); found %d" % kinds["service"])
if kinds["network"] > 1:
errors.append("at most one NET key (kind network); found %d" % kinds["network"])
+ if kinds["autofix"] > 1:
+ errors.append("at most one AUTO-FIX key (kind autofix); found %d" % kinds["autofix"])
+ for t in cfg["tiles"]:
+ if t.get("kind") == "bypass":
+ b = t.get("bypass") if isinstance(t.get("bypass"), dict) else {}
+ if not b.get("peers"):
+ errors.append("key %s: a bypass key needs bypass.peers (the MIDI peers it takes out)" % t.get("id"))
+ for h in b.get("hosts") or []:
+ if h and h not in cfg.get("hosts", {}):
+ errors.append("key %s: bypass host %s is not a computer in the settings" % (t.get("id"), h))
for fid, fix in fixes.items():
where = "fix %s" % fid
kind = fix.get("kind", "ssh")
@@ -917,7 +934,9 @@ def phrases(cfg):
"failover": "switches by itself %s" % sp.windows_phrase(fo_windows) if fo_windows else "never switches by itself",
"live": ("changes wait while a HyperDeck is recording" + (" or %s" % sp.windows_phrase(site["live_services"])
if site.get("live_services") else "")),
- "alarms": "Alarms are on every day. The ALARMS key silences them and stops every auto-fix until midnight, then both come back by themselves.",
+ "alarms": ("Alarms are on every day. The ALARMS key silences them until midnight; AUTO-FIX pauses or resumes "
+ "auto-fix on its own. Both come back by themselves." if any(t.get("kind") == "autofix" for t in cfg.get("tiles", []))
+ else "Alarms are on every day. The ALARMS key silences them and stops every auto-fix until midnight, then both come back by themselves."),
"hold": "HOLD %gs" % (cfg.get("page_hold_ms", 2000) / 1000.0),
"ingest": site.get("ingest_name", "NA Ingest"),
}
diff --git a/settings/autoconf.py b/settings/autoconf.py
index 65619da..f1f8515 100644
--- a/settings/autoconf.py
+++ b/settings/autoconf.py
@@ -570,6 +570,19 @@ def keep_from(new, old):
cfg = copy.deepcopy(new)
have = referenced(cfg)
kept = []
+ # Switch keys made for this campus by hand (Southcoast's LIGHTS) and an AUTO-FIX key: auto-setup makes neither.
+ for t in old.get("tiles", []):
+ if t.get("kind") not in ("bypass", "autofix") or t["id"] in {x["id"] for x in cfg["tiles"]}:
+ continue
+ if t["kind"] == "autofix" and any(x.get("kind") == "autofix" for x in cfg["tiles"]):
+ continue
+ used = {tuple(x["pos"]) for x in cfg["tiles"] if x.get("pos")} | (
+ {tuple(cfg["restart_page"]["jump_pos"])} if cfg.get("restart_page") else set())
+ if t.get("pos") and tuple(t["pos"]) not in used:
+ cfg["tiles"].append(copy.deepcopy(t))
+ else:
+ place_new(cfg, {k: v for k, v in copy.deepcopy(t).items() if k != "pos"})
+ kept.append(t["label"])
# Computers in the old setup that did not answer this time, and the keys that check them.
gone = {a: h for a, h in (old.get("hosts") or {}).items() if not h.get("self") and a not in cfg["hosts"]}
for alias, h in gone.items():
diff --git a/settings/keyguide.py b/settings/keyguide.py
index aeefe61..f364343 100644
--- a/settings/keyguide.py
+++ b/settings/keyguide.py
@@ -163,6 +163,14 @@ def key_guide(cfg):
parts.append("
Shows the worst of: %s. Hold: %s.
" % (", ".join(e(tiles[m]["label"]) for m in t["members"]), hold_words(t, cfg)))
if t.get("kind") == "service":
parts.append("
%s
" % e(ph["alarms"]))
+ if t.get("kind") == "autofix":
+ parts.append("
Press: pause auto-fix, or run it outside its hours (auto-fix %s). It goes back to its hours by itself "
+ "when they next start or end. Holds always fix.
" % e(ph["auto_fix"]))
+ if t.get("kind") == "bypass":
+ b = t.get("bypass") or {}
+ parts.append("
Hold: hand it to an operator - the MIDI links to %s are taken out%s, and the panel stops putting them "
+ "back. Hold again: everything is put back as it was.
" % (
+ e(", ".join(b.get("peers") or [])), (" and Companion's %s is switched off" % e(", ".join(b["conns"]))) if b.get("conns") else ""))
if t.get("kind") == "network":
n = cfg.get("network", {})
parts.append("
Watches this Mac's network port%s, and calls it a blip when %d connections from %d groups drop within %d s "
diff --git a/settings/server.py b/settings/server.py
index b69e8d0..cce69c5 100644
--- a/settings/server.py
+++ b/settings/server.py
@@ -276,6 +276,7 @@ class Site:
"live": live, "running": running, "service_on": snap.get("service_on") if fresh else None,
"alarms_override": snap.get("alarms_override") if fresh else None,
"auto_fix": {"phrase": schema.phrases(cfg)["auto_fix"], "state": snap.get("auto_fix")},
+ "bypass": (snap.get("bypass") or {}) if fresh else {},
"problems": sorted(problems, key=lambda p: (p["level"] != "err", p["since"] or 0)), "needs": needs,
"computers": self.computer_lines(snap), "pages": pages, "snapshot_age": (now - snap["time"]) if snap.get("time") else None,
"last_change": self.last_change(), "phrases": schema.phrases(cfg), "version": started.get("version"),
@@ -1030,7 +1031,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
if path == "/api/command":
text = str(body.get("text", "")).strip()
verb = text.split(" ")[0].lower() if text else ""
- if verb not in ("test", "resurvey", "service"):
+ if verb not in ("test", "resurvey", "service", "autofix"):
return self.send(400, {"error": "not a command the page may send"})
if verb == "test":
live, reason = site.applier.live()
diff --git a/settings/ui/app.js b/settings/ui/app.js
index 4a5e3c4..50b63ec 100644
--- a/settings/ui/app.js
+++ b/settings/ui/app.js
@@ -272,6 +272,7 @@
const live = isLive ? `
@@ -297,6 +301,7 @@
main.querySelector("#restartwd").onclick = () => confirmBox(w.running ? "Restart the watchdog? The deck goes purple for about 5 seconds; a fix that is running finishes first." : "Start the watchdog?", w.running ? "Restart" : "Start", async () => { try { const r = await post(w.running ? "/api/watchdog/restart" : "/api/watchdog/start"); alertBox(r.ok ? "Done: " + r.detail : "Not done: " + r.detail); } catch (e) { alertBox(e.message); } });
const stop = main.querySelector("#stopwd"); if (stop) stop.onclick = () => confirmBox("Stop the watchdog? The deck goes purple and every alarm is off until it is started again.", "Stop", async () => { try { const r = await post("/api/watchdog/stop", { word: "STOP" }); alertBox(r.ok ? "Stopped." : "Not stopped: " + r.detail); } catch (e) { alertBox(e.message); } }, "STOP");
main.querySelector("#alarms").onclick = async () => { try { await post("/api/command", { text: o.service_on === false ? "service on" : "service off" }); setTimeout(() => overview(_, true), 1500); } catch (e) { alertBox(e.message); } };
+ const afb = main.querySelector("#autofixbtn"); if (afb) afb.onclick = async () => { try { await post("/api/command", { text: af.switch ? "autofix off" : "autofix on" }); setTimeout(() => overview(_, true), 1500); } catch (e) { alertBox(e.message); } };
for (const b of main.querySelectorAll("[data-run]")) b.onclick = () => runNow(...b.dataset.run.split("|"));
if (!quiet) window.scrollTo(0, 0);
}
diff --git a/sites/_template/config.json b/sites/_template/config.json
index 4a9a807..9259a56 100644
--- a/sites/_template/config.json
+++ b/sites/_template/config.json
@@ -48,6 +48,15 @@
],
"kind": "service"
},
+ {
+ "id": "autofix",
+ "label": "AUTO-FIX",
+ "pos": [
+ 3,
+ 1
+ ],
+ "kind": "autofix"
+ },
{
"id": "net",
"label": "NET",
diff --git a/sites/sc/config.json b/sites/sc/config.json
index e9dcd10..6d7507f 100644
--- a/sites/sc/config.json
+++ b/sites/sc/config.json
@@ -63,7 +63,8 @@
"propresenter_ingest",
"prop_port_propresenter_api",
"hyperdeck",
- "midi_reconnect"
+ "midi_reconnect",
+ "ableton_sc_tracks_new"
]
},
"tiles": [
@@ -216,6 +217,55 @@
1,
1
]
+ },
+ {
+ "id": "ableton",
+ "label": "ABLETON",
+ "checks": [
+ {
+ "type": "remote",
+ "template": "mac_program",
+ "params": {
+ "host": "sc_tracks_new",
+ "process": "Live"
+ },
+ "msg": "ABLETON NOT RUNNING",
+ "fix": "ableton_sc_tracks_new",
+ "host": "sc_tracks_new",
+ "cmd": "pgrep -xq Live"
+ }
+ ],
+ "pos": [
+ 1,
+ 2
+ ]
+ },
+ {
+ "id": "autofix",
+ "label": "AUTO-FIX",
+ "kind": "autofix",
+ "pos": [
+ 3,
+ 1
+ ]
+ },
+ {
+ "id": "lights",
+ "label": "LIGHTS",
+ "kind": "bypass",
+ "pos": [
+ 1,
+ 3
+ ],
+ "bypass": {
+ "peers": [
+ "Lighting"
+ ],
+ "hosts": [
+ "",
+ "sc_tracks_new"
+ ]
+ }
}
],
"fixes": {
@@ -271,6 +321,34 @@
"params": {},
"short": "RECONNECT",
"label": "reconnect the peer to the network MIDI session it dropped out of (like Connect in Audio MIDI Setup), at its wired address"
+ },
+ "ableton_sc_tracks_new": {
+ "template": "ableton",
+ "params": {
+ "host": "sc_tracks_new",
+ "app": "Ableton Live 11 Suite",
+ "restart": false,
+ "wait_seconds": 15
+ },
+ "label": "start Ableton on SC Tracks NEW if it is not running",
+ "short": "ABLETON",
+ "kind": "ssh",
+ "host": "sc_tracks_new",
+ "run": "pgrep -xq Live || { L=$(ls -td ~/Library/Preferences/Ableton/Live*/ | head -1); S=$(grep -a 'info: Loading document \"' \"${L}Log.txt\" | grep -v '/Templates/' | tail -1 | sed -E 's/^.*Loading document \"(.*)\"[^\"]*$/\\1/'); if [ -n \"$S\" ] && [ -f \"$S\" ]; then open -a 'Ableton Live 11 Suite' \"$S\"; else open -a 'Ableton Live 11 Suite'; fi; }"
+ },
+ "restart_ableton_sc_tracks_new": {
+ "template": "ableton",
+ "params": {
+ "host": "sc_tracks_new",
+ "app": "Ableton Live 11 Suite",
+ "restart": true,
+ "wait_seconds": 15
+ },
+ "label": "restart Ableton on SC Tracks NEW (closes it first)",
+ "short": "ABLETON",
+ "kind": "ssh",
+ "host": "sc_tracks_new",
+ "run": "L=$(ls -td ~/Library/Preferences/Ableton/Live*/ | head -1); S=$(grep -a 'info: Loading document \"' \"${L}Log.txt\" | grep -v '/Templates/' | tail -1 | sed -E 's/^.*Loading document \"(.*)\"[^\"]*$/\\1/'); pkill -TERM -x Live; i=0; while pgrep -xq Live && [ $i -lt 30 ]; do sleep 0.5; i=$((i+1)); done; pgrep -xq Live && pkill -KILL -x Live; sleep 1; if [ -n \"$S\" ] && [ -f \"$S\" ]; then open -a 'Ableton Live 11 Suite' \"$S\"; else open -a 'Ableton Live 11 Suite'; fi"
}
},
"hosts": {
@@ -298,6 +376,17 @@
"http": "http://127.0.0.1:8000/api/connections"
}
]
+ },
+ "sc_tracks_new": {
+ "name": "SC Tracks NEW",
+ "os": "mac",
+ "ssh": "tracks@10.5.3.72",
+ "role": "found by auto-setup",
+ "runs": [
+ {
+ "kind": "sshd"
+ }
+ ]
}
},
"gear": {},
@@ -338,6 +427,15 @@
0,
1
]
+ },
+ {
+ "id": "rs_ableton",
+ "label": "ABLETON",
+ "fix": "restart_ableton_sc_tracks_new",
+ "pos": [
+ 0,
+ 2
+ ]
}
]
},
diff --git a/sites/ws/config.json b/sites/ws/config.json
index 2735a37..ec37864 100644
--- a/sites/ws/config.json
+++ b/sites/ws/config.json
@@ -575,6 +575,15 @@
1,
3
]
+ },
+ {
+ "id": "autofix",
+ "label": "AUTO-FIX",
+ "kind": "autofix",
+ "pos": [
+ 3,
+ 1
+ ]
}
],
"timing": {
diff --git a/status_panel.py b/status_panel.py
index a176384..73862ae 100644
--- a/status_panel.py
+++ b/status_panel.py
@@ -494,6 +494,8 @@ def check_midi_peer(chk, snap, ctx):
sessions = None # that Mac stopped reporting: no verdict
if sessions is None:
return [("peer", Result(OK))]
+ if chk["peer"] in ctx.get("bypassed", ()) or chk["session"] in ctx.get("bypassed", ()):
+ return [("peer", Result(OK))] # an operator has it (LIGHTS key): taken out on purpose
sess = sessions.get(chk["session"])
if sess is None:
return [("peer", Result(ERR, "NO SESSION"))]
@@ -508,6 +510,101 @@ def check_midi_peer(chk, snap, ctx):
return [("peer", Result(ERR, "MIDI LOST"))]
+def midi_on(cfg, host, call, jxa):
+ """Run a coremidi call on another Mac (python there, else the JavaScript for Automation `jxa`):
+ (result, error text)."""
+ r = subprocess.run(ssh_cmd(cfg, host) + remote_python("-"), input=remote_midi_script(call),
+ capture_output=True, text=True, timeout=30)
+ if r.returncode == 97: # no python on that Mac
+ r = subprocess.run(ssh_cmd(cfg, host) + ["/usr/bin/osascript -l JavaScript -"], input=jxa,
+ capture_output=True, text=True, timeout=30)
+ result = parse_remote_midi(r.stdout) if r.returncode == 0 else None
+ return result, (None if result is not None else (r.stdout + r.stderr).strip().replace("\n", " ")[-200:])
+
+
+def midi_peer(cfg, host, action, session, peer, address=None, repoint=False):
+ """Connect ("connect") or drop ("disconnect") one peer of a network MIDI session, on this Mac
+ (host "") or another: (result word or None, error text)."""
+ import coremidi
+ if action == "connect" and not str(address or "").rsplit(":", 1)[0].strip("[]"):
+ # A peer that called in keeps no address (":5004"): adding it back connects the session to
+ # itself (SC Tracks, 2026-10-21). It comes back when it calls again.
+ return None, "no address to call (%r): it reconnects when %s calls in" % (address, peer)
+ if not host:
+ if action == "connect":
+ return coremidi.connect_peer(session, peer, address, repoint=repoint), None
+ return coremidi.disconnect_peer(session, peer), None
+ if action == "connect":
+ return midi_on(cfg, host, "ns['connect_peer'](%r, %r, %r, repoint=%r)" % (session, peer, address, repoint),
+ coremidi.jxa_connect(session, peer, address, repoint=repoint))
+ return midi_on(cfg, host, "ns['disconnect_peer'](%r, %r)" % (session, peer), coremidi.jxa_disconnect(session, peer))
+
+
+def midi_sessions(cfg, host):
+ """The network MIDI sessions on this Mac (host "") or another, or None."""
+ import coremidi
+ if not host:
+ return coremidi.sessions()
+ return midi_on(cfg, host, "ns['sessions']()", coremidi.JXA_SESSIONS)[0]
+
+
+def bypass_apply(cfg, tile, on, entry, comp, names_ips, lock=None):
+ """Hand a bypass key's device to an operator, or take it back. `entry["dropped"]` keeps every peer
+ taken out ({host, session, peer, address}), so taking it back puts exactly those back; the MIDI
+ checks naming it are reconnected too. Its Companion connections are switched off, then on again."""
+ b = tile.get("bypass") or {}
+ names, ips = set(b.get("peers") or []), set(names_ips[1]) if on else set()
+ 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:
+ 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 []:
+ conn = conns.get(label)
+ if not conn:
+ log.info("BYPASS %s: Companion has no connection %s", tile["label"], label)
+ elif on and conn.get("enabled"):
+ log.info("BYPASS %s: Companion %s off -> %s", tile["label"], label, comp.disable_connection(conn["id"]))
+ elif not on and not conn.get("enabled"):
+ log.info("BYPASS %s: Companion %s on -> %s", tile["label"], label, comp.enable_connection(conn["id"]))
+ if on:
+ for host in hosts:
+ try:
+ sessions = midi_sessions(cfg, host)
+ except Exception as e:
+ sessions, err = None, str(e)
+ if sessions is None:
+ continue # that Mac is off or unreachable: tried again in 10 s
+ for sname, sess in sessions.items():
+ for p in sess.get("peers") or []:
+ ip = peer_ip(p.get("address"))
+ if p.get("name") in names or (ip and str(ip) in ips):
+ res, err = midi_peer(cfg, host, "disconnect", sname, p["name"])
+ log.info("BYPASS %s: %s out of %s%s -> %s", tile["label"], p["name"], sname,
+ " on " + host if host else "", res or err)
+ if res:
+ rec = {"host": host, "session": sname, "peer": p["name"], "address": p.get("address") or ""}
+ with (lock or threading.Lock()):
+ if rec not in entry["dropped"]:
+ entry["dropped"].append(rec)
+ return
+ done = set()
+ for rec in entry.get("dropped") or []:
+ res, err = midi_peer(cfg, rec["host"], "connect", rec["session"], rec["peer"], rec["address"])
+ done.add((rec["host"], rec["session"], rec["peer"]))
+ log.info("BYPASS %s: %s back into %s%s -> %s", tile["label"], rec["peer"], rec["session"],
+ " on " + rec["host"] if rec["host"] else "", res or err)
+ 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", ""), c["session"], c["peer"]) not in done:
+ res, err = midi_peer(cfg, c.get("host", ""), "connect", c["session"], c["peer"], c["address"])
+ log.info("BYPASS %s: %s back into %s -> %s", tile["label"], c["peer"], c["session"], res or err)
+
+
def peer_ip(address):
"""The IPv4 address of a MIDI peer ("10.1.1.1:5040"), or None for a Bonjour name or IPv6."""
host = str(address or "").strip().rsplit(":", 1)[0]
@@ -967,10 +1064,11 @@ class ServiceMode:
edge is worked out at the press and kept, so the override ends there even when the watchdog
was not running at that minute (a restart over 23:59 left the alarms muted for good)."""
- def __init__(self, windows, state):
+ def __init__(self, windows, state, name="service"):
self.windows = windows
+ self.name = name # "service" (ALARMS) or "autofix" (AUTO-FIX)
self.state = state if isinstance(state, State) else State(state)
- saved = self.state.get("service") or {}
+ saved = self.state.get(name) or {}
self.override, self.last_sched = saved.get("override"), saved.get("last_sched")
self.edge = saved.get("edge")
@@ -1027,7 +1125,18 @@ class ServiceMode:
self._save()
def _save(self):
- self.state.set("service", {"override": self.override, "last_sched": self.last_sched, "edge": self.edge})
+ self.state.set(self.name, {"override": self.override, "last_sched": self.last_sched, "edge": self.edge})
+
+ def until(self, now):
+ """When the state shown now next changes by itself (an override's end, else the schedule's next edge),
+ looked up once a minute."""
+ minute = int(now // 60)
+ cached = getattr(self, "_until", None)
+ if cached and cached[0] == (minute, self.override):
+ return cached[1]
+ found = self.back_at(now) if self.override is not None else self.next_change(now)
+ self._until = ((minute, self.override), found)
+ return found
# --- Engine ------------------------------------------------------------------
@@ -1050,6 +1159,16 @@ class Engine:
self.hold_ms = cfg.get("page_hold_ms", 2000) # how long a key is held to run its fix
self.state = State(state_path)
self.service = ServiceMode(cfg.get("service_windows", []), self.state)
+ # With an AUTO-FIX key, ALARMS and auto-fix are two switches (2026-10-21, the user: "seperatly
+ # silence alarms and auto fixes"); without one, ALARMS stops both, as before.
+ self.split = any(t.get("kind") == "autofix" for t in cfg["tiles"])
+ self.autofix = ServiceMode((cfg.get("auto_fix") or {}).get("windows", []), self.state, "autofix")
+ # Keys that hand a device to an operator (Southcoast's LIGHTS: a lighting op runs the desk, so
+ # nothing may send it MIDI): tile id -> {"since", "dropped": [peers taken out, to put back]}.
+ self.bypass = {k: v for k, v in (self.state.get("bypass") or {}).items() if k in self.tiles}
+ self.bypass_lock = threading.Lock()
+ self.bypass_busy = False
+ self.bypass_last = 0.0
self.data_dir = data_dir
self.stopping = False # SIGTERM arrived: finish what runs, start nothing new
self.running = {} # fix thread name -> {"key", "what", "since"} (published as fixes-running.json)
@@ -1184,7 +1303,8 @@ class Engine:
# raw results for every check key
def evaluate(self, snap, now, service_on):
ctx = {"now": now, "service": service_on, "any_recording": False, "dante_all": {},
- "ready_due": in_windows(self.ready_cfg.get("due", []), now), "ready_done": self.ready_done(now)}
+ "ready_due": in_windows(self.ready_cfg.get("due", []), now), "ready_done": self.ready_done(now),
+ "bypassed": self.bypass_names()[0]}
for t in all_tiles(self.cfg):
for chk in t.get("checks", []):
if chk["type"] == "dante":
@@ -1500,11 +1620,77 @@ class Engine:
"""Show `text` on the button for a while (hold feedback), instead of its usual text."""
self.notes[tile_id] = {"text": text, "level": level, "until": now + seconds}
+ # --- operator bypass (a LIGHTS key) ---------------------------------------------------------
+ def bypass_names(self):
+ """(peer and session names an operator has, their IPv4 addresses as the MIDI checks know them)."""
+ names, ips = set(), set()
+ for tid in self.bypass:
+ names |= set(((self.tiles.get(tid) or {}).get("bypass") or {}).get("peers") or [])
+ if names:
+ for t in self.cfg["tiles"]:
+ for c in t.get("checks", []):
+ if c.get("type") == "midi_peer" and (c.get("peer") in names):
+ ip = peer_ip(c.get("address"))
+ if ip:
+ ips.add(str(ip))
+ return names, ips
+
+ def toggle_bypass(self, tile_id, comp, now):
+ """A hold on a bypass key: hand it to an operator (take its MIDI links out, switch its Companion
+ connections off), or take it back (put everything back, as it was)."""
+ if self.bypass_busy:
+ self.note(tile_id, "BUSY\n" + self.wait_text, "idle", now, 4)
+ return
+ tile = self.tiles[tile_id]
+ with self.bypass_lock:
+ if tile_id in self.bypass:
+ entry = self.bypass.pop(tile_id)
+ on = False
+ else:
+ entry = self.bypass[tile_id] = {"since": now, "dropped": []}
+ on = True
+ self.state.set("bypass", self.bypass)
+ log.info("BYPASS %s: %s", tile["label"], "handed to an operator" if on else "back to the panel")
+ self.note(tile_id, "TAKING\nMIDI OUT" if on else "PUTTING\nMIDI BACK", "fixing", now, 6)
+ self.bypass_last = now
+ self.start_bypass_job(tile_id, on, entry, comp)
+
+ def start_bypass_job(self, tile_id, on, entry, comp):
+ self.bypass_busy = True
+
+ def job():
+ try:
+ bypass_apply(self.cfg, self.tiles[tile_id], on, entry, comp, self.bypass_names(), self.bypass_lock)
+ if tile_id in self.bypass:
+ self.state.set("bypass", self.bypass) # the peers taken out, to put back later
+ except Exception:
+ log.exception("bypass %s failed", tile_id)
+ finally:
+ self.bypass_busy = False
+ 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:
+ return
+ self.bypass_last = now
+ tid = next(iter(self.bypass))
+ self.start_bypass_job(tid, True, self.bypass[tid], comp)
+
def run_fixes(self, tile_id, snap, comp, now, cooldown=None):
cooldown = self.cooldown if cooldown is None else cooldown
if self.stopping:
self.note(tile_id, "RESTARTING\nSOON", "idle", now, 4)
return []
+ kind = self.tiles.get(tile_id, {}).get("kind")
+ if kind in ("service", "autofix"): # from North Attleboro's CAMPUS page: the hold is the press
+ mode = self.service if kind == "service" else self.autofix
+ mode.set(not mode.on(now), now)
+ log.info("%s %s (hold)", "SERVICE" if kind == "service" else "AUTO-FIX", "ON" if mode.on(now) else "OFF")
+ return []
+ if kind == "bypass":
+ self.toggle_bypass(tile_id, comp, now)
+ return []
if self.tiles.get(tile_id, {}).get("kind") == "network":
# NET: a hold clears a remembered network blip (link and gateway clear themselves).
if self.burst is None:
@@ -1631,17 +1817,23 @@ class Engine:
if not a or now - self.auto_last < 5 or self.stopping:
return []
self.auto_last = now
- # The ALARMS key is the on-the-fly switch: muted, nothing is fixed by itself at all (holds
- # still work). It comes back by itself with the alarms (2026-10-19: the user's choice).
- muted = not self.service.on(now)
- hours = in_windows(a.get("windows", []), now)
+ if self.split:
+ # The AUTO-FIX key: paused, nothing is fixed by itself (holds still work); pressed outside
+ # the hours, the gentle fixes run until the hours next start or end.
+ hours = self.autofix.on(now)
+ muted = self.autofix.override is False
+ else:
+ # The ALARMS key is the on-the-fly switch: muted, nothing is fixed by itself at all (holds
+ # still work). It comes back by itself with the alarms (2026-10-19: the user's choice).
+ muted = not self.service.on(now)
+ hours = in_windows(a.get("windows", []), now)
# A fix marked "anytime" (writing the timecode into Planning Center) runs by itself at any
# hour, not only in the auto-fix hours - but never while muted, like every other.
anytime = {f for f in a.get("fixes", []) if self.cfg["fixes"].get(f, {}).get("anytime")
and not self.cfg["fixes"].get(f, {}).get("never_auto") and not self.cfg["fixes"].get(f, {}).get("disabled")}
if muted:
if self.autolog and (self.auto_on or (anytime and not self.auto_muted)):
- self.autolog.write(now, "PANEL", "auto-fix off (ALARMS muted)")
+ self.autolog.write(now, "PANEL", "auto-fix off (%s)" % ("AUTO-FIX key" if self.split else "ALARMS muted"))
self.auto_on, self.auto_muted = False, True
self.auto_tries.clear()
self.auto_seen.clear()
@@ -2070,12 +2262,30 @@ def render(engine, service_on):
# they come back by themselves after a press.
word = "ON" if service_on else "OFF"
fix_word = "OFF" if (not service_on or (engine.cfg.get("auto_fix") or {}).get("disabled")) else "ON"
- text = "%s %s\nAUTO-FIX %s" % (tile["label"], word, fix_word)
+ text = ("%s\n%s" % (tile["label"], word)) if engine.split else "%s %s\nAUTO-FIX %s" % (tile["label"], word, fix_word)
back = engine.service.back_at(engine.now)
if back:
text += "\nTIL " + clock12(back)
out[tid] = colored(text, "service" if service_on else "off")
continue
+ if kind == "autofix":
+ # Its own switch: on in the auto-fix hours, a press pauses it (or runs it outside them).
+ if (engine.cfg.get("auto_fix") or {}).get("disabled"):
+ out[tid] = colored(tile["label"] + "\nOFF IN\nSETTINGS", "off")
+ continue
+ on = engine.autofix.on(engine.now)
+ until = engine.autofix.until(engine.now)
+ text = "%s\n%s" % (tile["label"], "ON" if on else "OFF")
+ if until and (on or engine.autofix.override is not None):
+ same_day = datetime.fromtimestamp(until).date() == datetime.fromtimestamp(engine.now).date()
+ text += "\nTIL " + ("" if same_day else datetime.fromtimestamp(until).strftime("%a ").upper()) + clock12(until)
+ out[tid] = colored(text, "service" if on else "off")
+ continue
+ if kind == "bypass":
+ # Handed to an operator: steady blue, says so; otherwise the panel runs it.
+ on = tid in engine.bypass
+ out[tid] = colored("%s\n%s" % (tile["label"], "OPERATOR\nMIDI OFF" if on else "AUTO"), "service" if on else "ok")
+ 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
@@ -2856,24 +3066,11 @@ def execute_fix(cfg, fix_id, chk):
return propresenter_port(cfg, fix, label)
if kind == "midi_connect":
# Like Connect in Audio MIDI Setup: add the peer back to the session it dropped out of.
- args = (chk["session"], chk["peer"], chk["address"])
- repoint = bool(chk.get("repoint"))
- if chk.get("host"):
- r = subprocess.run(ssh_cmd(cfg, chk["host"]) + remote_python("-"),
- input=remote_midi_script("ns['connect_peer'](%r, %r, %r, repoint=%r)" % (args + (repoint,))),
- capture_output=True, text=True, timeout=30)
- if r.returncode == 97: # no python on that Mac: the same, in JavaScript for Automation
- import coremidi
- r = subprocess.run(ssh_cmd(cfg, chk["host"]) + ["/usr/bin/osascript -l JavaScript -"],
- input=coremidi.jxa_connect(*args, repoint=repoint), capture_output=True, text=True, timeout=30)
- result = parse_remote_midi(r.stdout) if r.returncode == 0 else None
- if result is None:
- log.info("FIX %s -> %s on %s failed: %s", label, chk["peer"], chk["host"],
- (r.stdout + r.stderr).strip().replace("\n", " ")[-200:])
- return False
- else:
- import coremidi
- result = coremidi.connect_peer(*args, repoint=repoint)
+ result, err = midi_peer(cfg, chk.get("host", ""), "connect", chk["session"], chk["peer"], chk["address"],
+ repoint=bool(chk.get("repoint")))
+ if result is None:
+ log.info("FIX %s -> %s on %s failed: %s", label, chk["peer"], chk.get("host"), err)
+ return False
log.info("FIX %s -> %s at %s, session %s%s: %s", label, chk["peer"], chk["address"], chk["session"],
(" on " + chk["host"]) if chk.get("host") else "", result)
return result in ("asked", "already", "kicked", "moved")
@@ -3401,7 +3598,9 @@ def snapshot_summary(snap, engine, now):
for name, l in snap.listeners.items()},
"service_on": engine.service.on(now),
"alarms_override": engine.service.override,
- "auto_fix": {"on": engine.auto_on, "paused": engine.auto_paused},
+ "auto_fix": {"on": engine.auto_on, "paused": engine.auto_paused, "split": engine.split,
+ "switch": engine.autofix.on(now), "override": engine.autofix.override},
+ "bypass": {tid: {"since": b.get("since")} for tid, b in engine.bypass.items()},
"failover": dict(engine.failed),
"burst": engine.burst and {"start": engine.burst["start"], "conns": sorted(engine.burst["conns"]),
"cleared": engine.burst["cleared"]},
@@ -3446,6 +3645,10 @@ def handle_commands(text, engine, comp, snap, now):
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 == "autofix":
+ target = {"on": True, "off": False}.get(arg, not engine.autofix.on(now))
+ engine.autofix.set(target, now)
+ log.info("AUTO-FIX %s (button)", "ON" if target else "OFF")
elif verb in ("restart", "force"):
# Holds carry Companion's clock ("force rs_x 1791496225"). One pressed while the watchdog
# was down must not fire minutes later in a burst with the others.
@@ -3540,6 +3743,10 @@ def run(cfg, data_dir, dry_run=False, stop=None, config_sha=None):
engine.auto_fix(snap, comp, now)
except Exception: # a bug here must never take the panel down
log.exception("auto-fix failed; the panel carries on without it this round")
+ try:
+ engine.bypass_tick(comp, now)
+ except Exception:
+ log.exception("operator bypass failed; the panel carries on without it this round")
try:
engine.failover(snap, comp, now)
except Exception:
diff --git a/tests/test_campus.py b/tests/test_campus.py
index 5eba74c..b1d92ac 100644
--- a/tests/test_campus.py
+++ b/tests/test_campus.py
@@ -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)
diff --git a/tests/test_setup.py b/tests/test_setup.py
index fe97b75..d798036 100644
--- a/tests/test_setup.py
+++ b/tests/test_setup.py
@@ -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))
diff --git a/tests/test_status_panel.py b/tests/test_status_panel.py
index 0c16eed..803174b 100644
--- a/tests/test_status_panel.py
+++ b/tests/test_status_panel.py
@@ -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))