Add the health analytics collector

frigate/analytics/collectors/health.py

@@ -0,0 +1,59 @@
+"""Health section: uptime, CPU, enrichment speed, and notice counts."""
+
+from typing import Any
+
+from frigate.analytics.collectors.common import rate
+from frigate.analytics.context import ReportContext
+from frigate.analytics.schema import (
+    EnrichmentTiming,
+    HealthSection,
+    NoticeCounts,
+    NoticeKindKey,
+)
+
+TIMING_STATS = {
+    EnrichmentTiming.face: "face_recognition_speed",
+    EnrichmentTiming.lpr: "plate_recognition_speed",
+    EnrichmentTiming.plate_detection: "yolov9_plate_detection_speed",
+    EnrichmentTiming.image_embedding: "image_embedding_speed",
+    EnrichmentTiming.text_embedding: "text_embedding_speed",
+    EnrichmentTiming.review_description: "review_description_speed",
+    EnrichmentTiming.object_description: "object_description_speed",
+}
+REPORTABLE_KINDS = frozenset(key.value for key in NoticeKindKey)
+
+
+def notice_deltas(notice_stats: list[dict[str, Any]]) -> dict[Any, NoticeCounts]:
+    """What changed since the last accepted report, per reportable kind."""
+    deltas: dict[Any, NoticeCounts] = {}
+
+    for row in notice_stats:
+        if row["kind"] not in REPORTABLE_KINDS:
+            continue
+
+        occurrences = max(row["occurrences"] - row["reported_occurrences"], 0)
+        dismissals = max(row["dismissals"] - row["reported_dismissals"], 0)
+
+        if occurrences or dismissals:
+            deltas[NoticeKindKey(row["kind"])] = NoticeCounts(
+                occurrences=occurrences, dismissals=dismissals
+            )
+
+    return deltas
+
+
+def collect(ctx: ReportContext) -> HealthSection:
+    service = ctx.stats.get("service", {})
+    embeddings = ctx.stats.get("embeddings", {})
+    cpu = ctx.stats.get("cpu_usages", {}).get("frigate.full_system", {}).get("cpu")
+    timings = {
+        timing: rate(embeddings.get(key)) for timing, key in TIMING_STATS.items()
+    }
+
+    return HealthSection(
+        uptime_hours=int(rate(service.get("uptime")) // 3600),
+        cpu_percent=min(round(rate(cpu)), 100),
+        enrichment_ms={timing: value for timing, value in timings.items() if value > 0},
+        retention_unmet=bool(service.get("retention_unmet", False)),
+        notices=notice_deltas(ctx.notice_stats),
+    )

frigate/test/test_analytics_collectors_health.py

@@ -0,0 +1,65 @@
+"""Tests for the health collector."""
+
+import unittest
+
+from frigate.analytics.collectors import health
+from frigate.test.analytics_helpers import make_context
+
+
+class TestHealthCollector(unittest.TestCase):
+    def test_collects_uptime_cpu_timings_and_notice_deltas(self):
+        stats = {
+            "service": {"uptime": 7300, "retention_unmet": True},
+            "cpu_usages": {"frigate.full_system": {"cpu": "23.6", "mem": "40"}},
+            "embeddings": {
+                "face_recognition_speed": 41.234,
+                "text_embedding_speed": 0,
+                "devices": {},
+            },
+        }
+        notice_stats = [
+            {
+                "kind": "detector_stuck",
+                "occurrences": 5,
+                "dismissals": 1,
+                "reported_occurrences": 3,
+                "reported_dismissals": 1,
+            },
+            {
+                "kind": "shm_too_low",
+                "occurrences": 2,
+                "dismissals": 0,
+                "reported_occurrences": 2,
+                "reported_dismissals": 0,
+            },
+            {
+                "kind": "update_available",
+                "occurrences": 4,
+                "dismissals": 0,
+                "reported_occurrences": 0,
+                "reported_dismissals": 0,
+            },
+        ]
+
+        section = health.collect(make_context(stats=stats, notice_stats=notice_stats))
+
+        self.assertEqual(section.uptime_hours, 2)
+        self.assertEqual(section.cpu_percent, 24)
+        self.assertEqual(section.enrichment_ms, {"face": 41.23})
+        self.assertTrue(section.retention_unmet)
+        self.assertEqual(
+            {
+                kind: (c.occurrences, c.dismissals)
+                for kind, c in section.notices.items()
+            },
+            {"detector_stuck": (2, 0)},
+        )
+
+    def test_missing_stats_give_zeros(self):
+        section = health.collect(make_context())
+
+        self.assertEqual(
+            (section.uptime_hours, section.cpu_percent, section.enrichment_ms),
+            (0, 0, {}),
+        )
+        self.assertEqual(section.notices, {})