#!/usr/bin/env python3 import argparse import base64 import csv import json import logging import re import ssl import subprocess import time import urllib.request from datetime import datetime from pathlib import Path 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" COLUMNS = [ "timestamp", "router_uptime_s", "dsl_state", "dsl_uptime_s", "down_kbps", "up_kbps", "snr_down_db", "snr_up_db", "attn_down_db", "attn_up_db", "dsl_errors", "wan_ip", "ping_router_ms", "ping_isp_gateway_ms", "ping_internet_ms", ] UNIT_SECONDS = {"day": 86400, "hour": 3600, "min": 60, "sec": 1} SLOW_PING_MS = 150 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Log DSL, PPPoE and ping health of the Zyxel router to a CSV to tell ISP from device drops", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("--output", type=Path, default=Path.cwd() / "router_monitor.csv") parser.add_argument("--interval", type=float, default=30.0, help="Seconds between samples") parser.add_argument("--internet-target", default="8.8.8.8") parser.add_argument("--timeout", type=float, default=10.0, help="Router request timeout in seconds") return parser.parse_args() def build_opener() -> urllib.request.OpenerDirector: # Self-signed certificate; over HTTPS the router accepts plain JSON login. context = ssl.create_default_context() context.check_hostname = False context.verify_mode = ssl.CERT_NONE return urllib.request.build_opener( urllib.request.HTTPSHandler(context=context), urllib.request.HTTPCookieProcessor(), ) def login(opener: urllib.request.OpenerDirector, base_url: str, timeout: float) -> str: request = urllib.request.Request( url=f"{base_url}/UserLogin", data=json.dumps( { "Input_Account": ROUTER_USERNAME, "Input_Passwd": base64.b64encode(ROUTER_PASSWORD.encode()).decode(), "currLang": "en", "RememberPassword": 0, "SHA512_password": False, } ).encode(), headers={"Content-Type": "application/json"}, method="POST", ) with opener.open(request, timeout=timeout) as response: body = json.loads(response.read().decode()) if body.get("result") != "ZCFG_SUCCESS": raise RuntimeError(f"Login rejected: {body}") return body.get("sessionkey", "") def get_json(opener: urllib.request.OpenerDirector, url: str, timeout: float) -> object: with opener.open(url, timeout=timeout) as response: return json.loads(response.read().decode(errors="replace")) 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 find_number(pattern: str, text: str) -> float | None: match = re.search(pattern, text) return float(match.group(1)) if match else None def parse_dsl(text: str) -> dict: state = re.search(r"Line State:\s*(\S+)", text) uptime = re.search(r"DSL uptime\s*:(.*)", text) errors = sum(int(count) for count in re.findall(r"(?:CRC|FEC|HEC) error \w+: (\d+)", text)) return { "dsl_state": state.group(1) if state else "", "dsl_uptime_s": parse_duration(text=uptime.group(1)) if uptime else "", "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), "dsl_errors": errors, } def read_router(opener: urllib.request.OpenerDirector, base_url: str, session_key: str, timeout: float) -> dict: def fetch(path: str) -> object: return get_json(opener=opener, url=f"{base_url}{path}&sessionkey={session_key}", timeout=timeout) dsl = fetch("/cgi-bin/xDSLStatistics_handle?line=0") status = fetch("/cgi-bin/DAL?oid=status") wan = fetch("/cgi-bin/DAL?oid=wan") sample = parse_dsl(text=dsl[0]["result"]) sample["router_uptime_s"] = status["Object"][0]["DeviceInfo"]["UpTime"] vdsl = next((entry for entry in wan["Object"] if entry.get("Name") == "VDSL"), {}) sample["wan_ip"] = vdsl.get("IPAddress", "") sample["isp_gateway"] = vdsl.get("GatewayIPAddress", "") return sample def ping(host: str) -> 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 is_slow(sample: dict) -> bool: return (sample.get("ping_internet_ms") or 0) > SLOW_PING_MS def describe(sample: dict) -> str: def ms(key: str) -> str: value = sample.get(key) return "LOST" if value is None else f"{value:.0f}ms" return ( f"router {ms('ping_router_ms')} | internet {ms('ping_internet_ms')} | " f"dsl {sample.get('dsl_state') or '?'} up {sample.get('dsl_uptime_s', '?')}s | " f"errors {sample.get('dsl_errors', '?')} | wan {sample.get('wan_ip') or '-'}" ) def log_changes(previous: dict, current: dict) -> None: if not previous: return if current.get("router_uptime_s") != "" and previous.get("router_uptime_s") != "": if current["router_uptime_s"] < previous["router_uptime_s"]: logger.warning("ROUTER REBOOTED (uptime reset)") if current.get("dsl_uptime_s") != "" and previous.get("dsl_uptime_s") != "": if current["dsl_uptime_s"] < previous["dsl_uptime_s"]: logger.warning("DSL LINE RESYNCED (line side / ISP)") if current.get("dsl_state") != previous.get("dsl_state"): logger.warning("DSL state %s -> %s", previous.get("dsl_state"), current.get("dsl_state")) if current.get("wan_ip") != previous.get("wan_ip"): logger.warning("WAN IP %s -> %s (PPPoE session changed)", previous.get("wan_ip"), current.get("wan_ip")) for key, label in (("ping_router_ms", "router"), ("ping_internet_ms", "internet")): if (previous.get(key) is None) != (current.get(key) is None): logger.warning("Ping to %s %s", label, "LOST" if current.get(key) is None else "RESTORED") if is_slow(sample=current) != is_slow(sample=previous): if is_slow(sample=current): logger.warning("Internet ping SLOW: %.0f ms", current["ping_internet_ms"]) else: logger.warning("Internet ping back to normal") def main() -> None: args = parse_args() base_url = f"https://{ROUTER_HOST}" opener = build_opener() session_key = "" previous: dict = {} new_file = not args.output.exists() with args.output.open(mode="a", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=COLUMNS, extrasaction="ignore") if new_file: writer.writeheader() logger.info("Logging to %s every %.0fs", args.output, args.interval) while True: started = time.monotonic() sample = {"timestamp": datetime.now().isoformat(timespec="seconds")} try: if not session_key: session_key = login(opener=opener, base_url=base_url, timeout=args.timeout) sample.update( read_router(opener=opener, base_url=base_url, session_key=session_key, timeout=args.timeout) ) except Exception as exc: logger.warning("Router query failed: %s", exc) session_key = "" sample["ping_router_ms"] = ping(host=ROUTER_HOST) sample["ping_isp_gateway_ms"] = ping(host=sample.get("isp_gateway", "")) sample["ping_internet_ms"] = ping(host=args.internet_target) log_changes(previous=previous, current=sample) writer.writerow(sample) handle.flush() logger.info(describe(sample=sample)) previous = sample time.sleep(max(0.0, args.interval - (time.monotonic() - started))) if __name__ == "__main__": try: main() except KeyboardInterrupt: logger.info("Stopped")