christianandClaude Fable 5.1
d5089cfdde
The committed config.json is the campus as applied on NA Ingest; the READY-page work still in progress in this checkout is left out of it. The outlet fix now says, when the Dante card does not come back, that only the unit's power button resets it (verified: a 45 s cut does not), and still reconnects Companion. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
207 lines
8.8 KiB
Python
207 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Smart outlets the panel may switch (a hung recorder's mains, say), through Tuya's cloud with the
|
|
campus's own project keys. Credentials and the outlet list live in outlets.json beside the config
|
|
(0600, never in the repo). Also a command for the watchdog's fixes:
|
|
|
|
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] [--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.
|
|
"""
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
|
|
DATA_DIR = os.environ.get("STATUS_PANEL_DATA") or os.path.expanduser("~/Library/Application Support/status-panel")
|
|
FILE = os.path.join(DATA_DIR, "outlets.json")
|
|
REGIONS = {"us": "https://openapi.tuyaus.com", "eu": "https://openapi.tuyaeu.com", "cn": "https://openapi.tuyacn.com", "in": "https://openapi.tuyain.com"}
|
|
|
|
|
|
def load():
|
|
try:
|
|
with open(FILE, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (OSError, ValueError):
|
|
return {"tuya": {}, "outlets": {}}
|
|
|
|
|
|
def save(data):
|
|
tmp = FILE + ".tmp"
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=1)
|
|
os.chmod(tmp, 0o600)
|
|
os.replace(tmp, FILE)
|
|
|
|
|
|
class Tuya:
|
|
"""The few cloud calls the panel needs, signed the way Tuya's OpenAPI wants (HMAC-SHA256)."""
|
|
|
|
def __init__(self, access_id, access_secret, region="us"):
|
|
self.id, self.secret = access_id, access_secret
|
|
self.base = REGIONS.get(region, region)
|
|
self.token, self.token_until = None, 0
|
|
|
|
def _sign(self, method, path, body, token):
|
|
t = str(int(time.time() * 1000))
|
|
nonce = uuid.uuid4().hex
|
|
body_hash = hashlib.sha256((body or "").encode()).hexdigest()
|
|
string = "%s\n%s\n\n%s" % (method, body_hash, path)
|
|
payload = self.id + (token or "") + t + nonce + string
|
|
sign = hmac.new(self.secret.encode(), payload.encode(), hashlib.sha256).hexdigest().upper()
|
|
return {"client_id": self.id, "sign": sign, "t": t, "nonce": nonce, "sign_method": "HMAC-SHA256", **({"access_token": token} if token else {})}
|
|
|
|
def call(self, method, path, body=None, auth=True):
|
|
if auth and (not self.token or time.time() > self.token_until - 60):
|
|
r = self.call("GET", "/v1.0/token?grant_type=1", auth=False)
|
|
self.token, self.token_until = r["access_token"], time.time() + r.get("expire_time", 3600)
|
|
data = json.dumps(body) if body is not None else None
|
|
headers = dict(self._sign(method, path, data, self.token if auth else None), **{"Content-Type": "application/json"})
|
|
req = urllib.request.Request(self.base + path, data=data.encode() if data else None, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
out = json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
raise RuntimeError("Tuya %s %s: HTTP %s" % (method, path, e.code))
|
|
if not out.get("success"):
|
|
raise RuntimeError("Tuya %s %s: %s (%s)" % (method, path, out.get("msg"), out.get("code")))
|
|
return out["result"]
|
|
|
|
def devices(self):
|
|
"""Every device of the app accounts linked to the project: name, id, local key, address, category."""
|
|
out, last = [], ""
|
|
while True:
|
|
r = self.call("GET", "/v1.0/iot-01/associated-users/devices?size=100" + ("&last_row_key=%s" % last if last else ""))
|
|
for d in r.get("devices", []):
|
|
out.append({"id": d["id"], "name": d.get("name"), "local_key": d.get("local_key"), "ip": d.get("ip"), "category": d.get("category"),
|
|
"online": d.get("online"), "product": d.get("product_name")})
|
|
if not r.get("has_more"):
|
|
return out
|
|
last = r.get("last_row_key", "")
|
|
|
|
def status(self, device_id):
|
|
r = self.call("GET", "/v1.0/iot-03/devices/%s/status" % device_id)
|
|
return {x["code"]: x["value"] for x in r}
|
|
|
|
def switch(self, device_id, on, code=None):
|
|
if code is None:
|
|
st = self.status(device_id)
|
|
code = next((c for c in ("switch_1", "switch", "switch_2") if c in st), None) or next((c for c in st if c.startswith("switch")), "switch_1")
|
|
self.call("POST", "/v1.0/iot-03/devices/%s/commands" % device_id, {"commands": [{"code": code, "value": bool(on)}]})
|
|
return code
|
|
|
|
|
|
def client(data=None):
|
|
data = data or load()
|
|
t = data.get("tuya") or {}
|
|
if not (t.get("access_id") and t.get("access_secret")):
|
|
raise RuntimeError("no Tuya keys in %s (Settings > Outlets)" % FILE)
|
|
return Tuya(t["access_id"], t["access_secret"], t.get("region", "us"))
|
|
|
|
|
|
def outlet(data, name):
|
|
o = (data.get("outlets") or {}).get(name)
|
|
if not o:
|
|
raise RuntimeError("no outlet called %s (Settings > Outlets)" % name)
|
|
return o
|
|
|
|
|
|
def live_now():
|
|
try:
|
|
d = json.load(open(os.path.join(DATA_DIR, "live-now.json")))
|
|
return bool(d.get("live")) and time.time() - d.get("time", 0) < 60, d.get("reason", "")
|
|
except (OSError, ValueError):
|
|
return False, ""
|
|
|
|
|
|
def pings(address):
|
|
return subprocess.run(["/sbin/ping", "-c", "1", "-W", "1000", address], capture_output=True).returncode == 0
|
|
|
|
|
|
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) 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)
|
|
time.sleep(wait)
|
|
tu.switch(o["id"], True, code)
|
|
if not ping:
|
|
return True, "%s switched off %d s and on again" % (name, wait)
|
|
t0 = time.time()
|
|
while time.time() < t0 + 90:
|
|
if pings(ping):
|
|
break
|
|
time.sleep(3)
|
|
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 True, "%s power-cycled; %s answers again, but its Dante card (%s) only comes back from the unit's own power button: press it off and on" % (name, ping, dante)
|
|
|
|
|
|
def main(argv):
|
|
if not argv:
|
|
print(__doc__); return 2
|
|
verb, args = argv[0], argv[1:]
|
|
data = load()
|
|
if verb == "list":
|
|
for d in client(data).devices():
|
|
print("%-24s %s %s online=%s" % (d["name"], d["id"], d.get("product") or d.get("category"), d.get("online")))
|
|
return 0
|
|
if verb in ("status", "on", "off"):
|
|
o = outlet(data, args[0]); tu = client(data)
|
|
if verb == "status":
|
|
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, 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, dante, ignore, data)
|
|
print(words)
|
|
return 0 if ok else 1
|
|
print("unknown verb", verb); return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main(sys.argv[1:]))
|
|
except RuntimeError as e:
|
|
print(e, file=sys.stderr)
|
|
sys.exit(1)
|