Enhance modulation watchdog: severity based on QAM hierarchy

Modulation downgrades now trigger warning (small drop) or critical
(>=3 QAM levels, e.g. 256QAM to 16QAM). Upgrades remain info severity.
Messages distinguish 'dropped' vs 'improved'. Details include direction
field (downgrade/upgrade) and rank_drop per channel.

QAM hierarchy: QPSK/4QAM < 8QAM < 16QAM < 64QAM < 256QAM < 1024QAM < 4096QAM

Tests: 1 existing updated, 2 new added (175 total)

Author: Dennis Braun

app/event_detector.py

@@ -11,6 +11,23 @@
 SNR_CRIT_THRESHOLD = 25.0
 UNCORR_SPIKE_THRESHOLD = 1000
 
+# QAM hierarchy: higher value = better modulation
+QAM_ORDER = {
+    "QPSK": 1, "4QAM": 1,
+    "8QAM": 2,
+    "16QAM": 3,
+    "32QAM": 4,
+    "64QAM": 5,
+    "128QAM": 6,
+    "256QAM": 7,
+    "512QAM": 8,
+    "1024QAM": 9,
+    "2048QAM": 10,
+    "4096QAM": 11,
+}
+# A drop of this many levels or more counts as critical (e.g. 256QAM → 16QAM = 4 levels)
+QAM_CRITICAL_DROP = 3
+
 
 class EventDetector:
     """Compare consecutive analyses and emit event dicts."""
@@ -165,21 +182,46 @@ def _check_modulation(self, events, ts, cur_analysis, prev_analysis):
         cur_us = {ch["channel_id"]: ch.get("modulation", "") for ch in cur_analysis.get("us_channels", [])}
         prev_us = {ch["channel_id"]: ch.get("modulation", "") for ch in prev_analysis.get("us_channels", [])}
 
-        changed = []
+        downgrades = []
+        upgrades = []
         for ch_id in set(cur_ds) & set(prev_ds):
             if cur_ds[ch_id] != prev_ds[ch_id]:
-                changed.append({"channel": ch_id, "direction": "DS", "prev": prev_ds[ch_id], "current": cur_ds[ch_id]})
+                entry = {"channel": ch_id, "direction": "DS", "prev": prev_ds[ch_id], "current": cur_ds[ch_id]}
+                cur_rank = QAM_ORDER.get(cur_ds[ch_id], 0)
+                prev_rank = QAM_ORDER.get(prev_ds[ch_id], 0)
+                entry["rank_drop"] = prev_rank - cur_rank
+                if cur_rank < prev_rank:
+                    downgrades.append(entry)
+                else:
+                    upgrades.append(entry)
         for ch_id in set(cur_us) & set(prev_us):
             if cur_us[ch_id] != prev_us[ch_id]:
-                changed.append({"channel": ch_id, "direction": "US", "prev": prev_us[ch_id], "current": cur_us[ch_id]})
-
-        if changed:
+                entry = {"channel": ch_id, "direction": "US", "prev": prev_us[ch_id], "current": cur_us[ch_id]}
+                cur_rank = QAM_ORDER.get(cur_us[ch_id], 0)
+                prev_rank = QAM_ORDER.get(prev_us[ch_id], 0)
+                entry["rank_drop"] = prev_rank - cur_rank
+                if cur_rank < prev_rank:
+                    downgrades.append(entry)
+                else:
+                    upgrades.append(entry)
+
+        if downgrades:
+            max_drop = max(d["rank_drop"] for d in downgrades)
+            severity = "critical" if max_drop >= QAM_CRITICAL_DROP else "warning"
+            events.append({
+                "timestamp": ts,
+                "severity": severity,
+                "event_type": "modulation_change",
+                "message": f"Modulation dropped on {len(downgrades)} channel(s)",
+                "details": {"changes": downgrades, "direction": "downgrade"},
+            })
+        if upgrades:
             events.append({
                 "timestamp": ts,
                 "severity": "info",
                 "event_type": "modulation_change",
-                "message": f"Modulation changed on {len(changed)} channel(s)",
-                "details": {"changes": changed},
+                "message": f"Modulation improved on {len(upgrades)} channel(s)",
+                "details": {"changes": upgrades, "direction": "upgrade"},
             })
 
     def _check_errors(self, events, ts, cur, prev):

