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

Wiring view: every machine and what connects it (Companion, MIDI, Dante, watchdog, web, Art-Net), hide types or single links, scripts per computer

Browse Source 
RESI LIVE NOT REC follows the Resi auto-record service (custom autorec_status, not during its REC hold; stale heartbeat = no verdict);
INGEST watches AUTO REC (running, heartbeat, Resi reachable), hold restarts it. Probes list each computer's scripts by file name only.
HyperDeck addresses read from bonjourHost.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-27 13:23:02 -04:00
co-authored by Claude Opus 5.5
parent 6c78ed3221
commit 26804e4ce2
11 changed files with 655 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_settings.py
+53
View File
@@ -102,6 +102,40 @@ class TestServer(unittest.TestCase):
self.assertEqual(web["light"], "ok") # from snapshot.json, not a probe
self.assertIn("mac", o["setup_steps"])
def test_wiring_shows_machines_their_connections_and_scripts(self):
# 2026-10-27: "like a wiring diagram ... how many and what scripts are running on each machine"
live = {"tracks": {"alias": "tracks", "reachable": True, "time": time.time(), "items": [],
"found": {"scripts": [{"name": "statuspanel.py", "pid": 5, "by": "", "running": True}]}}}
json.dump(live, open(os.path.join(self.data, "machines-live.json"), "w"))
snap = json.load(open(os.path.join(self.data, "snapshot.json")))
snap.update({"conns": {"ATEM": {"category": "good", "enabled": True}, "x32": {"category": "error", "enabled": True},
"HyperDeck_1": {"category": "good", "enabled": True}, "tcp-HyperDeck-1": {"category": "good", "enabled": True}},
"conn_hosts": {"ATEM": "10.1.10.2", "x32": "10.1.1.112", "HyperDeck_1": "10.1.10.11", "tcp-HyperDeck-1": "10.1.10.11"},
"remote_midi": {"tracks": {"NA Tracks": {"peers": [{"address": "10.1.13.140:5004", "name": "SuperRack-LiveBox", "state": "running"}]}}},
"dante_map": {"NA-Tracks": {"switch": "10.66.1.100", "port": "0/6", "seen": time.time()},
"FOH-SuperRack-LiveBox": {"switch": "10.66.1.102", "port": "0/9", "seen": time.time()},
"STUDIO-B-MON": {"switch": "10.66.1.100", "port": "0/2", "seen": time.time()}}})
path = os.path.join(self.data, "snapshot.json")
old = open(path).read()
try:
json.dump(snap, open(path, "w"))
w = json.loads(get(self.port, "/api/wiring")[1])
finally:
open(path, "w").write(old)
os.remove(os.path.join(self.data, "machines-live.json"))
nodes = {n["id"]: n for n in w["nodes"]}
edges = [(e["type"], e["from"], e["to"], e["light"]) for e in w["edges"]]
self.assertEqual(w["me"], "h:ingest")
self.assertEqual(nodes["h:tracks"]["scripts"][0]["name"], "statuspanel.py")
self.assertIsNone(nodes["h:prop"]["scripts"]) # not looked yet
self.assertIn(("companion", "h:ingest", "d:10.1.1.112", "problem"), edges)
self.assertEqual(sorted(nodes["d:10.1.10.11"]["names"]), ["HyperDeck_1", "tcp-HyperDeck-1"]) # one deck, one box
self.assertIn(("midi", "h:tracks", "d:10.1.13.140", "ok"), edges)
self.assertIn(("dante", "h:tracks", "sw:10.66.1.100", "ok"), edges) # NA-Tracks is the Tracks Mac
self.assertIn("FOH-SuperRack-LiveBox", nodes["d:10.1.13.140"]["names"]) # same box as its MIDI peer
self.assertIn("d:studiobmon", nodes) # a monitor, not the Studio B Mac
self.assertEqual({t["id"] for t in w["types"]} >= {"companion", "midi", "watch", "dante"}, True)
def test_gear_and_keys_and_pages(self):
gear = json.loads(get(self.port, "/api/gear")[1])
self.assertTrue(any(r["name"] == "HyperDeck_2" and "hd2" in r["watched_by"] for r in gear))
@@ -362,6 +396,25 @@ class TestBindingAndProbes(unittest.TestCase):
self.assertEqual(by["sshd"]["light"], "ok")
self.assertEqual((setup["profile"], setup["wake_on_lan"], setup["firewall_allow"]), ("Public", "Enabled", "10.1.1.1,10.6.3.222"))
self.assertIn("\\StatusPanel\\Start Arena", found["tasks"])
scripts = {x["name"]: x["running"] for x in found["scripts"]}
self.assertEqual((scripts["Start Vista"], scripts["Start rtpMIDI"]), (True, False)) # Vista.exe runs; rtpMIDI.exe does not
def test_scripts_are_listed_by_file_name_only(self):
agents = {"org.waterschurch.prop-guard": 412, "org.waterschurch.status-panel": None, "com.apple.x": 9}
lines = ["412 /usr/bin/python3 /Users/m/Library/Application Support/status-panel/prop_guard.py /Users/m/cfg --token=SECRET",
"77 /bin/bash /Users/m/backup/backup-campuses.sh", "55 /Applications/Companion.app/Contents/Resources/node main.js",
"60 node /Users/m/Library/Application Support/companion/modules/x/main.js", "9 zsh -c echo SP_SCRIPTS",
"88 /Applications/Xcode.app/Contents/Developer/Library/Frameworks/Python3.framework/Versions/3.9/Resources/"
"Python.app/Contents/MacOS/Python /Users/m/Library/Application Support/resi-autorec/resi_autorec.py"]
got = probes.parse_scripts(lines, agents)
self.assertEqual([(x["name"], x["by"], x["running"]) for x in got],
[("backup-campuses.sh", "", True), ("prop_guard.py", "org.waterschurch.prop-guard", True),
("resi_autorec.py", "", True),
("status-panel", "org.waterschurch.status-panel", False)])
self.assertNotIn("SECRET", json.dumps(got)) # never the rest of the line
win = probes.parse_scripts(['812 powershell.exe -File "C:\\ProgramData\\StatusPanel\\start-vista.ps1"'],
tasks={"\\StatusPanel\\Start Vista": "Running", "\\StatusPanel\\Start rtpMIDI": "Ready"})
self.assertEqual([(x["name"], x["running"]) for x in win], [("Start Vista", True), ("start-vista.ps1", True), ("Start rtpMIDI", False)])
def test_probe_commands_never_carry_names_from_the_settings(self):
cfg = json.loads(json.dumps(CONFIG))