status-indicators/settings/outlet.py

#!/usr/bin/env python3
"""Smart outlets the panel may switch (a hung recorder's mains, say), through Tuya's cloud with the
location'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)