Live power APIs: detail/load/EMS queries + power module + host_probe8

scripts/host_probe8.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""Host probe v8 — live household wattage (stdlib only).
+
+Three socket APIs (all verified in disassembly; detail/load pending live
+confirmation on this unit, EMS already answered live):
+* DEVICE_DETAIL_INFO (1301, no args) -> totalPower  [home-page poll]
+* LOAD_STATUS        (1411, no args) -> SW1ExpPower [per-switch watts]
+* EMS                (1707, no args) -> grid/load V+A [watts via math]
+
+Run on the SoftAP-joined host:
+    python3 host_probe8.py --sn A03X2516013210060005 --ports 9000
+"""
+
+import argparse
+import json
+import socket
+import time
+import zlib
+
+DETAIL, LOAD, EMS = 0x515, 0x583, 0x6AB
+SNNO = [0]
+
+
+def secret_key(sn):
+    return format(sum(sn.encode()) & 0xFF, "x")
+
+
+def build_envelope(sn, cmd, args=None, snno=None, ts=None, equip=None):
+    if snno is None:
+        SNNO[0] = (SNNO[0] % 9999) + 1
+        snno = SNNO[0]
+    data = {"opt": 0}
+    if args:
+        data.update(args)
+    compact = json.dumps(data, separators=(",", ":"))
+    env = {"cmdType": cmd, "equipNo": equip if equip is not None else sn,
+           "type": 0,
+           "timeStamp": ts if ts is not None else int(time.time()),
+           "snno": snno,
+           "len": len(compact.encode()),
+           "crc": format(zlib.crc32(compact.encode()), "X"),
+           "dataArea": data}
+    return json.dumps(env, separators=(",", ":"))
+
+
+def encrypt(json_str, key):
+    parts = json_str.split(",")
+    a = (parts[0] + "," + parts[1] + ",").encode()
+    b = ",".join(parts[2:]).encode()
+    n = int(key, 16)
+    return a + bytes((b[i] + i + len(a) + n) & 0xFF
+                     for i in range(len(b)))
+
+
+def decrypt_wire(wire):
+    first = wire.find(b",")
+    second = wire.find(b",", first + 1)
+    if first < 0 or second < 0:
+        return ""
+    a, b = wire[:second + 1], wire[second + 1:]
+    try:
+        probe = json.loads(a[:-1].decode() + "}")
+    except Exception:
+        return ""
+    if not isinstance(probe, dict):
+        return ""
+    n = int(secret_key(str(probe.get("equipNo", ""))), 16)
+    try:
+        return (a + bytes((t - i - len(a) - n) & 0xFF
+                          for i, t in enumerate(b))).decode()
+    except Exception:
+        return ""
+
+
+def recv_all(sock, idle=8.0):
+    out = bytearray()
+    sock.settimeout(idle)
+    while True:
+        try:
+            chunk = sock.recv(65536)
+        except (socket.timeout, TimeoutError):
+            break
+        if not chunk:
+            break
+        out.extend(chunk)
+    return bytes(out)
+
+
+def run(label, host, port, tx, timeout):
+    rx, err, reply = b"", None, None
+    try:
+        with socket.create_connection((host, port),
+                                      timeout=timeout) as sock:
+            sock.sendall(tx)
+            rx = recv_all(sock)
+            if rx:
+                dec = decrypt_wire(rx)
+                reply = json.loads(dec) if dec else {"_raw": rx.hex()}
+    except (ConnectionError, socket.error, TimeoutError) as exc:
+        err = "%s: %s" % (type(exc).__name__, exc)
+    print("--- %s -> %s:%d" % (label, host, port))
+    print("  rx (%d B)" % len(rx))
+    if reply is not None:
+        print("  reply: " + json.dumps(reply, default=str)[:2000])
+    if err:
+        print("  error: %s" % err)
+    return reply
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--sn", required=True)
+    ap.add_argument("--host", default="10.100.1.1")
+    ap.add_argument("--ports", default="9000")
+    ap.add_argument("--timeout", type=float, default=5.0)
+    args = ap.parse_args()
+    key = secret_key(args.sn)
+    print("secret key for %s: %s (radix 16)" % (args.sn, key))
+    for port in [int(p) for p in args.ports.split(",") if p.strip()]:
+        for label, cmd in (("detail/1301", DETAIL), ("load/1411", LOAD),
+                           ("ems/1707", EMS)):
+            tx = encrypt(build_envelope(args.sn, cmd), key)
+            run(label, args.host, port, tx, args.timeout)
+
+
+if __name__ == "__main__":
+    main()