feat: implement apply_spike_suppression for health score (#141)

Author: itsDNNS

app/analyzer.py

@@ -9,6 +9,8 @@
 import os
 import re
 
+from .tz import utc_now, _parse_utc
+
 log = logging.getLogger("docsis.analyzer")
 
 # --- Dynamic thresholds (set by module loader) ---
@@ -132,6 +134,49 @@ def _get_spike_expiry_hours():
     return errors.get("spike_expiry_hours", 48)
 
 
+def apply_spike_suppression(analysis, last_spike_ts):
+    """Suppress uncorrectable error penalization if a past spike has expired.
+
+    Called as a post-processing step after analyze(). If the most recent
+    error_spike event is older than spike_expiry_hours and no new spike has
+    occurred since, the uncorrectable error percentage and related health
+    issues are suppressed.
+
+    Args:
+        analysis: dict from analyze() — modified in place
+        last_spike_ts: UTC timestamp string of latest error_spike, or None
+    """
+    if not last_spike_ts:
+        return
+
+    expiry_hours = _get_spike_expiry_hours()
+    now = _parse_utc(utc_now())
+    spike_dt = _parse_utc(last_spike_ts)
+    hours_since = (now - spike_dt).total_seconds() / 3600
+
+    if hours_since < expiry_hours:
+        return  # Still in observation period
+
+    summary = analysis["summary"]
+    summary["ds_uncorr_pct"] = 0.0
+    summary["health_issues"] = [i for i in summary["health_issues"] if "uncorr" not in i]
+    summary["spike_suppression"] = {
+        "active": True,
+        "last_spike": last_spike_ts,
+        "hours_since_spike": round(hours_since, 1),
+        "expiry_hours": expiry_hours,
+    }
+
+    # Recalculate health from remaining issues
+    issues = summary["health_issues"]
+    if not issues:
+        summary["health"] = "good"
+    elif any("critical" in i for i in issues):
+        summary["health"] = "poor"
+    else:
+        summary["health"] = "marginal"
+
+
 def _parse_qam_order(modulation_str):
     """Extract QAM order from modulation string. Returns None if unparseable."""
     if not modulation_str:

tests/test_analyzer.py

@@ -1,6 +1,7 @@
 """Tests for DOCSIS channel health analyzer."""
 
 import pytest
+from unittest.mock import patch
 from app import analyzer
 from app.analyzer import analyze, _parse_float, _parse_qam_order, _resolve_modulation, _channel_bitrate_mbps
 
@@ -771,3 +772,101 @@ def test_default_spike_expiry_hours(self):
         from app.analyzer import _get_spike_expiry_hours
         hours = _get_spike_expiry_hours()
         assert hours == 48
+
+
+class TestSpikeSuppression:
+    """Tests for apply_spike_suppression()."""
+
+    def _make_analysis_with_uncorr(self, uncorr_pct=86.6, health="poor",
+                                    extra_issues=None):
+        """Build a minimal analysis dict with uncorrectable error issues."""
+        issues = ["uncorr_errors_critical"]
+        if extra_issues:
+            issues.extend(extra_issues)
+        return {
+            "summary": {
+                "health": health,
+                "health_issues": issues,
+                "ds_uncorr_pct": uncorr_pct,
+                "ds_correctable_errors": 155000,
+                "ds_uncorrectable_errors": 1000000,
+                "ds_total": 33,
+                "us_total": 4,
+            },
+            "ds_channels": [],
+            "us_channels": [],
+        }
+
+    def test_no_spike_no_change(self):
+        """No spike timestamp — analysis stays unchanged."""
+        from app.analyzer import apply_spike_suppression
+        analysis = self._make_analysis_with_uncorr()
+        apply_spike_suppression(analysis, None)
+        assert analysis["summary"]["ds_uncorr_pct"] == 86.6
+        assert "uncorr_errors_critical" in analysis["summary"]["health_issues"]
+        assert analysis["summary"]["health"] == "poor"
+        assert "spike_suppression" not in analysis["summary"]
+
+    @patch("app.analyzer.utc_now")
+    def test_recent_spike_no_suppression(self, mock_now):
+        """Spike < 48h ago — still in observation period, no suppression."""
+        from app.analyzer import apply_spike_suppression
+        mock_now.return_value = "2026-02-28T12:00:00Z"
+        analysis = self._make_analysis_with_uncorr()
+        apply_spike_suppression(analysis, "2026-02-27T14:00:00Z")
+        assert analysis["summary"]["ds_uncorr_pct"] == 86.6
+        assert "uncorr_errors_critical" in analysis["summary"]["health_issues"]
+        assert analysis["summary"]["health"] == "poor"
+        assert "spike_suppression" not in analysis["summary"]
+
+    @patch("app.analyzer.utc_now")
+    def test_expired_spike_suppresses(self, mock_now):
+        """Spike >= 48h ago — suppression active."""
+        from app.analyzer import apply_spike_suppression
+        mock_now.return_value = "2026-03-01T15:00:00Z"  # 72.5h after spike
+        analysis = self._make_analysis_with_uncorr()
+        apply_spike_suppression(analysis, "2026-02-27T14:30:00Z")
+        assert analysis["summary"]["ds_uncorr_pct"] == 0.0
+        assert "uncorr_errors_critical" not in analysis["summary"]["health_issues"]
+        assert "uncorr_errors_high" not in analysis["summary"]["health_issues"]
+        assert analysis["summary"]["health"] == "good"
+        sup = analysis["summary"]["spike_suppression"]
+        assert sup["active"] is True
+        assert sup["last_spike"] == "2026-02-27T14:30:00Z"
+        assert sup["expiry_hours"] == 48
+
+    @patch("app.analyzer.utc_now")
+    def test_expired_spike_other_issues_remain(self, mock_now):
+        """Spike expired but other critical issues exist — health stays poor."""
+        from app.analyzer import apply_spike_suppression
+        mock_now.return_value = "2026-03-01T15:00:00Z"
+        analysis = self._make_analysis_with_uncorr(
+            extra_issues=["snr_critical"]
+        )
+        apply_spike_suppression(analysis, "2026-02-27T14:00:00Z")
+        assert analysis["summary"]["ds_uncorr_pct"] == 0.0
+        assert "uncorr_errors_critical" not in analysis["summary"]["health_issues"]
+        assert "snr_critical" in analysis["summary"]["health_issues"]
+        assert analysis["summary"]["health"] == "poor"
+        assert analysis["summary"]["spike_suppression"]["active"] is True
+
+    @patch("app.analyzer.utc_now")
+    def test_expired_spike_warning_issues_marginal(self, mock_now):
+        """Spike expired, only warning issues remain — health becomes marginal."""
+        from app.analyzer import apply_spike_suppression
+        mock_now.return_value = "2026-03-01T15:00:00Z"
+        analysis = self._make_analysis_with_uncorr(
+            extra_issues=["snr_warn"]
+        )
+        apply_spike_suppression(analysis, "2026-02-27T14:00:00Z")
+        assert analysis["summary"]["health"] == "marginal"
+
+    @patch("app.analyzer.utc_now")
+    def test_spike_at_exact_boundary(self, mock_now):
+        """Spike exactly 48h ago — suppressed (>= boundary)."""
+        from app.analyzer import apply_spike_suppression
+        mock_now.return_value = "2026-03-01T14:00:00Z"
+        analysis = self._make_analysis_with_uncorr()
+        apply_spike_suppression(analysis, "2026-02-27T14:00:00Z")
+        assert analysis["summary"]["ds_uncorr_pct"] == 0.0
+        assert analysis["summary"]["spike_suppression"]["active"] is True