Add the install and hardware analytics collectors

frigate/analytics/collectors/__init__.py

@@ -0,0 +1 @@
+"""One collector per report section."""

frigate/analytics/collectors/common.py

@@ -0,0 +1,37 @@
+"""Helpers the section collectors share."""
+
+import math
+from collections import Counter
+from enum import Enum
+from typing import Any, TypeVar
+
+E = TypeVar("E", bound=Enum)
+
+
+def closed(enum: type[E], value: Any, fallback: E) -> E:
+    """The member for a value, or the fallback for one the enum doesn't know."""
+    try:
+        return enum(str(value))
+    except ValueError:
+        return fallback
+
+
+def rate(value: Any) -> float:
+    """A finite, non-negative number rounded to 2 decimals, else 0.
+
+    A NaN would serialize as null and fail the schema's number type.
+    """
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return 0.0
+
+    if not math.isfinite(number):
+        return 0.0
+
+    return round(max(number, 0.0), 2)
+
+
+def histogram(counter: "Counter[E]") -> dict[E, int]:
+    """Drop the empty buckets, since an absent key means zero."""
+    return {key: total for key, total in counter.items() if total > 0}

frigate/analytics/collectors/hardware.py

@@ -0,0 +1,107 @@
+"""Hardware section: CPU, memory, GPUs, decode and detection hardware, storage."""
+
+import os
+from collections import Counter
+from typing import Any
+
+import psutil
+
+from frigate.analytics.collectors.common import closed, histogram
+from frigate.analytics.context import ReportContext
+from frigate.analytics.schema import (
+    DecodeFamily,
+    GpuInfo,
+    GpuVendor,
+    HardwareKey,
+    HardwareSection,
+    StorageInfo,
+)
+from frigate.const import RECORD_DIR
+from frigate.detectors.hardware import hardware_prober
+from frigate.util.hwaccel import hwaccel_options
+
+DEVICE_TREE_MODEL = "/proc/device-tree/model"
+CPUINFO = "/proc/cpuinfo"
+
+
+def cpu_model() -> str:
+    """The board model on ARM boards, else the CPU's model name."""
+    try:
+        with open(DEVICE_TREE_MODEL) as f:
+            board = f.read().strip("\x00\n ")
+
+        if board:
+            return board[:64]
+    except OSError:
+        pass
+
+    try:
+        with open(CPUINFO) as f:
+            for line in f:
+                key, _, value = line.partition(":")
+
+                if key.strip() == "model name" and value.strip():
+                    return value.strip()[:64]
+    except OSError:
+        pass
+
+    return "unknown"
+
+
+def gpus(stats: dict[str, Any]) -> list[GpuInfo]:
+    found: list[GpuInfo] = []
+
+    for name, entry in stats.get("gpu_usages", {}).items():
+        vendor = entry.get("vendor") if isinstance(entry, dict) else None
+        found.append(
+            GpuInfo(
+                vendor=closed(GpuVendor, vendor, GpuVendor.other),
+                name=str(name)[:64],
+            )
+        )
+
+    return found
+
+
+def decode_families() -> list[DecodeFamily]:
+    _, available = hwaccel_options()
+    families = [
+        closed(DecodeFamily, family.key, DecodeFamily.other) for family in available
+    ]
+    return list(dict.fromkeys(families))
+
+
+def detection_hardware() -> dict[HardwareKey, int]:
+    units: Counter[HardwareKey] = Counter()
+
+    for found in hardware_prober.probe():
+        units[closed(HardwareKey, found.key, HardwareKey.other)] += found.count
+
+    return histogram(units)
+
+
+def storage(stats: dict[str, Any]) -> StorageInfo:
+    # stats report sizes in MB
+    entry = stats.get("service", {}).get("storage", {}).get(RECORD_DIR) or {}
+    total_mb = float(entry.get("total") or 0)
+    used_mb = float(entry.get("used") or 0)
+
+    return StorageInfo(
+        record_fs=str(entry.get("mount_type") or "unknown")[:16],
+        record_total_gb=round(total_mb / 1024),
+        record_used_pct=min(round(used_mb / total_mb * 100), 100)
+        if total_mb > 0
+        else 0,
+    )
+
+
+def collect(ctx: ReportContext) -> HardwareSection:
+    return HardwareSection(
+        cpu_model=cpu_model(),
+        cpu_cores=os.cpu_count() or 0,
+        memory_gb=round(psutil.virtual_memory().total / 2**30),
+        gpus=gpus(ctx.stats),
+        decode_families=decode_families(),
+        detection_hardware=detection_hardware(),
+        storage=storage(ctx.stats),
+    )

