Build analytics reports from the section collectors

frigate/analytics/report.py

@@ -0,0 +1,88 @@
+"""Build a report from the section collectors."""
+
+import logging
+import time
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any
+from uuid import uuid4
+
+from pydantic import BaseModel
+
+from frigate.analytics.collectors import (
+    cameras,
+    detection,
+    features,
+    hardware,
+    health,
+    install,
+)
+from frigate.analytics.context import ReportContext
+from frigate.analytics.schema import SCHEMA_VERSION, AnalyticsReport
+from frigate.analytics.state import load_state
+from frigate.config import FrigateConfig
+
+if TYPE_CHECKING:
+    from frigate.notices.registry import NoticeRegistry
+    from frigate.stats.emitter import StatsEmitter
+
+logger = logging.getLogger(__name__)
+
+COLLECTORS: dict[str, Callable[[ReportContext], BaseModel | None]] = {
+    "install": install.collect,
+    "hardware": hardware.collect,
+    "detection": detection.collect,
+    "cameras": cameras.collect,
+    "features": features.collect,
+    "health": health.collect,
+}
+
+# shown in the preview until the first report creates a real ID
+PREVIEW_INSTALL_ID = "0" * 32
+
+
+def build_report(
+    ctx: ReportContext, install_id: str, sent_at: int | None = None
+) -> AnalyticsReport:
+    """Run every collector; one that fails sends its section as null."""
+    sections: dict[str, Any] = {}
+
+    for name, collect in COLLECTORS.items():
+        try:
+            sections[name] = collect(ctx)
+        except Exception:
+            # a collector bug must cost one section, never the whole report
+            logger.warning("Analytics %s section failed", name, exc_info=True)
+            sections[name] = None
+
+    return AnalyticsReport.model_validate(
+        {
+            "schema_version": SCHEMA_VERSION,
+            "install_id": install_id,
+            "report_id": str(uuid4()),
+            "sent_at": int(time.time()) if sent_at is None else sent_at,
+            **sections,
+        }
+    )
+
+
+def gather_context(
+    config: FrigateConfig,
+    stats_emitter: "StatsEmitter | None",
+    notice_registry: "NoticeRegistry | None",
+) -> ReportContext:
+    return ReportContext(
+        config=config,
+        stats=stats_emitter.get_latest_stats() if stats_emitter is not None else {},
+        notice_stats=notice_registry.stats() if notice_registry is not None else [],
+    )
+
+
+def preview_report(
+    config: FrigateConfig,
+    stats_emitter: "StatsEmitter | None",
+    notice_registry: "NoticeRegistry | None",
+) -> AnalyticsReport:
+    """The report the next send would carry, without sending it."""
+    state = load_state()
+    ctx = gather_context(config, stats_emitter, notice_registry)
+    return build_report(ctx, state.install_id if state else PREVIEW_INSTALL_ID)

frigate/test/test_analytics_report.py

@@ -0,0 +1,107 @@
+"""Tests for building a report from the collectors."""
+
+import json
+import logging
+import os
+import unittest
+from unittest.mock import Mock, patch
+
+from peewee_migrate import Router
+from playhouse.sqlite_ext import SqliteExtDatabase
+from playhouse.sqliteq import SqliteQueueDatabase
+
+from frigate.analytics import report
+from frigate.analytics.collectors import hardware
+from frigate.analytics.schema import AnalyticsReport, InstallSection
+from frigate.models import User
+from frigate.test.analytics_helpers import make_context
+from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS
+
+INSTALL = InstallSection(
+    version="0.19.0",
+    image_variant="dev",
+    install_type="docker",
+    arch="x86_64",
+    kernel="6.8",
+    run_as_root=False,
+)
+
+
+class TestBuildReport(unittest.TestCase):
+    def test_a_failing_collector_nulls_only_its_section(self):
+        collectors = {
+            "install": lambda ctx: INSTALL,
+            "hardware": Mock(side_effect=RuntimeError("boom")),
+            "detection": lambda ctx: None,
+            "cameras": lambda ctx: None,
+            "features": lambda ctx: None,
+            "health": lambda ctx: None,
+        }
+
+        with patch.dict(report.COLLECTORS, collectors):
+            built = report.build_report(make_context(), "a" * 32, sent_at=1790000000)
+
+        self.assertEqual(built.install, INSTALL)
+        self.assertIsNone(built.hardware)
+        self.assertEqual(built.schema_version, 1)
+        self.assertEqual(built.sent_at, 1790000000)
+        self.assertRegex(built.report_id, r"^[0-9a-f-]{36}$")
+
+
+class TestRealCollectors(unittest.TestCase):
+    def setUp(self):
+        migrate_db = SqliteExtDatabase(TEST_DB)
+        del logging.getLogger("peewee_migrate").handlers[:]
+        Router(migrate_db).run()
+        migrate_db.close()
+        self.db = SqliteQueueDatabase(TEST_DB)
+        self.db.bind([User])
+
+    def tearDown(self):
+        # close() leaves the queue's writer thread running against the deleted file
+        self.db.stop()
+
+        if not self.db.is_closed():
+            self.db.close()
+
+        for file in TEST_DB_CLEANUPS:
+            try:
+                os.remove(file)
+            except OSError:
+                pass
+
+    def test_every_section_builds_and_round_trips_through_json(self):
+        with (
+            patch.object(hardware.hardware_prober, "probe", return_value=[]),
+            patch.object(hardware, "hwaccel_options", return_value=("", [])),
+        ):
+            built = report.build_report(make_context(), "a" * 32)
+
+        body = built.model_dump_json()
+        parsed = json.loads(body)
+
+        for section in (
+            "install",
+            "hardware",
+            "detection",
+            "cameras",
+            "features",
+            "health",
+        ):
+            with self.subTest(section=section):
+                self.assertIsNotNone(parsed[section])
+
+        self.assertEqual(AnalyticsReport.model_validate_json(body), built)
+
+    def test_preview_uses_a_placeholder_id_before_the_first_report(self):
+        stats_emitter = Mock()
+        stats_emitter.get_latest_stats.return_value = {}
+
+        with (
+            patch.object(report, "load_state", return_value=None),
+            patch.object(hardware.hardware_prober, "probe", return_value=[]),
+            patch.object(hardware, "hwaccel_options", return_value=("", [])),
+        ):
+            built = report.preview_report(make_context().config, stats_emitter, None)
+
+        self.assertEqual(built.install_id, report.PREVIEW_INSTALL_ID)