tests/test_events.py

@@ -235,7 +235,8 @@ def test_us_channel_count_change(self, detector):
         assert len(ch_events) == 1
         assert "US" in ch_events[0]["message"]
 
-    def test_modulation_change(self, detector):
+    def test_modulation_downgrade_warning(self, detector):
+        """256QAM → 64QAM = 2-level drop = warning"""
         ds1 = [{"channel_id": 1, "modulation": "256QAM", "power": 3.0, "snr": 35.0,
                 "correctable_errors": 10, "uncorrectable_errors": 5,
                 "docsis_version": "3.0", "health": "good", "health_detail": "", "frequency": "602 MHz"}]
@@ -246,7 +247,40 @@ def test_modulation_change(self, detector):
         events = detector.check(_make_analysis(ds_total=1, ds_channels=ds2))
         mod_events = [e for e in events if e["event_type"] == "modulation_change"]
         assert len(mod_events) == 1
+        assert mod_events[0]["severity"] == "warning"
+        assert "dropped" in mod_events[0]["message"]
+        assert mod_events[0]["details"]["direction"] == "downgrade"
+
+    def test_modulation_downgrade_critical(self, detector):
+        """256QAM → 16QAM = 4-level drop = critical"""
+        ds1 = [{"channel_id": 1, "modulation": "256QAM", "power": 3.0, "snr": 35.0,
+                "correctable_errors": 10, "uncorrectable_errors": 5,
+                "docsis_version": "3.0", "health": "good", "health_detail": "", "frequency": "602 MHz"}]
+        ds2 = [{"channel_id": 1, "modulation": "16QAM", "power": 3.0, "snr": 35.0,
+                "correctable_errors": 10, "uncorrectable_errors": 5,
+                "docsis_version": "3.0", "health": "good", "health_detail": "", "frequency": "602 MHz"}]
+        detector.check(_make_analysis(ds_total=1, ds_channels=ds1))
+        events = detector.check(_make_analysis(ds_total=1, ds_channels=ds2))
+        mod_events = [e for e in events if e["event_type"] == "modulation_change"]
+        assert len(mod_events) == 1
+        assert mod_events[0]["severity"] == "critical"
+        assert "dropped" in mod_events[0]["message"]
+
+    def test_modulation_upgrade_info(self, detector):
+        """64QAM → 256QAM = upgrade = info"""
+        ds1 = [{"channel_id": 1, "modulation": "64QAM", "power": 3.0, "snr": 35.0,
+                "correctable_errors": 10, "uncorrectable_errors": 5,
+                "docsis_version": "3.0", "health": "good", "health_detail": "", "frequency": "602 MHz"}]
+        ds2 = [{"channel_id": 1, "modulation": "256QAM", "power": 3.0, "snr": 35.0,
+                "correctable_errors": 10, "uncorrectable_errors": 5,
+                "docsis_version": "3.0", "health": "good", "health_detail": "", "frequency": "602 MHz"}]
+        detector.check(_make_analysis(ds_total=1, ds_channels=ds1))
+        events = detector.check(_make_analysis(ds_total=1, ds_channels=ds2))
+        mod_events = [e for e in events if e["event_type"] == "modulation_change"]
+        assert len(mod_events) == 1
         assert mod_events[0]["severity"] == "info"
+        assert "improved" in mod_events[0]["message"]
+        assert mod_events[0]["details"]["direction"] == "upgrade"
 
     def test_error_spike(self, detector):
         detector.check(_make_analysis(ds_uncorrectable_errors=100))