interpret: bufferbloat grading and overall verdict scoring

src/netcheck/interpret.py

@@ -2,8 +2,19 @@ from __future__ import annotations
 
 from typing import Iterable
 
-from netcheck.config import Band, Thresholds, VpnBands
-from netcheck.models import CfTrace, DnsLeak, Finding, IpGeo, LocalNet, PingResult, Signal, TraceResult, VpnAssessment
+from netcheck.config import Band, BufferbloatBands, Thresholds, VpnBands
+from netcheck.models import (
+    CfTrace,
+    DnsLeak,
+    Finding,
+    IpGeo,
+    LocalNet,
+    PingResult,
+    Signal,
+    SpeedResult,
+    TraceResult,
+    VpnAssessment,
+)
 from netcheck.netinfo import is_tunnel_iface, mtu_anomaly
 
 _SEVERITY_ORDER = {"ok": 0, "info": 1, "warn": 2, "crit": 3}
@@ -234,3 +245,78 @@ def assess_vpn(
         tunnel_iface=tunnel_iface,
         dns_leak=dns_leak,
     )
+
+
+_CONSEQUENCES = {
+    "A": "Video calls and games stay smooth while the line is fully loaded.",
+    "B": "Barely noticeable; a large upload may add a beat to a video call.",
+    "C": "Calls and games get choppy whenever something else is downloading.",
+    "D": "Any big transfer makes calls stutter and pages feel stuck.",
+    "E": "The connection feels broken while it is busy, even though bandwidth is fine.",
+    "F": "A single download makes a call unusable; this is queue bloat, not a slow line.",
+    "?": "Not measured — the speedtest tier that measures it did not run.",
+}
+_GRADE_SEVERITY = {"A": "ok", "B": "ok", "C": "warn", "D": "crit", "E": "crit", "F": "crit", "?": "info"}
+
+
+def grade_bufferbloat(delta_ms: float | None, bands: BufferbloatBands) -> str:
+    if delta_ms is None:
+        return "?"
+    delta = max(0.0, delta_ms)
+    for grade, ceiling in (("A", bands.a), ("B", bands.b), ("C", bands.c), ("D", bands.d), ("E", bands.e)):
+        if delta <= ceiling:
+            return grade
+    return "F"
+
+
+def bufferbloat_consequence(grade: str) -> str:
+    return _CONSEQUENCES.get(grade, _CONSEQUENCES["?"])
+
+
+def speed_findings(speed: SpeedResult, bands: BufferbloatBands) -> list[Finding]:
+    if speed.method == "none":
+        tried = ", ".join(a.tier for a in speed.tier_attempts) or "none"
+        return [
+            Finding(
+                id="speed.unavailable",
+                severity="info",
+                title="No bandwidth measurement",
+                detail=f"Every speedtest tier failed or was disabled (tried: {tried}).",
+                advice="Install the Ookla speedtest binary, or rerun without --quick.",
+            )
+        ]
+    findings: list[Finding] = []
+    for direction, delta in (("down", speed.bufferbloat_down_ms), ("up", speed.bufferbloat_up_ms)):
+        grade = grade_bufferbloat(delta, bands)
+        severity = _GRADE_SEVERITY[grade]
+        if severity in ("ok", "info"):
+            continue
+        findings.append(
+            Finding(
+                id=f"speed.bufferbloat_{direction}",
+                severity=severity,
+                title=f"Bufferbloat under load ({direction}stream): grade {grade}",
+                detail=f"Latency rose by {delta} ms while saturating the {direction}stream direction.",
+                metric=f"bufferbloat_{direction}_ms",
+                value=grade,
+                threshold=bands.c,
+                advice=bufferbloat_consequence(grade)
+                + " Enabling SQM/fq_codel on the router is the standard fix.",
+            )
+        )
+    return findings
+
+
+_SCORE_PENALTY = {"ok": 0, "info": 3, "warn": 10, "crit": 25}
+
+
+def overall_verdict(findings: list[Finding]) -> tuple[str, int, str]:
+    score = 100
+    for f in findings:
+        score -= _SCORE_PENALTY[f.severity]
+    score = max(0, score)
+    status = worst(f.severity for f in findings)
+    if status in ("ok", "info"):
+        return "ok", score, "No problems found on this connection."
+    headline = [f.title for f in findings if f.severity == status]
+    return status, score, "; ".join(headline[:3])

