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

Find Dante PoE devices by MAC automatically instead of hard-coding ports.

Browse Source 
Every 10 minutes the watchdog maps each Dante device: name and IP from
Companion, MAC from NA Ingest's ARP table, switch port and PoE state from
the Netgear switches in dante_switches, saved to dante-map.json. A DANTE hold
on a missing PoE-powered device power-cycles the port it is on now (found
live by MAC, or its last known port if it has aged out and nothing else is
there); never a port with more than 2 devices. Re-patching, adding or
renaming devices needs no config change. First live map: 19 devices, the 4
Studio B boxes on PoE.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-08 13:31:51 -04:00
co-authored by Claude Opus 5.5
parent c8b9216ea5
commit d06798adf0
5 changed files with 275 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_status_panel.py
+78 -21
View File
@@ -538,28 +538,51 @@ class TestFixes(unittest.TestCase):
tiles = self.engine.step(self.snap, time.time())
self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nFIX FAILED\nDVS", "failed"))
def test_studio_b_poe_boxes_are_power_cycled(self):
def test_dante_hold_power_cycles_any_poe_device_from_the_map(self):
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
devices.remove("STUDIO-B-MON") # PoE box (per the map)
devices.remove("BOH-WING") # console: not PoE
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
self.snap.dante_map = {
"STUDIO-B-MON": {"mac": "00:1d:c1:51:18:84", "switch": "10.66.1.102", "port": "0/15", "shared": 1, "poe": True},
"BOH-WING": {"mac": "00:1d:c1:03:ec:3a", "switch": "10.66.1.102", "port": "0/3", "shared": 1, "poe": False},
}
self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller"), ("poe", "STUDIO-B-MON")])
self.snap.dante_map = {} # not mapped yet: just re-check via the module
self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller")])
def test_poe_hold_hands_the_mapped_entry_to_the_switch_code(self):
calls = []
self.engine.poe_executor = lambda cfg, name, entry: calls.append((name, entry["mac"])) or True
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
devices.remove("STUDIO-B-MON")
devices.remove("STUDIO-B-TB")
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
self.assertEqual(self.plan("dante"), [("fix", "poe_stb_mon"), ("fix", "poe_stb_tb")])
for fid in ("poe_stb_solo", "poe_stb_mon", "poe_stb_tb", "poe_stb_tbmon"):
f = CONFIG["fixes"][fid]
self.assertEqual(f["switch"], "10.66.1.102")
self.assertRegex(f["mac"], r"^00:1d:c1:")
self.snap.dante_map = {"STUDIO-B-TB": {"mac": "00:1d:c1:89:1a:0a", "switch": "10.66.1.102",
"port": "0/16", "shared": 1, "poe": True}}
_, t = run_steps(self.engine, self.snap, SUNDAY, 4)
self.engine.run_fixes("dante", self.snap, None, t)
time.sleep(0.2)
self.assertEqual(calls, [("STUDIO-B-TB", "00:1d:c1:89:1a:0a")])
tiles = self.engine.step(self.snap, time.time())
self.assertEqual(tiles["dante"]["x"], "DANTE\nCHECKING\nPOWER CYCLE")
def test_netgear_fix_calls_the_switch_with_the_mac_guard(self):
import types
calls = []
fake = types.ModuleType("netgear")
fake.poe_cycle = lambda host, port, mac=None: calls.append((host, port, mac))
sys.modules["netgear"] = fake
try:
self.assertTrue(sp.execute_fix(CONFIG, "poe_stb_solo", {}))
finally:
del sys.modules["netgear"]
self.assertEqual(calls, [("10.66.1.102", "0/14", "00:1d:c1:51:34:c8")])
def test_dante_map_poller(self):
class Comp:
def variable(self, label, name):
return {"STUDIO-B-MON_ip": "169.254.133.24", "NA-Tracks_ip": "169.254.99.70"}.get(name)
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(["STUDIO-B-MON", "NA-Tracks"])
self.snap.dante_map = {"BOH-WING": {"mac": "x", "poe": False}} # an older entry is kept
arp = lambda: {"169.254.133.24": "00:1d:c1:51:18:84", "169.254.99.70": "00:e0:4c:ff:b6:8a"}
survey = lambda switches, macs: {"00:1d:c1:51:18:84": {"switch": "10.66.1.102", "port": "0/15",
"shared": 1, "poe": True}}
path = os.path.join(tempfile.mkdtemp(), "dante-map.json")
sp.poll_dante_map(self.snap, CONFIG, Comp(), path, arp=arp, survey=survey)
self.assertEqual(self.snap.dante_map["STUDIO-B-MON"]["port"], "0/15")
self.assertTrue(self.snap.dante_map["STUDIO-B-MON"]["poe"])
self.assertNotIn("poe", self.snap.dante_map["NA-Tracks"]) # a Mac: not on a PoE port
self.assertIn("BOH-WING", self.snap.dante_map)
self.assertEqual(sp.load_dante_map(path), self.snap.dante_map)
self.assertEqual(sp.normalize_mac("0:1d:c1:89:1a:a"), "00:1d:c1:89:1a:0a")
def test_ingest_dvs_and_no_auto_fix(self):
devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")])
@@ -588,7 +611,7 @@ class TestFixes(unittest.TestCase):
def test_every_fix_is_valid(self):
for f in CONFIG["fixes"].values():
self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck", "netgear_poe"))
self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck"))
class FakeDeck:
@@ -663,6 +686,39 @@ class TestHyperDeckProbe(unittest.TestCase):
deck.close()
class TestNetgearParsing(unittest.TestCase):
BIG = """ VLAN ID MAC Address Interface IfIndex Status
66 00:1D:C1:51:34:C8 0/14 14 Learned
66 00:1D:C1:51:18:84 0/15 15 Learned
66 00:E0:4C:9A:1A:F3 0/26 26 Learned
66 00:0E:DD:E2:8E:C2 0/26 26 Learned
66 00:0E:DD:E2:8F:60 0/26 26 Learned"""
SMALL = """ 66 00:1D:C1:51:34:C8 0/10 10 Learned
66 00:1D:C1:51:18:84 0/10 10 Learned
66 00:E0:4C:9A:1A:F3 0/3 3 Learned"""
def test_access_port_is_where_the_device_has_fewest_neighbours(self):
import netgear
best = netgear.best_ports({"10.66.1.100": self.SMALL, "10.66.1.102": self.BIG},
["00:1d:c1:51:34:c8", "00:e0:4c:9a:1a:f3", "00:0e:dd:e2:8e:c2"])
self.assertEqual(best["00:1d:c1:51:34:c8"], ("10.66.1.102", "0/14", 1)) # not the uplink 0/10
self.assertEqual(best["00:e0:4c:9a:1a:f3"], ("10.66.1.100", "0/3", 1))
self.assertEqual(best["00:0e:dd:e2:8e:c2"][2], 3) # shared port: too many
self.assertGreater(3, netgear.MAX_SHARED) # 3 = uplink
self.assertTrue(netgear.poe_delivering("0/14 Yes 32000 1 1500 29 54 Delivering Power No Error"))
self.assertFalse(netgear.poe_delivering("0/14 Yes 32000 0 0 0 0 Searching No Error"))
def test_never_cycles_an_uplink(self):
import netgear
orig = netgear.survey
netgear.survey = lambda switches, macs: {macs[0]: {"switch": "s", "port": "0/26", "shared": 3, "poe": False}}
try:
with self.assertRaises(RuntimeError):
netgear.poe_cycle_device(["s"], "00:0e:dd:e2:8e:c2")
finally:
netgear.survey = orig
class FakeCompanion(BaseHTTPRequestHandler):
"""Just enough of Companion 5's HTTP API."""
store = {}
@@ -777,7 +833,8 @@ class TestAgainstFakeCompanion(unittest.TestCase):
store["conns"]["ATEM"]["category"] = "error"
store["custom"]["sp_cmd"] = ";ack atem"
data_dir = tempfile.mkdtemp()
originals = sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote
originals = sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map
sp.poll_dante_map = lambda *a, **k: None # never log in to a switch from a test
sp.poll_mdns = lambda snap, seconds=3: None
sp.poll_system = lambda snap, cfg: None
sp.poll_midi = lambda snap, cfg: None # CoreMIDI has its own test
@@ -797,7 +854,7 @@ class TestAgainstFakeCompanion(unittest.TestCase):
finally:
stop.set()
loop.join(timeout=15)
sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote = originals
sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map = originals
store["conns"]["ATEM"]["category"] = "good"
self.assertFalse(loop.is_alive()) # it shuts down cleanly
self.assertTrue(store["custom"]["sp_hb"].isdigit())