Add TP-Link Deco network stats tools

scripts/deco_bridge.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+import json
+import os
+import subprocess
+import sys
+
+from tplinkrouterc6u import TPLinkDecoClient
+
+
+def default_gateway():
+    try:
+        result = subprocess.run(
+            ["/sbin/route", "-n", "get", "default"],
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+    except Exception:
+        return None
+
+    for line in result.stdout.splitlines():
+        key, _, value = line.partition(":")
+        if key.strip() == "gateway":
+            return value.strip()
+    return None
+
+
+def config():
+    host = os.environ.get("DECO_HOST") or default_gateway()
+    password = os.environ.get("DECO_PASSWORD")
+    if not password:
+        raise RuntimeError("DECO_PASSWORD must be set in .env.")
+    if not host:
+        raise RuntimeError("DECO_HOST must be set in .env; default gateway detection failed.")
+
+    verify_ssl = os.environ.get("DECO_VERIFY_SSL", "true").lower() not in {
+        "0",
+        "false",
+        "no",
+    }
+    timeout = int(os.environ.get("DECO_TIMEOUT", "10"))
+    return {
+        "host": host,
+        "password": password,
+        "username": os.environ.get("DECO_USERNAME", "admin"),
+        "verify_ssl": verify_ssl,
+        "timeout": timeout,
+    }
+
+
+def client():
+    return TPLinkDecoClient(**config())
+
+
+def device_to_dict(device):
+    return {
+        "hostname": device.hostname,
+        "mac": device.macaddr,
+        "ip": device.ipaddr,
+        "connection": getattr(device.type, "value", str(device.type)),
+        "upSpeed": device.up_speed,
+        "downSpeed": device.down_speed,
+        "active": device.active,
+    }
+
+
+def firmware_to_dict(firmware):
+    return {
+        "model": firmware.model,
+        "hardwareVersion": firmware.hardware_version,
+        "firmwareVersion": firmware.firmware_version,
+    }
+
+
+def ipv4_to_dict(status):
+    return {
+        "wanMac": status.wan_macaddr,
+        "wanIp": status.wan_ipv4_ipaddr,
+        "wanGateway": status.wan_ipv4_gateway,
+        "wanConnectionType": status.wan_ipv4_conntype,
+        "wanNetmask": status.wan_ipv4_netmask,
+        "wanPrimaryDns": status.wan_ipv4_pridns,
+        "wanSecondaryDns": status.wan_ipv4_snddns,
+        "lanMac": status.lan_macaddr,
+        "lanIp": status.lan_ipv4_ipaddr,
+        "lanNetmask": status.lan_ipv4_netmask,
+    }
+
+
+def status_to_dict(status, include_clients=True):
+    data = {
+        "wanMac": status.wan_macaddr,
+        "lanMac": status.lan_macaddr,
+        "wanIp": status.wan_ipv4_addr,
+        "lanIp": status.lan_ipv4_addr,
+        "wanGateway": status.wan_ipv4_gateway,
+        "connectionType": status.conn_type,
+        "cpuUsage": status.cpu_usage,
+        "memoryUsage": status.mem_usage,
+        "clientsTotal": status.clients_total,
+        "wiredClientsTotal": status.wired_total,
+        "wifiClientsTotal": status.wifi_clients_total,
+        "guestClientsTotal": status.guest_clients_total,
+        "iotClientsTotal": status.iot_clients_total,
+        "wifi": {
+            "host2g": status.wifi_2g_enable,
+            "host5g": status.wifi_5g_enable,
+            "host6g": status.wifi_6g_enable,
+            "guest2g": status.guest_2g_enable,
+            "guest5g": status.guest_5g_enable,
+            "guest6g": status.guest_6g_enable,
+        },
+    }
+    if include_clients:
+        data["clients"] = [device_to_dict(device) for device in status.devices]
+    return data
+
+
+def run(action):
+    deco = client()
+    try:
+        if action == "overview":
+            status = deco.get_status()
+            firmware = deco.get_firmware()
+            return {
+                "status": status_to_dict(status, include_clients=False),
+                "firmware": firmware_to_dict(firmware),
+            }
+        if action == "clients":
+            status = deco.get_status()
+            return {"clients": [device_to_dict(device) for device in status.devices]}
+        if action == "ipv4":
+            return ipv4_to_dict(deco.get_ipv4_status())
+        if action == "firmware":
+            return firmware_to_dict(deco.get_firmware())
+        raise RuntimeError(f"Unknown action: {action}")
+    finally:
+        try:
+            deco.logout()
+        except Exception:
+            pass
+
+
+def main():
+    if len(sys.argv) != 2:
+        raise RuntimeError("Usage: deco_bridge.py <overview|clients|ipv4|firmware>")
+    print(json.dumps(run(sys.argv[1]), indent=2))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as err:
+        print(json.dumps({"error": str(err)}), file=sys.stderr)
+        sys.exit(1)

scripts/setup-python.sh

@@ -0,0 +1,11 @@
+#!/bin/zsh
+set -euo pipefail
+
+PROJECT_DIR="$(cd "$(dirname "$0")/.." && pwd)"
+cd "$PROJECT_DIR"
+
+if [[ ! -x .venv/bin/python ]]; then
+  python3 -m venv .venv
+fi
+
+.venv/bin/python -m pip install -r requirements.txt