diff --git a/router_ui.py b/router_ui.py
new file mode 100755
index 0000000..f6e976d
--- /dev/null
+++ b/router_ui.py
@@ -0,0 +1,547 @@
+#!/usr/bin/env -S uv run --script
+# /// script
+# dependencies = ["bottle"]
+# ///
+
+import argparse
+import base64
+import json
+import logging
+import re
+import ssl
+import subprocess
+import threading
+import urllib.error
+import urllib.request
+import webbrowser
+from collections.abc import Callable
+from socketserver import ThreadingMixIn
+from wsgiref.simple_server import WSGIRequestHandler, WSGIServer, make_server
+
+import bottle
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
+logger = logging.getLogger(__name__)
+
+ROUTER_HOST = "192.168.1.1"
+ROUTER_USERNAME = "abdus"
+ROUTER_PASSWORD = "xAsametk50"
+INTERNET_TARGET = "8.8.8.8"
+UNIT_SECONDS = {"day": 86400, "hour": 3600, "min": 60, "sec": 1}
+
+PAGE = r"""
+
+
+
+
+Router
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+"""
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser(
+ description="Local dashboard for the Zyxel router",
+ formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+ )
+ parser.add_argument("--host", default="127.0.0.1")
+ parser.add_argument("--port", type=int, default=0, help="0 picks a random free port")
+ parser.add_argument("--timeout", type=float, default=10.0, help="Router request timeout in seconds")
+ parser.add_argument("--no-browser", action="store_true", help="Do not open the page on start")
+ return parser.parse_args()
+
+
+class Router:
+ def __init__(self, host: str, timeout: float) -> None:
+ self.base_url = f"https://{host}"
+ self.timeout = timeout
+ self.lock = threading.Lock()
+ self.session_key = ""
+ # Self-signed certificate; over HTTPS the router accepts a plain JSON login.
+ context = ssl.create_default_context()
+ context.check_hostname = False
+ context.verify_mode = ssl.CERT_NONE
+ self.opener = urllib.request.build_opener(
+ urllib.request.HTTPSHandler(context=context),
+ urllib.request.HTTPCookieProcessor(),
+ )
+
+ def request(self, path: str, body: dict | None = None) -> object:
+ separator = "&" if "?" in path else "?"
+ suffix = f"{separator}sessionkey={self.session_key}" if self.session_key else ""
+ request = urllib.request.Request(
+ url=f"{self.base_url}{path}{suffix}",
+ data=json.dumps(body).encode() if body is not None else None,
+ headers={"Content-Type": "application/json"},
+ method="POST" if body is not None else "GET",
+ )
+ with self.opener.open(request, timeout=self.timeout) as response:
+ raw = response.read().decode(errors="replace")
+ return json.loads(raw) if raw.strip() else {}
+
+ def login(self) -> None:
+ self.session_key = ""
+ reply = self.request(
+ path="/UserLogin",
+ body={
+ "Input_Account": ROUTER_USERNAME,
+ "Input_Passwd": base64.b64encode(ROUTER_PASSWORD.encode()).decode(),
+ "currLang": "en",
+ "RememberPassword": 0,
+ "SHA512_password": False,
+ },
+ )
+ if reply.get("result") != "ZCFG_SUCCESS":
+ raise RuntimeError(f"Login rejected: {reply}")
+ self.session_key = reply.get("sessionkey", "")
+ logger.info("Logged in to router")
+
+ def call(self, path: str, body: dict | None = None) -> object:
+ with self.lock:
+ if not self.session_key:
+ self.login()
+ try:
+ return self.request(path=path, body=body)
+ except (urllib.error.URLError, json.JSONDecodeError, ConnectionError):
+ logger.info("Router call failed, logging in again")
+ self.login()
+ return self.request(path=path, body=body)
+
+ def reboot(self) -> object:
+ with self.lock:
+ if not self.session_key:
+ self.login()
+ try:
+ return self.request(path="/cgi-bin/Reboot", body={})
+ finally:
+ self.session_key = ""
+
+
+def find_number(pattern: str, text: str) -> float | None:
+ match = re.search(pattern, text)
+ return float(match.group(1)) if match else None
+
+
+def parse_duration(text: str) -> int:
+ return sum(int(number) * UNIT_SECONDS[unit] for number, unit in re.findall(r"(\d+)\s*(day|hour|min|sec)", text))
+
+
+def parse_dsl(text: str) -> dict:
+ state = re.search(r"Line State:\s*(\S+)", text)
+ uptime = re.search(r"DSL uptime\s*:(.*)", text)
+ return {
+ "state": state.group(1) if state else None,
+ "uptime_s": parse_duration(text=uptime.group(1)) if uptime else None,
+ "down_kbps": find_number(pattern=r"near-end fast channel bit rate: (\d+)", text=text),
+ "up_kbps": find_number(pattern=r"far-end fast channel bit rate: (\d+)", text=text),
+ "snr_down_db": find_number(pattern=r"noise margin downstream: ([\d.]+)", text=text),
+ "snr_up_db": find_number(pattern=r"noise margin upstream: ([\d.]+)", text=text),
+ "attn_down_db": find_number(pattern=r"attenuation downstream: ([\d.]+)", text=text),
+ "attn_up_db": find_number(pattern=r"attenuation upstream: ([\d.]+)", text=text),
+ "errors": sum(int(count) for count in re.findall(r"(?:CRC|FEC|HEC) error \w+: (\d+)", text)),
+ }
+
+
+def ping(host: str | None) -> float | None:
+ if not host:
+ return None
+ result = subprocess.run(
+ ["ping", "-c", "1", "-W", "2000", host],
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+ return find_number(pattern=r"time=([\d.]+)", text=result.stdout)
+
+
+def read_device(router: Router) -> dict:
+ info = router.call(path="/cgi-bin/DAL?oid=status")["Object"][0]["DeviceInfo"]
+ return {
+ "model": info.get("ModelName"),
+ "firmware": info.get("SoftwareVersion"),
+ "uptime_s": info.get("UpTime"),
+ }
+
+
+def read_dsl(router: Router) -> dict:
+ return parse_dsl(text=router.call(path="/cgi-bin/xDSLStatistics_handle?line=0")[0]["result"])
+
+
+def read_wan(router: Router) -> dict:
+ entries = router.call(path="/cgi-bin/DAL?oid=wan")["Object"]
+ vdsl = next((entry for entry in entries if entry.get("Name") == "VDSL"), {})
+ return {
+ "ip": vdsl.get("IPAddress") or None,
+ "gateway": vdsl.get("GatewayIPAddress") or None,
+ "dns": vdsl.get("DNSServer") or None,
+ }
+
+
+def read_hosts(router: Router) -> list[dict]:
+ hosts = router.call(path="/cgi-bin/DAL?oid=lanhosts")["Object"][0]["lanhosts"]
+ return [
+ {
+ "name": host.get("HostName") or host.get("DeviceName") or "unknown",
+ "ip": host.get("IPAddress"),
+ "mac": host.get("PhysAddress"),
+ "connection": host.get("X_ZYXEL_ConnectionType"),
+ "active": bool(host.get("Active")),
+ "signal": host.get("X_ZYXEL_SignalStrength"),
+ }
+ for host in hosts
+ ]
+
+
+def safely(reader, router: Router) -> object:
+ try:
+ return reader(router)
+ except Exception as exc:
+ logger.warning("%s failed: %s", reader.__name__, exc)
+ return None
+
+
+def run_parallel(tasks: dict[str, Callable[[], object]]) -> dict[str, object]:
+ # Daemon threads, so Ctrl+C does not wait for slow router calls or pings.
+ results: dict[str, object] = {}
+
+ def worker(name: str, task: Callable[[], object]) -> None:
+ results[name] = task()
+
+ threads = [threading.Thread(target=worker, args=(name, task), daemon=True) for name, task in tasks.items()]
+ for thread in threads:
+ thread.start()
+ for thread in threads:
+ thread.join()
+ return results
+
+
+def overview(router: Router) -> dict:
+ results = run_parallel(
+ tasks={
+ "device": lambda: safely(read_device, router),
+ "dsl": lambda: safely(read_dsl, router),
+ "wan": lambda: safely(read_wan, router),
+ "hosts": lambda: safely(read_hosts, router),
+ "router_ms": lambda: ping(host=ROUTER_HOST),
+ "internet_ms": lambda: ping(host=INTERNET_TARGET),
+ }
+ )
+ wan = results["wan"]
+ gateway = wan["gateway"] if wan and wan["ip"] else None
+ return {
+ "device": results["device"],
+ "dsl": results["dsl"],
+ "wan": wan,
+ "hosts": results["hosts"],
+ "ping": {
+ "router_ms": results["router_ms"],
+ "isp_gateway_ms": ping(host=gateway),
+ "internet_ms": results["internet_ms"],
+ },
+ }
+
+
+def build_app(router: Router) -> bottle.Bottle:
+ app = bottle.Bottle()
+
+ @app.get("/")
+ def index() -> str:
+ return PAGE
+
+ @app.get("/api/overview")
+ def api_overview() -> dict:
+ return overview(router=router)
+
+ @app.get("/api/log")
+ def api_log() -> dict:
+ bottle.response.content_type = "application/json"
+ try:
+ return json.dumps(router.call(path="/cgi-bin/DAL?oid=zlog")["Object"])
+ except Exception as exc:
+ logger.warning("log failed: %s", exc)
+ bottle.response.status = 502
+ return json.dumps([])
+
+ @app.post("/api/reboot")
+ def api_reboot() -> dict:
+ logger.warning("Rebooting router on request")
+ try:
+ router.reboot()
+ except Exception as exc:
+ # The router may drop the connection while it goes down.
+ logger.info("Reboot call ended with: %s", exc)
+ return {"ok": True}
+
+ return app
+
+
+class ThreadingServer(ThreadingMixIn, WSGIServer):
+ daemon_threads = True
+
+
+class QuietHandler(WSGIRequestHandler):
+ def log_message(self, format: str, *args: object) -> None:
+ pass
+
+
+def main() -> None:
+ args = parse_args()
+ router = Router(host=ROUTER_HOST, timeout=args.timeout)
+ server = make_server(
+ host=args.host,
+ port=args.port,
+ app=build_app(router=router),
+ server_class=ThreadingServer,
+ handler_class=QuietHandler,
+ )
+ url = f"http://{args.host}:{server.server_port}"
+ logger.info("Serving %s", url)
+ if not args.no_browser:
+ opener = threading.Timer(interval=0.5, function=webbrowser.open, args=[url])
+ opener.daemon = True
+ opener.start()
+ try:
+ server.serve_forever(poll_interval=0.2)
+ except KeyboardInterrupt:
+ logger.info("Stopping")
+ finally:
+ server.server_close()
+
+
+if __name__ == "__main__":
+ main()