frigate/analytics/collectors/install.py

@@ -0,0 +1,63 @@
+"""Install section: version, image variant, install type, platform."""
+
+import os
+import platform
+import re
+
+from frigate.analytics.collectors.common import closed
+from frigate.analytics.context import ReportContext
+from frigate.analytics.schema import Arch, ImageVariant, InstallSection, InstallType
+from frigate.version import VERSION
+
+KERNEL_PATTERN = re.compile(r"^(\d{1,3})\.(\d{1,3})")
+
+
+def image_variant(value: str | None) -> ImageVariant:
+    """The published image from the build-time FRIGATE_IMAGE_VARIANT, dev when unset."""
+    if not value:
+        return ImageVariant.dev
+
+    return closed(ImageVariant, value, ImageVariant.other)
+
+
+def install_type() -> InstallType:
+    # the add-on is a container too, so it has to be checked first
+    if os.path.isfile("/data/options.json"):
+        return InstallType.ha_addon
+
+    if os.environ.get("KUBERNETES_SERVICE_HOST"):
+        return InstallType.kubernetes
+
+    if os.path.exists("/run/.containerenv"):
+        return InstallType.podman
+
+    if os.path.exists("/.dockerenv"):
+        return InstallType.docker
+
+    return InstallType.unknown
+
+
+def arch(machine: str) -> Arch:
+    match machine.lower():
+        case "x86_64" | "amd64":
+            return Arch.x86_64
+        case "aarch64" | "arm64":
+            return Arch.aarch64
+        case _:
+            return Arch.other
+
+
+def kernel(release: str) -> str:
+    match = KERNEL_PATTERN.match(release)
+    return f"{match.group(1)}.{match.group(2)}" if match else "unknown"
+
+
+def collect(ctx: ReportContext) -> InstallSection:
+    return InstallSection(
+        version=VERSION[:32],
+        image_variant=image_variant(os.environ.get("FRIGATE_IMAGE_VARIANT")),
+        install_type=install_type(),
+        arch=arch(platform.machine()),
+        kernel=kernel(platform.release()),
+        run_as_root=os.geteuid() == 0,
+    )

frigate/analytics/context.py

@@ -0,0 +1,13 @@
+"""What the collectors read, gathered once per report."""
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from frigate.config import FrigateConfig
+
+
+@dataclass(frozen=True)
+class ReportContext:
+    config: FrigateConfig
+    stats: dict[str, Any]
+    notice_stats: list[dict[str, Any]] = field(default_factory=list)

frigate/test/analytics_helpers.py

@@ -0,0 +1,36 @@
+"""Shared fixtures for the analytics tests."""
+
+from typing import Any
+
+from frigate.analytics.context import ReportContext
+from frigate.config import FrigateConfig
+
+FRONT_CAMERA: dict[str, Any] = {
+    "ffmpeg": {"inputs": [{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}]},
+    "detect": {"height": 720, "width": 1280, "fps": 5},
+}
+
+
+def make_config(config: dict[str, Any] | None = None) -> FrigateConfig:
+    """A parsed config; top-level keys in config replace the defaults here.
+
+    hwaccel_args is set so parsing never probes the test host's GPU.
+    """
+    base: dict[str, Any] = {
+        "mqtt": {"host": "mqtt"},
+        "ffmpeg": {"hwaccel_args": []},
+        "cameras": {"front": FRONT_CAMERA},
+    }
+    return FrigateConfig(**{**base, **(config or {})})
+
+
+def make_context(
+    config: FrigateConfig | None = None,
+    stats: dict[str, Any] | None = None,
+    notice_stats: list[dict[str, Any]] | None = None,
+) -> ReportContext:
+    return ReportContext(
+        config=config or make_config(),
+        stats=stats or {},
+        notice_stats=notice_stats or [],
+    )