tests/test_interpret.py

@@ -283,3 +283,93 @@ def test_assess_vpn_returns_a_complete_assessment():
     assert assessment.tunnel_iface == "wg0"
     assert len(assessment.signals) == 2
     assert assessment.confidence >= 0.75
+
+
+from netcheck.config import BufferbloatBands
+from netcheck.models import Finding, SpeedResult
+from netcheck.interpret import bufferbloat_consequence, grade_bufferbloat, overall_verdict, speed_findings
+
+
+def test_bufferbloat_grades_follow_the_configured_bands():
+    b = BufferbloatBands()
+    assert grade_bufferbloat(0.0, b) == "A"
+    assert grade_bufferbloat(5.0, b) == "A"
+    assert grade_bufferbloat(5.1, b) == "B"
+    assert grade_bufferbloat(30.0, b) == "B"
+    assert grade_bufferbloat(60.0, b) == "C"
+    assert grade_bufferbloat(200.0, b) == "D"
+    assert grade_bufferbloat(400.0, b) == "E"
+    assert grade_bufferbloat(400.1, b) == "F"
+
+
+def test_bufferbloat_grade_of_an_unmeasured_delta_is_unknown():
+    assert grade_bufferbloat(None, BufferbloatBands()) == "?"
+
+
+def test_negative_delta_is_graded_a_not_crashed():
+    assert grade_bufferbloat(-3.0, BufferbloatBands()) == "A"
+
+
+def test_each_grade_has_a_plain_language_consequence():
+    for grade in "ABCDEF?":
+        assert bufferbloat_consequence(grade)
+    assert "call" in bufferbloat_consequence("F").lower()
+
+
+def test_speed_findings_report_a_bad_bufferbloat_grade():
+    speed = SpeedResult(method="cloudflare", download_mbps=300.0, upload_mbps=40.0, bufferbloat_down_ms=150.0)
+    findings = speed_findings(speed, BufferbloatBands())
+    bloat = [f for f in findings if f.id == "speed.bufferbloat_down"][0]
+    assert bloat.severity == "crit"
+    assert bloat.value == "D"
+    assert "call" in bloat.advice.lower() or "queue" in bloat.advice.lower()
+
+
+def test_speed_findings_are_silent_on_a_clean_line():
+    speed = SpeedResult(
+        method="ookla_bin",
+        download_mbps=300.0,
+        upload_mbps=40.0,
+        bufferbloat_down_ms=3.0,
+        bufferbloat_up_ms=4.0,
+    )
+    assert speed_findings(speed, BufferbloatBands()) == []
+
+
+def test_speed_findings_report_an_exhausted_cascade():
+    speed = SpeedResult(method="none")
+    findings = speed_findings(speed, BufferbloatBands())
+    assert [f.id for f in findings] == ["speed.unavailable"]
+    assert findings[0].severity == "info"
+
+
+def test_overall_verdict_is_healthy_with_no_findings():
+    status, score, summary = overall_verdict([])
+    assert status == "ok"
+    assert score == 100
+    assert "no problems" in summary.lower()
+
+
+def test_overall_verdict_degrades_with_severity():
+    warn = Finding(id="a", severity="warn", title="Jitter high", detail="")
+    crit = Finding(id="b", severity="crit", title="Loss on path", detail="")
+    info = Finding(id="c", severity="info", title="No speed data", detail="")
+
+    status, score, summary = overall_verdict([info])
+    assert (status, score) == ("ok", 97)
+
+    status, score, summary = overall_verdict([warn])
+    assert status == "warn"
+    assert score == 90
+
+    status, score, summary = overall_verdict([warn, crit])
+    assert status == "crit"
+    assert score == 65
+    assert "Loss on path" in summary
+
+
+def test_overall_score_never_goes_below_zero():
+    findings = [Finding(id=f"f{i}", severity="crit", title="x", detail="") for i in range(20)]
+    status, score, _ = overall_verdict(findings)
+    assert status == "crit"
+    assert score == 0