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

Smart outlets: the panel can power-cycle a device through a SmartLife (Tuya) plug

Browse Source 
settings/outlet.py talks to Tuya's cloud with the campus's own project keys (kept in outlets.json
beside the config, 0600, never in the repo): list, status, on/off, and 'cycle <outlet> --ping
<address>' which leaves a unit that answers alone, cuts power 8 s, and waits for the address to
answer; refused while a HyperDeck is recording. A builder 'Power-cycle a smart outlet' makes the
fix from fields; the TASCAM key's ping check now holds to it (verified live: the hung DA-6400
answered again 26 s after the cycle and Companion reconnected). Settings > Outlets: keys, the
outlet list with live state and on/off, + Outlet from the account's devices. Calendar in 12-hour time.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-14 19:14:53 -04:00
co-authored by Claude Fable 5.1
parent 095f20b7a6
commit f61761f9bc
6 changed files with 297 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
settings/outlet.py
+187
View File
@@ -0,0 +1,187 @@
#!/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] [--ignore-live]
off, N s (default 8), on; then, with --ping, wait up to 90 s for the address to answer.
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 cycle(name, wait=8, ping=None, ignore_live=False, data=None):
"""off, wait, on; then wait for the address to answer. 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
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)
end = time.time() + 90
while time.time() < end:
if pings(ping):
return True, "%s power-cycled; %s answers again after %d s" % (name, ping, 90 - int(end - time.time()))
time.sleep(3)
return False, "%s power-cycled but %s still does not answer after 90 s" % (name, ping)
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, ignore = 8, 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] == "--ignore-live": ignore = True; i += 1
else: i += 1
ok, words = cycle(args[0], wait, ping, 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)