playground/router_monitor.py

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