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

Outlet fix: waits for the device's Dante card too; the Tascam cut lasts 45 s

Browse Source 
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-14 19:41:41 -04:00
co-authored by Claude Fable 5.1
parent f61761f9bc
commit edfd218ac9
2 changed files with 34 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
settings/builders.py
+3 -2
View File
@@ -200,9 +200,10 @@ TEMPLATES = {
"os": "self", "closes": True,
"params": [("outlet", "Outlet", "As named under Settings > Outlets.", "str", None),
("address", "Address to watch", "The device's address: no power cycle while it answers; afterwards wait for it to answer.", "str", ""),
("dante", "Dante name", "Its Dante device name, if it has a Dante card: the fix waits for it to be back too.", "str", ""),
("wait_seconds", "Off for", "Seconds the outlet stays off.", "int", 8)],
"render": lambda p, cfg: "/usr/bin/python3 \"$HOME/Library/Application Support/status-panel/settings/outlet.py\" cycle %s%s --wait %d"
% (q(p["outlet"]), (" --ping " + q(p["address"])) if p.get("address") else "", int(p.get("wait_seconds", 8))),
"render": lambda p, cfg: "/usr/bin/python3 \"$HOME/Library/Application Support/status-panel/settings/outlet.py\" cycle %s%s%s --wait %d"
% (q(p["outlet"]), (" --ping " + q(p["address"])) if p.get("address") else "", (" --dante " + q(p["dante"])) if p.get("dante") else "", int(p.get("wait_seconds", 8))),
"suggest": lambda p, cfg: "power-cycle the %s outlet when %s does not answer, then wait for it to come back (never while a HyperDeck is recording)"
% (p["outlet"], p.get("address") or "its device"),
},
settings/outlet.py
+31 -12
View File
@@ -6,8 +6,9 @@ campus's own project keys. Credentials and the outlet list live in outlets.json
outlet.py list the outlets on the account (name, id, on/off)
outlet.py status <outlet> on / off
outlet.py on|off <outlet>
outlet.py cycle <outlet> [--wait N] [--ping ADDRESS] [--ignore-live]
off, N s (default 8), on; then, with --ping, wait up to 90 s for the address to answer.
outlet.py cycle <outlet> [--wait N] [--ping ADDRESS] [--dante NAME] [--ignore-live]
off, N s (default 8), on; then, with --ping, wait up to 90 s for the address to answer and,
with --dante, up to 180 s for that device to be back in Companion's Dante Controller list.
With --ping, a unit that answers already is left alone. Refused while a HyperDeck is
recording (live-now.json) unless --ignore-live.
"""
@@ -127,15 +128,24 @@ def pings(address):
return subprocess.run(["/sbin/ping", "-c", "1", "-W", "1000", address], capture_output=True).returncode == 0
def cycle(name, wait=8, ping=None, ignore_live=False, data=None):
"""off, wait, on; then wait for the address to answer. Returns (ok, words)."""
def dante_has(name):
"""Is the device in the Dante Controller connection's device list in Companion (local API)?"""
try:
with urllib.request.urlopen("http://127.0.0.1:8000/api/variable/audinate-dantecontroller/devices/value", timeout=5) as r:
return name in r.read().decode(errors="replace")
except (OSError, urllib.error.URLError):
return False
def cycle(name, wait=8, ping=None, dante=None, ignore_live=False, data=None):
"""off, wait, on; then wait for the address to answer (and the Dante device to be back). Returns (ok, words)."""
data = data or load()
if not ignore_live:
live, reason = live_now()
if live:
return False, "not now: %s" % reason
if ping and pings(ping):
return True, "%s answers ping already: no power cycle" % ping
if ping and pings(ping) and (not dante or dante_has(dante)):
return True, "%s answers already: no power cycle" % ping
o = outlet(data, name)
tu = client(data)
code = tu.switch(o["id"], False)
@@ -143,12 +153,20 @@ def cycle(name, wait=8, ping=None, ignore_live=False, data=None):
tu.switch(o["id"], True, code)
if not ping:
return True, "%s switched off %d s and on again" % (name, wait)
end = time.time() + 90
while time.time() < end:
t0 = time.time()
while time.time() < t0 + 90:
if pings(ping):
return True, "%s power-cycled; %s answers again after %d s" % (name, ping, 90 - int(end - time.time()))
break
time.sleep(3)
return False, "%s power-cycled but %s still does not answer after 90 s" % (name, ping)
else:
return False, "%s power-cycled but %s still does not answer after 90 s" % (name, ping)
if not dante:
return True, "%s power-cycled; %s answers again after %d s" % (name, ping, int(time.time() - t0))
while time.time() < t0 + 180:
if dante_has(dante):
return True, "%s power-cycled; %s and its Dante card are back after %d s" % (name, ping, int(time.time() - t0))
time.sleep(5)
return False, "%s power-cycled; %s answers but its Dante card (%s) is not back after 180 s - press its power button off and on" % (name, ping, dante)
def main(argv):
@@ -166,14 +184,15 @@ def main(argv):
st = tu.status(o["id"]); print("on" if any(v for c, v in st.items() if c.startswith("switch")) else "off"); return 0
tu.switch(o["id"], verb == "on"); print(verb); return 0
if verb == "cycle":
wait, ping, ignore = 8, None, False
wait, ping, dante, ignore = 8, None, None, False
i = 1
while i < len(args):
if args[i] == "--wait": wait = int(args[i + 1]); i += 2
elif args[i] == "--ping": ping = args[i + 1]; i += 2
elif args[i] == "--dante": dante = args[i + 1]; i += 2
elif args[i] == "--ignore-live": ignore = True; i += 1
else: i += 1
ok, words = cycle(args[0], wait, ping, ignore, data)
ok, words = cycle(args[0], wait, ping, dante, ignore, data)
print(words)
return 0 if ok else 1
print("unknown verb", verb); return 2