watch: cycle scheduling, live dashboard and one time-series artifact per session

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

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"]