frigate/test/test_analytics_collectors_system.py

@@ -0,0 +1,188 @@
+"""Tests for the install and hardware collectors."""
+
+import os
+import tempfile
+import unittest
+from collections import Counter
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from frigate.analytics.collectors import hardware, install
+from frigate.analytics.collectors.common import closed, histogram, rate
+from frigate.analytics.schema import (
+    Arch,
+    DecodeFamily,
+    GpuVendor,
+    HardwareKey,
+    ImageVariant,
+    InstallType,
+)
+from frigate.const import RECORD_DIR
+from frigate.test.analytics_helpers import make_context
+
+
+class TestCommon(unittest.TestCase):
+    def test_closed_falls_back_for_unknown_values(self):
+        self.assertEqual(closed(Arch, "x86_64", Arch.other), Arch.x86_64)
+        self.assertEqual(closed(Arch, "sparc", Arch.other), Arch.other)
+        self.assertEqual(closed(Arch, None, Arch.other), Arch.other)
+
+    def test_rate_rounds_and_rejects_bad_numbers(self):
+        self.assertEqual(rate("12.346"), 12.35)
+        self.assertEqual(rate(-3), 0.0)
+        self.assertEqual(rate(float("nan")), 0.0)
+        self.assertEqual(rate(float("inf")), 0.0)
+        self.assertEqual(rate(None), 0.0)
+
+    def test_histogram_drops_empty_buckets(self):
+        self.assertEqual(
+            histogram(Counter({Arch.x86_64: 2, Arch.other: 0})), {"x86_64": 2}
+        )
+
+
+class TestInstallCollector(unittest.TestCase):
+    def test_collects_version_variant_and_platform(self):
+        with (
+            patch.dict(os.environ, {"FRIGATE_IMAGE_VARIANT": "tensorrt-jp6"}),
+            patch.object(install, "install_type", return_value=InstallType.docker),
+            patch("platform.machine", return_value="aarch64"),
+            patch("platform.release", return_value="6.8.0-45-generic"),
+            patch("os.geteuid", return_value=1000),
+        ):
+            section = install.collect(make_context())
+
+        self.assertEqual(section.image_variant, ImageVariant.tensorrt_jp6)
+        self.assertEqual(section.install_type, InstallType.docker)
+        self.assertEqual(section.arch, Arch.aarch64)
+        self.assertEqual(section.kernel, "6.8")
+        self.assertFalse(section.run_as_root)
+
+    def test_image_variant_is_dev_when_unset_and_other_when_unknown(self):
+        self.assertEqual(install.image_variant(None), ImageVariant.dev)
+        self.assertEqual(install.image_variant(""), ImageVariant.dev)
+        self.assertEqual(install.image_variant("h8l"), ImageVariant.other)
+
+    def test_install_type_checks_the_add_on_first(self):
+        with patch("os.path.isfile", return_value=True):
+            self.assertEqual(install.install_type(), InstallType.ha_addon)
+
+        with (
+            patch("os.path.isfile", return_value=False),
+            patch.dict(os.environ, {"KUBERNETES_SERVICE_HOST": "10.0.0.1"}),
+        ):
+            self.assertEqual(install.install_type(), InstallType.kubernetes)
+
+        env = {k: v for k, v in os.environ.items() if k != "KUBERNETES_SERVICE_HOST"}
+
+        with (
+            patch("os.path.isfile", return_value=False),
+            patch.dict(os.environ, env, clear=True),
+            patch("os.path.exists", side_effect=lambda path: path == "/.dockerenv"),
+        ):
+            self.assertEqual(install.install_type(), InstallType.docker)
+
+    def test_kernel_and_arch_fall_back(self):
+        self.assertEqual(install.kernel("weird"), "unknown")
+        self.assertEqual(install.arch("armv7l"), Arch.other)
+        self.assertEqual(install.arch("amd64"), Arch.x86_64)
+
+
+class TestHardwareCollector(unittest.TestCase):
+    def setUp(self):
+        self.dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.dir.cleanup)
+
+    def write(self, name: str, content: str) -> str:
+        path = os.path.join(self.dir.name, name)
+
+        with open(path, "w") as f:
+            f.write(content)
+
+        return path
+
+    def test_cpu_model_prefers_the_board_model(self):
+        board = self.write("model", "Raspberry Pi 5 Model B Rev 1.0\x00")
+        cpuinfo = self.write("cpuinfo", "model name\t: Cortex-A76\n")
+
+        with (
+            patch.object(hardware, "DEVICE_TREE_MODEL", board),
+            patch.object(hardware, "CPUINFO", cpuinfo),
+        ):
+            self.assertEqual(hardware.cpu_model(), "Raspberry Pi 5 Model B Rev 1.0")
+
+    def test_cpu_model_reads_cpuinfo_and_truncates(self):
+        cpuinfo = self.write("cpuinfo", "processor\t: 0\nmodel name\t: " + "x" * 80)
+
+        with (
+            patch.object(hardware, "DEVICE_TREE_MODEL", self.dir.name + "/none"),
+            patch.object(hardware, "CPUINFO", cpuinfo),
+        ):
+            self.assertEqual(hardware.cpu_model(), "x" * 64)
+
+    def test_collects_gpus_decode_detection_and_storage(self):
+        stats = {
+            "gpu_usages": {
+                "NVIDIA GeForce RTX 3060": {"vendor": "nvidia"},
+                "mystery": {},
+            },
+            "service": {
+                "storage": {
+                    RECORD_DIR: {
+                        "total": 2048000.0,
+                        "used": 512000.0,
+                        "mount_type": "nfs4",
+                    }
+                }
+            },
+        }
+        probed = [
+            SimpleNamespace(key="edgetpu:usb", count=2),
+            SimpleNamespace(key="new:thing", count=1),
+            SimpleNamespace(key="cpu", count=1),
+        ]
+        families = [
+            SimpleNamespace(key="vaapi"),
+            SimpleNamespace(key="intel-qsv"),
+            SimpleNamespace(key="future"),
+        ]
+
+        with (
+            patch.object(hardware.hardware_prober, "probe", return_value=probed),
+            patch.object(hardware, "hwaccel_options", return_value=("vaapi", families)),
+            patch.object(hardware, "cpu_model", return_value="Intel(R) N100"),
+            patch("os.cpu_count", return_value=4),
+            patch(
+                "psutil.virtual_memory", return_value=SimpleNamespace(total=16 * 2**30)
+            ),
+        ):
+            section = hardware.collect(make_context(stats=stats))
+
+        self.assertEqual(section.cpu_model, "Intel(R) N100")
+        self.assertEqual(section.cpu_cores, 4)
+        self.assertEqual(section.memory_gb, 16)
+        self.assertEqual(
+            [(gpu.vendor, gpu.name) for gpu in section.gpus],
+            [
+                (GpuVendor.nvidia, "NVIDIA GeForce RTX 3060"),
+                (GpuVendor.other, "mystery"),
+            ],
+        )
+        self.assertEqual(
+            section.decode_families,
+            [DecodeFamily.vaapi, DecodeFamily.intel_qsv, DecodeFamily.other],
+        )
+        self.assertEqual(
+            section.detection_hardware,
+            {HardwareKey.edgetpu_usb: 2, HardwareKey.other: 1, HardwareKey.cpu: 1},
+        )
+        self.assertEqual(section.storage.record_fs, "nfs4")
+        self.assertEqual(section.storage.record_total_gb, 2000)
+        self.assertEqual(section.storage.record_used_pct, 25)
+
+    def test_storage_without_stats_is_unknown_and_empty(self):
+        section = hardware.storage({})
+
+        self.assertEqual(
+            (section.record_fs, section.record_total_gb, section.record_used_pct),
+            ("unknown", 0, 0),
+        )