From 0d6d5405b376760019fed48d096d95488f824690 Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 03:37:01 +0300 Subject: [PATCH] watch: cycle scheduling, live dashboard and one time-series artifact per session --- src/netcheck/watch.py | 194 ++++++++++++++++++++++++++++++++++++++++++ tests/test_watch.py | 119 ++++++++++++++++++++++++++ 2 files changed, 313 insertions(+) create mode 100644 src/netcheck/watch.py create mode 100644 tests/test_watch.py diff --git a/src/netcheck/watch.py b/src/netcheck/watch.py new file mode 100644 index 0000000..a7fb556 --- /dev/null +++ b/src/netcheck/watch.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +import asyncio +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import httpx +from rich.console import Console +from rich.live import Live +from rich.table import Table + +from netcheck import __version__ +from netcheck.config import Settings +from netcheck.exporter import atomic_write, compact_timestamp, dump_json, sanitize_name, sparkline +from netcheck.ip_geo import gather_identity +from netcheck.models import PingResult, SpeedResult, to_jsonable +from netcheck.netinfo import detect_capabilities +from netcheck.orchestration import run_module, utc_now_iso +from netcheck.probes.latency import ping_fanout + +WATCH_SCHEMA_VERSION = 1 + + +def is_speedtest_cycle(cycle: int, every_n: int) -> bool: + # Cycle 1 measures speed so the session opens with a baseline to compare against. + if every_n <= 0 or cycle < 1: + return False + return (cycle - 1) % every_n == 0 + + +def next_delay(cycle_started: float, now: float, interval: float) -> float: + # Subtracting the work already done keeps the cycle cadence from drifting + # further behind on every tick. + return max(0.0, interval - (now - cycle_started)) + + +def summarize_cycle( + cycle: int, + at: str, + pings: list[PingResult], + speed: SpeedResult | None, +) -> dict[str, Any]: + return { + "cycle": cycle, + "at": at, + "hosts": { + p.label: { + "host": p.host, + "method": p.method, + "avg_ms": p.avg_ms, + "jitter_ms": p.jitter_ms, + "loss_pct": p.loss_pct, + } + for p in pings + }, + "speed": None + if speed is None + else { + "method": speed.method, + "download_mbps": speed.download_mbps, + "upload_mbps": speed.upload_mbps, + "bufferbloat_grade": speed.bufferbloat_grade, + }, + } + + +@dataclass +class WatchSession: + started_at: str + asn: str | None = None + interval_seconds: int = 60 + speedtest_every_n_cycles: int = 10 + cycles: list[dict] = field(default_factory=list) + + def add(self, summary: dict) -> None: + self.cycles.append(summary) + + def history(self, label: str) -> list[float | None]: + return [(cycle["hosts"].get(label) or {}).get("avg_ms") for cycle in self.cycles] + + def labels(self) -> list[str]: + seen: list[str] = [] + for cycle in self.cycles: + for label in cycle["hosts"]: + if label not in seen: + seen.append(label) + return seen + + def filename(self) -> str: + return f"watch_{sanitize_name(self.asn)}_{compact_timestamp(self.started_at)}.json" + + def to_report(self) -> dict[str, Any]: + return { + "schema_version": WATCH_SCHEMA_VERSION, + "kind": "watch", + "meta": { + "started_at": self.started_at, + "finished_at": utc_now_iso(), + "asn": self.asn, + "interval_seconds": self.interval_seconds, + "speedtest_every_n_cycles": self.speedtest_every_n_cycles, + "cycle_count": len(self.cycles), + }, + "cycles": to_jsonable(self.cycles), + } + + +def write_session(session: WatchSession, logs_dir: Path) -> Path: + return atomic_write(Path(logs_dir) / session.filename(), dump_json(session.to_report())) + + +def render_dashboard(session: WatchSession) -> Table: + table = Table(title=f"netcheck --watch · {session.asn or 'unknown ASN'} · cycle {len(session.cycles)}") + for column in ("Host", "Avg ms", "Jitter", "Loss", "Trend"): + table.add_column(column) + latest = session.cycles[-1]["hosts"] if session.cycles else {} + for label in session.labels(): + row = latest.get(label) or {} + table.add_row( + label, + "—" if row.get("avg_ms") is None else f"{row['avg_ms']:g}", + "—" if row.get("jitter_ms") is None else f"{row['jitter_ms']:g}", + f"{row.get('loss_pct', 0)}%", + sparkline(session.history(label)[-40:]), + ) + speed = next((c["speed"] for c in reversed(session.cycles) if c["speed"]), None) + if speed: + table.caption = ( + f"last speedtest: {speed['download_mbps']} / {speed['upload_mbps']} Mbps " + f"via {speed['method']} · bufferbloat {speed['bufferbloat_grade']}" + ) + return table + + +async def run_watch(settings: Settings, options) -> Path: + # Imported here, not at module import time: cli imports watch lazily for --watch, + # and a module-level import back into cli would close the cycle. + from netcheck.cli import _speed_section + + caps = detect_capabilities() + session = WatchSession( + started_at=utc_now_iso(), + interval_seconds=settings.watch.interval_seconds, + speedtest_every_n_cycles=settings.watch.speedtest_every_n_cycles, + ) + hosts = [(h.label, h.host) for h in settings.probing.reference_hosts] + if options.extra_host: + hosts.append(("target-host", options.extra_host)) + console = Console() + console.print(f"netcheck {__version__} · watching every {session.interval_seconds}s · Ctrl-C to stop") + + async with httpx.AsyncClient( + timeout=settings.timeouts.http_seconds, + follow_redirects=True, + headers={"User-Agent": f"netcheck/{__version__}"}, + ) as client: + try: + geo, _cf, _flags, _raw = await gather_identity(client, settings.providers) + session.asn = geo.asn + except (httpx.HTTPError, OSError): + session.asn = None + + cycle = 1 + try: + with Live(render_dashboard(session), console=console, refresh_per_second=settings.watch.dashboard_refresh_hz) as live: + while True: + began = time.perf_counter() + pings = await ping_fanout( + hosts, + caps, + settings.probing.quick_ping_count, + settings.probing.ping_interval_seconds, + settings.probing.ping_timeout_seconds, + ) + speed = None + if is_speedtest_cycle(cycle, session.speedtest_every_n_cycles) and not options.quick: + result = await run_module( + "speed", + _speed_section(client, settings, options, None), + timeout=settings.timeouts.speedtest_seconds, + ) + speed = result.data if result.status == "ok" else None + session.add(summarize_cycle(cycle, utc_now_iso(), pings, speed)) + live.update(render_dashboard(session)) + await asyncio.sleep(next_delay(began, time.perf_counter(), session.interval_seconds)) + cycle += 1 + except (KeyboardInterrupt, asyncio.CancelledError): + pass + + path = write_session(session, Path(settings.output.logs_dir)) + console.print(f"Watch session written to {path}") + return path diff --git a/tests/test_watch.py b/tests/test_watch.py new file mode 100644 index 0000000..a215318 --- /dev/null +++ b/tests/test_watch.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from netcheck.models import PingResult, SpeedResult +from netcheck.watch import ( + WATCH_SCHEMA_VERSION, + WatchSession, + is_speedtest_cycle, + next_delay, + summarize_cycle, + write_session, +) + + +def ping(label: str, avg: float | None, loss: float = 0.0) -> PingResult: + return PingResult( + label=label, + host="1.1.1.1", + resolved_ip="1.1.1.1", + method="icmp_win", + sent=5, + received=5 if avg is not None else 0, + loss_pct=loss, + avg_ms=avg, + jitter_ms=1.5, + ) + + +def test_the_first_cycle_measures_speed_so_the_session_has_a_baseline(): + assert is_speedtest_cycle(1, every_n=10) is True + + +def test_speed_is_measured_once_every_n_cycles(): + assert [c for c in range(1, 25) if is_speedtest_cycle(c, every_n=10)] == [1, 11, 21] + + +def test_an_interval_of_one_measures_speed_every_cycle(): + assert all(is_speedtest_cycle(c, every_n=1) for c in range(1, 6)) + + +def test_a_non_positive_interval_disables_the_speedtest_entirely(): + assert [c for c in range(1, 10) if is_speedtest_cycle(c, every_n=0)] == [] + assert is_speedtest_cycle(3, every_n=-5) is False + + +def test_cycle_numbers_below_one_are_never_speedtest_cycles(): + assert is_speedtest_cycle(0, every_n=10) is False + + +def test_next_delay_subtracts_the_time_the_cycle_already_took(): + assert next_delay(cycle_started=100.0, now=104.0, interval=60.0) == pytest.approx(56.0) + + +def test_a_cycle_that_overran_its_interval_does_not_sleep_negatively(): + assert next_delay(cycle_started=100.0, now=400.0, interval=60.0) == 0.0 + + +def test_summarize_cycle_flattens_the_hosts_and_omits_speed_when_it_did_not_run(): + summary = summarize_cycle(3, "2026-08-08T19:15:00Z", [ping("cloudflare-dns", 12.4)], None) + assert summary["cycle"] == 3 + assert summary["at"] == "2026-08-08T19:15:00Z" + assert summary["hosts"]["cloudflare-dns"]["avg_ms"] == 12.4 + assert summary["hosts"]["cloudflare-dns"]["loss_pct"] == 0.0 + assert summary["speed"] is None + + +def test_summarize_cycle_keeps_the_speed_figures_when_it_did_run(): + speed = SpeedResult(method="cloudflare", download_mbps=284.3, upload_mbps=41.7, bufferbloat_grade="B") + summary = summarize_cycle(1, "2026-08-08T19:12:00Z", [ping("cloudflare-dns", 12.4)], speed) + assert summary["speed"]["download_mbps"] == 284.3 + assert summary["speed"]["bufferbloat_grade"] == "B" + + +def test_summarize_cycle_records_a_dead_host_without_inventing_a_number(): + summary = summarize_cycle(2, "2026-08-08T19:13:00Z", [ping("google-dns", None, loss=100.0)], None) + assert summary["hosts"]["google-dns"]["avg_ms"] is None + assert summary["hosts"]["google-dns"]["loss_pct"] == 100.0 + + +def test_session_history_returns_the_series_for_one_host_with_gaps_preserved(): + session = WatchSession(started_at="2026-08-08T19:12:00Z", asn="AS64500") + session.add(summarize_cycle(1, "t1", [ping("cloudflare-dns", 12.0)], None)) + session.add(summarize_cycle(2, "t2", [ping("cloudflare-dns", None, loss=100.0)], None)) + session.add(summarize_cycle(3, "t3", [ping("cloudflare-dns", 14.0)], None)) + assert session.history("cloudflare-dns") == [12.0, None, 14.0] + assert session.history("absent-host") == [None, None, None] + + +def test_session_filename_marks_it_as_a_watch_artifact(): + session = WatchSession(started_at="2026-08-08T19:12:00Z", asn="AS64500") + assert session.filename() == "watch_AS64500_20260808T191200Z.json" + assert WatchSession(started_at="2026-08-08T19:12:00Z").filename().startswith("watch_unknown_") + + +def test_a_session_writes_exactly_one_artifact_holding_every_cycle(tmp_path: Path): + session = WatchSession( + started_at="2026-08-08T19:12:00Z", asn="AS64500", interval_seconds=60, speedtest_every_n_cycles=10 + ) + for cycle in range(1, 4): + session.add(summarize_cycle(cycle, f"t{cycle}", [ping("cloudflare-dns", 12.0 + cycle)], None)) + path = write_session(session, tmp_path) + assert list(tmp_path.iterdir()) == [path] + payload = json.loads(path.read_text(encoding="utf-8")) + assert payload["schema_version"] == WATCH_SCHEMA_VERSION + assert payload["kind"] == "watch" + assert payload["meta"]["interval_seconds"] == 60 + assert payload["meta"]["speedtest_every_n_cycles"] == 10 + assert [c["cycle"] for c in payload["cycles"]] == [1, 2, 3] + + +def test_an_empty_session_still_writes_a_well_formed_artifact(tmp_path: Path): + path = write_session(WatchSession(started_at="2026-08-08T19:12:00Z"), tmp_path) + payload = json.loads(path.read_text(encoding="utf-8")) + assert payload["cycles"] == [] + assert payload["meta"]["finished_at"]