Use luci-rpc leases and hostapd clients

services/openwrt.py

@@ -90,7 +90,11 @@ def _extract_wifi_clients(wireless: dict[str, Any]) -> list[str]:
 
 
 def _extract_leases(leases: dict[str, Any]) -> list[str]:
-    items = leases.get("leases") if isinstance(leases, dict) else None
+    items = None
+    if isinstance(leases, dict):
+        items = leases.get("leases") or leases.get("dhcp_leases") or leases.get("ipv4_leases")
+    elif isinstance(leases, list):
+        items = leases
     if not isinstance(items, list):
         return []
     out = []
@@ -104,6 +108,46 @@ def _extract_leases(leases: dict[str, Any]) -> list[str]:
     return out
 
 
+def _extract_ifnames(wireless: dict[str, Any]) -> list[str]:
+    ifnames: list[str] = []
+    if not isinstance(wireless, dict):
+        return ifnames
+    for radio in wireless.values():
+        if not isinstance(radio, dict):
+            continue
+        for iface in radio.get("interfaces", []) or []:
+            if not isinstance(iface, dict):
+                continue
+            ifname = iface.get("ifname")
+            if ifname:
+                ifnames.append(str(ifname))
+    return ifnames
+
+
+def _parse_hostapd_clients(raw: str, ifname: str) -> list[str]:
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError:
+        return []
+    data = payload.get("clients") if isinstance(payload, dict) else None
+    if isinstance(data, dict):
+        items = data.items()
+    elif isinstance(payload, dict):
+        items = payload.items()
+    else:
+        return []
+    clients: list[str] = []
+    for mac, meta in items:
+        if not isinstance(meta, dict):
+            continue
+        signal = meta.get("signal")
+        rx = _format_rate((meta.get("rx") or {}).get("rate"))
+        tx = _format_rate((meta.get("tx") or {}).get("rate"))
+        sig = f"{signal}dBm" if isinstance(signal, (int, float)) else "?"
+        clients.append(f"{ifname} {mac} {sig} rx:{rx} tx:{tx}")
+    return clients
+
+
 def _parse_proc_fallback(raw: str) -> tuple[int | None, list[float] | None]:
     uptime = None
     load = None
@@ -162,12 +206,13 @@ async def get_openwrt_status(cfg: dict[str, Any]) -> str:
     ssh_cmd += ["-p", str(port), f"{user}@{host}"]
 
     remote = (
-        "if ubus call system info >/tmp/ow_sys.json 2>/dev/null; then "
+        "ubus call system info >/tmp/ow_sys.json 2>/dev/null && "
-        "cat /tmp/ow_sys.json; else cat /proc/uptime; echo; cat /proc/loadavg; fi; echo __SEP__;"
+        "cat /tmp/ow_sys.json || (cat /proc/uptime; echo; cat /proc/loadavg); "
+        "echo __SEP__;"
         "ubus call network.interface.wan status; echo __SEP__;"
         "ubus call network.wireless status; echo __SEP__;"
-        "if ubus call dhcp ipv4leases >/tmp/ow_leases.json 2>/dev/null; then "
+        "ubus call luci-rpc getDHCPLeases '{\"family\":4}' >/tmp/ow_leases.json 2>/dev/null && "
-        "cat /tmp/ow_leases.json; else cat /tmp/dhcp.leases; fi"
+        "cat /tmp/ow_leases.json || cat /tmp/dhcp.leases"
     )
     cmd = ssh_cmd + ["sh", "-c", remote]
     rc, out = await run_cmd(cmd, timeout=timeout_sec + 5)
@@ -211,6 +256,18 @@ async def get_openwrt_status(cfg: dict[str, Any]) -> str:
     wan_state = "up" if wan_up else "down"
 
     wifi_clients = _extract_wifi_clients(wireless)
+    ifnames = _extract_ifnames(wireless)
+    if ifnames:
+        for ifname in ifnames:
+            cmd_clients = ssh_cmd + [
+                "sh",
+                "-c",
+                f"ubus call hostapd.{ifname} get_clients 2>/dev/null || echo '{{}}'",
+            ]
+            rc2, out2 = await run_cmd(cmd_clients, timeout=timeout_sec + 5)
+            if rc2 == 0 and out2.strip():
+                wifi_clients.extend(_parse_hostapd_clients(out2.strip(), ifname))
+
     leases_list = _extract_leases(leases) if leases else _parse_leases_fallback(parts[3])
 
     lines = [