mac_mcp/scripts/deco_bridge.py

#!/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)