Refactor Hardware Stats (#24150)

* Refactor hardware stats to have consolidated ffmpeg, detector, and enrichments running.

* Cleanup hardware access that is not passed into the container

* Remove network stats from hardware refactor

frigate/stats/emitter.py

@@ -11,6 +11,7 @@ from typing import Any
 from frigate.comms.inter_process import InterProcessRequestor
 from frigate.config import FrigateConfig
 from frigate.const import FREQUENCY_STATS_POINTS
+from frigate.stats.hardware import HardwareStats
 from frigate.stats.prometheus import update_metrics
 from frigate.stats.util import stats_snapshot
 from frigate.types import StatsTrackingTypes
@@ -32,7 +33,7 @@ class StatsEmitter(threading.Thread):
         self.config = config
         self.stats_tracking = stats_tracking
         self.stop_event = stop_event
-        self.hwaccel_errors: dict[str, float] = {}
+        self.hardware_stats = HardwareStats(config)
         self.stats_history: list[dict[str, Any]] = []
 
         # create communication for stats
@@ -44,7 +45,7 @@ class StatsEmitter(threading.Thread):
             return self.stats_history[-1]
         else:
             stats = stats_snapshot(
-                self.config, self.stats_tracking, self.hwaccel_errors
+                self.config, self.stats_tracking, self.hardware_stats
             )
             self.stats_history.append(stats)
             return stats
@@ -134,7 +135,7 @@ class StatsEmitter(threading.Thread):
 
             logger.debug("Starting stats collection")
             stats = stats_snapshot(
-                self.config, self.stats_tracking, self.hwaccel_errors
+                self.config, self.stats_tracking, self.hardware_stats
             )
             self.stats_history.append(stats)
             self.stats_history = self.stats_history[-MAX_STATS_POINTS:]
@@ -144,4 +145,5 @@ class StatsEmitter(threading.Thread):
 
             logger.debug("Finished stats collection")
 
+        self.hardware_stats.stop()
         logger.info("Exiting stats emitter...")

frigate/stats/hardware.py

@@ -0,0 +1,472 @@
+"""Config driven hardware usage collection for stats."""
+
+import logging
+import os
+import time
+from collections.abc import Callable
+from concurrent.futures import Future, ThreadPoolExecutor, wait
+from dataclasses import dataclass, field
+from functools import cache, partial
+from glob import glob
+from typing import Any
+
+from frigate.config import FrigateConfig
+from frigate.config.camera.updater import (
+    CameraConfigUpdateEnum,
+    CameraConfigUpdateSubscriber,
+)
+from frigate.config.classification import SemanticSearchModelEnum
+from frigate.const import FFMPEG_HWACCEL_AMF, FFMPEG_HWACCEL_VULKAN
+from frigate.detectors.hardware import DEV_ROOT, hardware_prober
+from frigate.util.services import (
+    get_amd_gpu_stats,
+    get_axcl_npu_stats,
+    get_cpu_stats,
+    get_hailo_temps,
+    get_intel_gpu_stats,
+    get_jetson_stats,
+    get_nvidia_gpu_stats,
+    get_openvino_npu_stats,
+    get_rockchip_gpu_stats,
+    get_rockchip_npu_stats,
+    is_vaapi_amd_driver,
+)
+
+logger = logging.getLogger(__name__)
+
+HARDWARE_ERROR_COOLDOWN_SECONDS = 3600
+POLL_TIMEOUT_SECONDS = 10
+
+# hardware names polled here, in the order used to resolve a generic GPU demand
+GPU_HARDWARE_KEYS = {
+    "onnx:nvidia": "nvidia",
+    "tensorrt": "jetson",
+    "onnx:amd": "amd_gpu",
+    "openvino:GPU": "intel_gpu",
+    "rknn": "rockchip",
+}
+
+
+@dataclass
+class HardwarePollResult:
+    """Usage entries from one hardware poll, split by stats bucket."""
+
+    gpu: dict[str, dict[str, Any]] = field(default_factory=dict)
+    npu: dict[str, dict[str, Any]] = field(default_factory=dict)
+    ok: bool = True
+
+
+@cache
+def _is_amd_vaapi() -> bool:
+    """Cached vaapi driver check, the driver cannot change at runtime."""
+    return is_vaapi_amd_driver()
+
+
+def _present_hardware() -> set[str]:
+    return {hardware.key for hardware in hardware_prober.probe()}
+
+
+def _gpu_device_nodes_exist(name: str) -> bool:
+    """Whether the GPU's device nodes are visible to this container.
+
+    The prober reads /proc and /sys, which show the host kernel's hardware
+    even when a device was not passed into the container, so a GPU is only
+    usable when its device nodes exist too.
+    """
+    if name == "nvidia":
+        return bool(glob(f"{DEV_ROOT}/nvidia*"))
+    elif name in ("amd_gpu", "intel_gpu"):
+        return bool(glob(f"{DEV_ROOT}/dri/*"))
+
+    return True
+
+
+def _present_gpu() -> str | None:
+    """The hardware name of the first usable GPU found on this system, if any."""
+    present = _present_hardware()
+
+    for key, name in GPU_HARDWARE_KEYS.items():
+        if key in present and _gpu_device_nodes_exist(name):
+            return name
+
+    return None
+
+
+def _poll_nvidia(config: FrigateConfig) -> HardwarePollResult:
+    nvidia_usage = get_nvidia_gpu_stats()
+
+    if not nvidia_usage:
+        return HardwarePollResult(
+            gpu={"nvidia-gpu": {"vendor": "nvidia", "gpu": "", "mem": ""}}, ok=False
+        )
+
+    gpu: dict[str, dict[str, Any]] = {}
+
+    for entry in nvidia_usage.values():
+        gpu[entry["name"]] = {
+            "vendor": "nvidia",
+            "gpu": str(round(float(entry["gpu"]), 2)) + "%",
+            "mem": str(round(float(entry["mem"]), 2)) + "%",
+            "enc": str(round(float(entry["enc"]), 2)) + "%",
+            "dec": str(round(float(entry["dec"]), 2)) + "%",
+            "temp": str(entry["temp"]),
+        }
+
+    return HardwarePollResult(gpu=gpu)
+
+
+def _poll_jetson(config: FrigateConfig) -> HardwarePollResult:
+    jetson_usage = get_jetson_stats()
+
+    if not jetson_usage:
+        return HardwarePollResult(
+            gpu={"jetson-gpu": {"vendor": "nvidia", "gpu": "", "mem": ""}}, ok=False
+        )
+
+    return HardwarePollResult(gpu={"jetson-gpu": {"vendor": "nvidia", **jetson_usage}})
+
+
+def _poll_intel_gpu(config: FrigateConfig) -> HardwarePollResult:
+    intel_usage = get_intel_gpu_stats(config.telemetry.stats.intel_gpu_device)
+
+    if not intel_usage:
+        return HardwarePollResult(
+            gpu={"intel-gpu": {"vendor": "intel", "gpu": "", "mem": ""}}, ok=False
+        )
+
+    gpu: dict[str, dict[str, Any]] = {}
+
+    for entry in intel_usage.values():
+        name = entry.pop("name")
+        gpu[name] = entry
+
+    return HardwarePollResult(gpu=gpu)
+
+
+def _poll_amd_gpu(config: FrigateConfig) -> HardwarePollResult:
+    amd_usage = get_amd_gpu_stats()
+
+    if not amd_usage:
+        return HardwarePollResult(
+            gpu={"amd-vaapi": {"vendor": "amd", "gpu": "", "mem": ""}}, ok=False
+        )
+
+    return HardwarePollResult(gpu={"amd-vaapi": {"vendor": "amd", **amd_usage}})
+
+
+def _poll_rockchip(config: FrigateConfig) -> HardwarePollResult:
+    result = HardwarePollResult()
+    rga_usage = get_rockchip_gpu_stats()
+
+    if rga_usage:
+        result.gpu["rockchip"] = {"vendor": "rockchip", **rga_usage}
+
+    npu_usage = get_rockchip_npu_stats()
+
+    if npu_usage:
+        result.npu["rockchip"] = npu_usage
+
+    result.ok = bool(rga_usage or npu_usage)
+    return result
+
+
+def _poll_rpi(config: FrigateConfig) -> HardwarePollResult:
+    # RPi v4l2m2m is currently not able to get usage stats
+    return HardwarePollResult(
+        gpu={"rpi-v4l2m2m": {"vendor": "rpi", "gpu": "", "mem": ""}}
+    )
+
+
+def _poll_intel_npu(config: FrigateConfig) -> HardwarePollResult:
+    npu_usage = get_openvino_npu_stats()
+
+    if not npu_usage:
+        return HardwarePollResult(ok=False)
+
+    return HardwarePollResult(npu={"openvino": npu_usage})
+
+
+def _poll_axengine(config: FrigateConfig) -> HardwarePollResult:
+    npu_usage = get_axcl_npu_stats()
+
+    if not npu_usage:
+        return HardwarePollResult(ok=False)
+
+    return HardwarePollResult(npu={"axengine": npu_usage})
+
+
+POLLERS: dict[str, Callable[[FrigateConfig], HardwarePollResult]] = {
+    "nvidia": _poll_nvidia,
+    "jetson": _poll_jetson,
+    "intel_gpu": _poll_intel_gpu,
+    "amd_gpu": _poll_amd_gpu,
+    "rockchip": _poll_rockchip,
+    "rpi": _poll_rpi,
+    "intel_npu": _poll_intel_npu,
+    "axengine": _poll_axengine,
+}
+
+
+def _hwaccel_hardware(args: str) -> str | None:
+    """Map an ffmpeg hwaccel arg string (preset or raw args) to a hardware name."""
+    if "cuvid" in args or "nvidia" in args:
+        return "nvidia"
+    elif "nvmpi" in args or "jetson" in args:
+        return "jetson"
+    elif "qsv" in args:
+        return "intel_gpu"
+    elif FFMPEG_HWACCEL_AMF in args or "amf" in args:
+        return "amd_gpu"
+    elif "vaapi" in args:
+        return "amd_gpu" if _is_amd_vaapi() else "intel_gpu"
+    elif "preset-rk" in args or "rkmpp" in args:
+        return "rockchip"
+    elif "v4l2m2m" in args or "rpi" in args:
+        return "rpi"
+    elif FFMPEG_HWACCEL_VULKAN in args or "vulkan" in args:
+        # vulkan is vendor neutral, attribute it to whichever GPU is present
+        return _present_gpu()
+
+    return None
+
+
+class HardwareStats:
+    """Catalogs and polls all hardware relevant to the config."""
+
+    def __init__(self, config: FrigateConfig) -> None:
+        self.config = config
+        # hardware name -> bound poll function; the catalog of what to monitor
+        self._monitored: dict[str, Callable[[], HardwarePollResult]] = {}
+        self._errors: dict[str, float] = {}
+        self._executor = ThreadPoolExecutor(
+            max_workers=8, thread_name_prefix="hw_stats"
+        )
+        self._config_subscriber = CameraConfigUpdateSubscriber(
+            config,
+            config.cameras,
+            [
+                CameraConfigUpdateEnum.add,
+                CameraConfigUpdateEnum.remove,
+                CameraConfigUpdateEnum.ffmpeg,
+                CameraConfigUpdateEnum.audio_transcription,
+                CameraConfigUpdateEnum.semantic_search,
+                CameraConfigUpdateEnum.face_recognition,
+                CameraConfigUpdateEnum.lpr,
+            ],
+        )
+        self.update_config()
+
+    def update_config(self) -> None:
+        """Recalculate all hardware that needs to be monitored from the config."""
+        names = self._scan_ffmpeg() | self._scan_detectors() | self._scan_enrichments()
+        monitored: dict[str, Callable[[], HardwarePollResult]] = {}
+
+        for name in sorted(names):
+            if name == "intel_gpu" and not self.config.telemetry.stats.intel_gpu_stats:
+                continue
+
+            if name == "amd_gpu" and not self.config.telemetry.stats.amd_gpu_stats:
+                continue
+
+            monitored[name] = partial(POLLERS[name], self.config)
+
+        self._monitored = monitored
+        logger.debug("Monitoring hardware for stats: %s", list(monitored))
+
+    def update_stats(self, all_stats: dict[str, Any]) -> None:
+        """Poll the monitored hardware concurrently and fill usage stats."""
+        if self._config_subscriber.check_for_updates():
+            self.update_config()
+
+        now = time.monotonic()
+        hardware_futures: dict[Future[HardwarePollResult], str] = {}
+
+        for name, poll in self._monitored.items():
+            last_error = self._errors.get(name)
+
+            if last_error is not None:
+                if now - last_error < HARDWARE_ERROR_COOLDOWN_SECONDS:
+                    continue
+
+                self._errors.pop(name, None)
+
+            hardware_futures[self._executor.submit(poll)] = name
+
+        cpu_future = self._executor.submit(get_cpu_stats)
+        futures: list[Future[Any]] = [*hardware_futures, cpu_future]
+        done, _ = wait(futures, timeout=POLL_TIMEOUT_SECONDS)
+
+        gpu_usages: dict[str, dict[str, Any]] = {}
+        npu_usages: dict[str, dict[str, Any]] = {}
+
+        for future, name in hardware_futures.items():
+            if future not in done:
+                logger.warning("Timed out collecting %s stats", name)
+                self._errors[name] = time.monotonic()
+                continue
+
+            try:
+                result = future.result()
+            except Exception:
+                logger.exception("Failed to collect %s stats", name)
+                self._errors[name] = time.monotonic()
+                continue
+
+            gpu_usages.update(result.gpu)
+            npu_usages.update(result.npu)
+
+            if not result.ok:
+                self._errors[name] = time.monotonic()
+
+        if gpu_usages:
+            all_stats["gpu_usages"] = gpu_usages
+
+        if npu_usages:
+            all_stats["npu_usages"] = npu_usages
+
+        if cpu_future in done:
+            try:
+                cpu_stats = cpu_future.result()
+            except Exception:
+                logger.exception("Failed to collect cpu stats")
+            else:
+                if cpu_stats:
+                    all_stats["cpu_usages"] = cpu_stats
+
+    def stop(self) -> None:
+        self._config_subscriber.stop()
+        self._executor.shutdown(wait=False, cancel_futures=True)
+
+    def _scan_ffmpeg(self) -> set[str]:
+        """Hardware used by camera hwaccel args (presets or raw ffmpeg args)."""
+        hwaccel_args: list[str] = []
+
+        for camera in self.config.cameras.values():
+            args = camera.ffmpeg.hwaccel_args
+
+            if isinstance(args, list):
+                args = " ".join(args)
+
+            if args and args not in hwaccel_args:
+                hwaccel_args.append(args)
+
+            for stream_input in camera.ffmpeg.inputs:
+                args = stream_input.hwaccel_args
+
+                if isinstance(args, list):
+                    args = " ".join(args)
+
+                if args and args not in hwaccel_args:
+                    hwaccel_args.append(args)
+
+        names: set[str] = set()
+
+        for args in hwaccel_args:
+            name = _hwaccel_hardware(args)
+
+            if name is not None:
+                names.add(name)
+
+        return names
+
+    def _scan_detectors(self) -> set[str]:
+        """Hardware used by the configured detection models."""
+        names: set[str] = set()
+
+        for model in self.config.models:
+            for spec in self.config.devices_for_model(model):
+                if spec.detector == "rknn":
+                    names.add("rockchip")
+                elif spec.detector == "axengine":
+                    names.add("axengine")
+                elif spec.detector == "tensorrt":
+                    names.add("jetson")
+                elif spec.detector == "openvino":
+                    if spec.device == "NPU":
+                        names.add("intel_npu")
+                    elif spec.device is None or spec.device.startswith("GPU"):
+                        names.add("intel_gpu")
+                elif spec.detector == "onnx":
+                    gpu = _present_gpu()
+
+                    if gpu is not None:
+                        names.add(gpu)
+
+        return names
+
+    def _scan_enrichments(self) -> set[str]:
+        """Hardware used by enabled enrichments, resolved to the present GPU."""
+        config = self.config
+        names: set[str] = set()
+        gpu = _present_gpu()
+
+        if gpu is not None:
+            semantic = config.semantic_search
+
+            if (
+                semantic.enabled
+                # GenAI providers run remotely and use no local hardware
+                and (
+                    semantic.model is None
+                    or isinstance(semantic.model, SemanticSearchModelEnum)
+                )
+                and (
+                    semantic.device
+                    or ("GPU" if semantic.model_size == "large" else "CPU")
+                )
+                != "CPU"
+            ):
+                names.add(gpu)
+
+            if (
+                config.face_recognition.enabled
+                and (config.face_recognition.device or "GPU") != "CPU"
+            ):
+                names.add(gpu)
+
+            if config.lpr.enabled and (config.lpr.device or "AUTO") != "CPU":
+                names.add(gpu)
+
+        # audio transcription runs on CUDA only
+        transcription_configs = [config.audio_transcription] + [
+            camera.audio_transcription for camera in config.cameras.values()
+        ]
+
+        if (
+            any(c.enabled and c.device == "GPU" for c in transcription_configs)
+            and "onnx:nvidia" in _present_hardware()
+            and _gpu_device_nodes_exist("nvidia")
+        ):
+            names.add("nvidia")
+
+        return names
+
+
+def read_temperature(path: str) -> float | None:
+    """Read a sysfs temperature file, converting millidegrees to degrees.
+
+    Returns None when the file does not exist.
+    """
+    if os.path.isfile(path):
+        with open(path) as f:
+            line = f.readline().strip()
+            return int(line) / 1000
+    return None
+
+
+def get_hardware_temperatures(detector_type: str) -> list[float | None]:
+    """Per unit temperatures for a detector type, in stable device order."""
+    if detector_type == "edgetpu":
+        # PCIe Corals expose a temperature through the apex driver
+        base = "/sys/class/apex/"
+
+        if os.path.isdir(base):
+            return [
+                read_temperature(os.path.join(base, apex, "temp"))
+                for apex in sorted(os.listdir(base))
+            ]
+    elif detector_type == "hailo8l":
+        hailo_temps = get_hailo_temps()
+        return [hailo_temps[name] for name in sorted(hailo_temps.keys())]
+
+    return []

frigate/stats/util.py

@@ -1,8 +1,6 @@
 """Utilities for stats."""
 
-import asyncio
 import logging
-import os
 import shutil
 import time
 from json import JSONDecodeError
@@ -16,29 +14,17 @@ from frigate.config import FrigateConfig
 from frigate.const import CACHE_DIR, CLIPS_DIR, RECORD_DIR
 from frigate.data_processing.types import DataProcessorMetrics
 from frigate.object_detection.base import ObjectDetectProcess
+from frigate.stats.hardware import HardwareStats, get_hardware_temperatures
 from frigate.types import StatsTrackingTypes
 from frigate.util.services import (
     calculate_shm_requirements,
-    get_amd_gpu_stats,
-    get_axcl_npu_stats,
     get_bandwidth_stats,
-    get_cpu_stats,
     get_fs_type,
-    get_hailo_temps,
-    get_intel_gpu_stats,
-    get_jetson_stats,
-    get_nvidia_gpu_stats,
-    get_openvino_npu_stats,
-    get_rockchip_gpu_stats,
-    get_rockchip_npu_stats,
-    is_vaapi_amd_driver,
 )
 from frigate.version import VERSION
 
 logger = logging.getLogger(__name__)
 
-HWACCEL_ERROR_COOLDOWN_SECONDS = 3600
-
 
 def get_latest_version(config: FrigateConfig) -> str:
     if not config.telemetry.version_check:
@@ -78,64 +64,6 @@ def stats_init(
     return stats_tracking
 
 
-def read_temperature(path: str) -> float | None:
-    if os.path.isfile(path):
-        with open(path) as f:
-            line = f.readline().strip()
-            return int(line) / 1000
-    return None
-
-
-def get_temperatures() -> dict[str, float]:
-    temps = {}
-
-    # Get temperatures for all attached Corals
-    base = "/sys/class/apex/"
-    if os.path.isdir(base):
-        for apex in os.listdir(base):
-            temp = read_temperature(os.path.join(base, apex, "temp"))
-            if temp is not None:
-                temps[apex] = temp
-
-    # Get temperatures for Hailo devices
-    temps.update(get_hailo_temps())
-
-    return temps
-
-
-def get_detector_temperature(
-    detector_type: str,
-    detector_index_by_type: dict[str, int],
-) -> float | None:
-    """Get temperature for a specific detector based on its type."""
-    if detector_type == "edgetpu":
-        # Get temperatures for all attached Corals
-        base = "/sys/class/apex/"
-        if os.path.isdir(base):
-            apex_devices = sorted(os.listdir(base))
-            index = detector_index_by_type.get("edgetpu", 0)
-            if index < len(apex_devices):
-                apex_name = apex_devices[index]
-                temp = read_temperature(os.path.join(base, apex_name, "temp"))
-                if temp is not None:
-                    return temp
-    elif detector_type == "hailo8l":
-        # Get temperatures for Hailo devices
-        hailo_temps = get_hailo_temps()
-        if hailo_temps:
-            hailo_device_names = sorted(hailo_temps.keys())
-            index = detector_index_by_type.get("hailo8l", 0)
-            if index < len(hailo_device_names):
-                device_name = hailo_device_names[index]
-                return hailo_temps[device_name]
-    elif detector_type == "rknn":
-        # Rockchip temperatures are handled by the GPU / NPU stats
-        # as there are not detector specific temperatures
-        pass
-
-    return None
-
-
 def get_detector_stats(
     stats_tracking: StatsTrackingTypes,
 ) -> dict[str, dict[str, Any]]:
@@ -161,190 +89,20 @@ def get_detector_stats(
             "pid": pid,
         }
 
-        temp = get_detector_temperature(detector_type, {detector_type: current_index})
+        temps = get_hardware_temperatures(detector_type)
 
-        if temp is not None:
-            detector_stat["temperature"] = round(temp, 1)
+        if current_index < len(temps) and temps[current_index] is not None:
+            detector_stat["temperature"] = round(temps[current_index], 1)
 
         detector_stats[name] = detector_stat
 
     return detector_stats
 
 
-def get_processing_stats(
-    config: FrigateConfig,
-    stats: dict[str, str],
-    hwaccel_errors: dict[str, float],
-) -> None:
-    """Get stats for cpu / gpu."""
-
-    async def run_tasks() -> None:
-        stats_tasks = [
-            asyncio.create_task(set_gpu_stats(config, stats, hwaccel_errors)),
-            asyncio.create_task(set_cpu_stats(stats)),
-            asyncio.create_task(set_npu_usages(config, stats)),
-        ]
-
-        if config.telemetry.stats.network_bandwidth:
-            stats_tasks.append(asyncio.create_task(set_bandwidth_stats(config, stats)))
-
-        await asyncio.wait(stats_tasks)
-
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-    loop.run_until_complete(run_tasks())
-    loop.close()
-
-
-async def set_cpu_stats(all_stats: dict[str, Any]) -> None:
-    """Set cpu usage from top."""
-    cpu_stats = get_cpu_stats()
-
-    if cpu_stats:
-        all_stats["cpu_usages"] = cpu_stats
-
-
-async def set_bandwidth_stats(config: FrigateConfig, all_stats: dict[str, Any]) -> None:
-    """Set bandwidth from nethogs."""
-    bandwidth_stats = get_bandwidth_stats(config)
-
-    if bandwidth_stats:
-        all_stats["bandwidth_usages"] = bandwidth_stats
-
-
-async def set_gpu_stats(
-    config: FrigateConfig,
-    all_stats: dict[str, Any],
-    hwaccel_errors: dict[str, float],
-) -> None:
-    """Parse GPUs from hwaccel args and use for stats."""
-    hwaccel_args = []
-
-    for camera in config.cameras.values():
-        args = camera.ffmpeg.hwaccel_args
-
-        if isinstance(args, list):
-            args = " ".join(args)
-
-        if args and args not in hwaccel_args:
-            hwaccel_args.append(args)
-
-        for stream_input in camera.ffmpeg.inputs:
-            args = stream_input.hwaccel_args
-
-            if isinstance(args, list):
-                args = " ".join(args)
-
-            if args and args not in hwaccel_args:
-                hwaccel_args.append(args)
-
-    stats: dict[str, dict] = {}
-    intel_gpu_collected = False
-    now = time.monotonic()
-
-    for args in hwaccel_args:
-        last_error = hwaccel_errors.get(args)
-        if last_error is not None:
-            if now - last_error < HWACCEL_ERROR_COOLDOWN_SECONDS:
-                continue
-            hwaccel_errors.pop(args, None)
-
-        if "cuvid" in args or "nvidia" in args:
-            # nvidia GPU
-            nvidia_usage = get_nvidia_gpu_stats()
-
-            if nvidia_usage:
-                for i in range(len(nvidia_usage)):
-                    stats[nvidia_usage[i]["name"]] = {
-                        "vendor": "nvidia",
-                        "gpu": str(round(float(nvidia_usage[i]["gpu"]), 2)) + "%",
-                        "mem": str(round(float(nvidia_usage[i]["mem"]), 2)) + "%",
-                        "enc": str(round(float(nvidia_usage[i]["enc"]), 2)) + "%",
-                        "dec": str(round(float(nvidia_usage[i]["dec"]), 2)) + "%",
-                        "temp": str(nvidia_usage[i]["temp"]),
-                    }
-
-            else:
-                stats["nvidia-gpu"] = {"vendor": "nvidia", "gpu": "", "mem": ""}
-                hwaccel_errors[args] = time.monotonic()
-        elif "nvmpi" in args or "jetson" in args:
-            # nvidia Jetson
-            jetson_usage = get_jetson_stats()
-
-            if jetson_usage:
-                stats["jetson-gpu"] = {"vendor": "nvidia", **jetson_usage}
-            else:
-                stats["jetson-gpu"] = {"vendor": "nvidia", "gpu": "", "mem": ""}
-                hwaccel_errors[args] = time.monotonic()
-        elif "qsv" in args or ("vaapi" in args and not is_vaapi_amd_driver()):
-            if not config.telemetry.stats.intel_gpu_stats:
-                continue
-
-            if not intel_gpu_collected:
-                # intel GPU (QSV or VAAPI both use the same physical GPU)
-                intel_gpu_collected = True
-                intel_usage = get_intel_gpu_stats(
-                    config.telemetry.stats.intel_gpu_device
-                )
-
-                if intel_usage:
-                    for entry in intel_usage.values():
-                        name = entry.pop("name")
-                        stats[name] = entry
-                else:
-                    stats["intel-gpu"] = {"vendor": "intel", "gpu": "", "mem": ""}
-                    hwaccel_errors[args] = time.monotonic()
-        elif "vaapi" in args:
-            if not config.telemetry.stats.amd_gpu_stats:
-                continue
-
-            # AMD VAAPI GPU
-            amd_usage = get_amd_gpu_stats()
-
-            if amd_usage:
-                stats["amd-vaapi"] = {"vendor": "amd", **amd_usage}
-            else:
-                stats["amd-vaapi"] = {"vendor": "amd", "gpu": "", "mem": ""}
-                hwaccel_errors[args] = time.monotonic()
-        elif "preset-rk" in args:
-            rga_usage = get_rockchip_gpu_stats()
-
-            if rga_usage:
-                stats["rockchip"] = {"vendor": "rockchip", **rga_usage}
-        elif "v4l2m2m" in args or "rpi" in args:
-            # RPi v4l2m2m is currently not able to get usage stats
-            stats["rpi-v4l2m2m"] = {"vendor": "rpi", "gpu": "", "mem": ""}
-
-    if stats:
-        all_stats["gpu_usages"] = stats
-
-
-async def set_npu_usages(config: FrigateConfig, all_stats: dict[str, Any]) -> None:
-    stats: dict[str, dict] = {}
-
-    for model in config.models:
-        for device in config.devices_for_model(model):
-            if device.detector == "rknn":
-                # Rockchip NPU usage
-                rk_usage = get_rockchip_npu_stats()
-                stats["rockchip"] = rk_usage
-            elif device.detector == "openvino" and device.device == "NPU":
-                # OpenVINO NPU usage
-                ov_usage = get_openvino_npu_stats()
-                stats["openvino"] = ov_usage
-            elif device.detector == "axengine":
-                # AXERA NPU usage
-                axcl_usage = get_axcl_npu_stats()
-                stats["axengine"] = axcl_usage
-
-    if stats:
-        all_stats["npu_usages"] = stats
-
-
 def stats_snapshot(
     config: FrigateConfig,
     stats_tracking: StatsTrackingTypes,
-    hwaccel_errors: dict[str, float],
+    hardware_stats: HardwareStats,
 ) -> dict[str, Any]:
     """Get a snapshot of the current stats that are being tracked."""
     camera_metrics = stats_tracking["camera_metrics"]
@@ -487,7 +245,13 @@ def stats_snapshot(
                 embeddings_metrics.classification_cps[key].value, 2
             )
 
-    get_processing_stats(config, stats, hwaccel_errors)
+    hardware_stats.update_stats(stats)
+
+    if config.telemetry.stats.network_bandwidth:
+        bandwidth_stats = get_bandwidth_stats(config)
+
+        if bandwidth_stats:
+            stats["bandwidth_usages"] = bandwidth_stats
 
     stats["service"] = {
         "uptime": (int(time.time()) - stats_tracking["started"]),

frigate/test/test_hardware_stats.py

@@ -0,0 +1,418 @@
+"""Tests for config driven hardware stats collection."""
+
+import time
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from frigate.config import FrigateConfig
+from frigate.detectors.device import DeviceSpec
+from frigate.detectors.hardware import DetectionHardware
+from frigate.stats import hardware
+from frigate.stats.hardware import (
+    HardwarePollResult,
+    HardwareStats,
+    _gpu_device_nodes_exist,
+    _hwaccel_hardware,
+    get_hardware_temperatures,
+)
+
+
+def found(key: str) -> DetectionHardware:
+    """A probe result carrying only the fields the scanners read."""
+    return DetectionHardware(
+        key=key,
+        detector=key.partition(":")[0],
+        name=key,
+        units=[],
+        count=0,
+        unlimited=True,
+    )
+
+
+class HardwareStatsTestCase(unittest.TestCase):
+    def setUp(self):
+        subscriber = patch("frigate.stats.hardware.CameraConfigUpdateSubscriber")
+        self.subscriber = subscriber.start()
+        self.addCleanup(subscriber.stop)
+        self.subscriber.return_value.check_for_updates.return_value = {}
+
+        # probed hardware is treated as passed into the container by default
+        device_nodes = patch(
+            "frigate.stats.hardware._gpu_device_nodes_exist", return_value=True
+        )
+        self.device_nodes = device_nodes.start()
+        self.addCleanup(device_nodes.stop)
+
+        hardware._is_amd_vaapi.cache_clear()
+        self.addCleanup(hardware._is_amd_vaapi.cache_clear)
+
+    def make_config(self, hwaccel_args: str = "", **overrides) -> FrigateConfig:
+        config = {
+            "mqtt": {"host": "mqtt"},
+            "cameras": {
+                "back": {
+                    "ffmpeg": {
+                        "hwaccel_args": hwaccel_args,
+                        "inputs": [
+                            {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}
+                        ],
+                    },
+                },
+            },
+            **overrides,
+        }
+
+        with patch("frigate.config.config.auto_detect_hwaccel", return_value=""):
+            return FrigateConfig(**config)
+
+    def make_stats(
+        self, config: FrigateConfig | None = None, present: tuple[str, ...] = ()
+    ) -> HardwareStats:
+        with patch.object(
+            hardware.hardware_prober,
+            "probe",
+            return_value=[found(key) for key in present],
+        ):
+            stats = HardwareStats(config or self.make_config())
+
+        self.addCleanup(stats.stop)
+        return stats
+
+
+class TestHwaccelHardware(HardwareStatsTestCase):
+    def test_preset_and_raw_args(self):
+        cases = {
+            "preset-nvidia": "nvidia",
+            "-c:v h264_cuvid": "nvidia",
+            "preset-jetson-h264": "jetson",
+            "-c:v h264_nvmpi": "jetson",
+            "preset-intel-qsv-h264": "intel_gpu",
+            "-hwaccel qsv": "intel_gpu",
+            "preset-amd-amf": "amd_gpu",
+            "preset-rk-h264": "rockchip",
+            "preset-rkmpp": "rockchip",
+            "preset-rpi-64-h264": "rpi",
+            "-c:v h264_v4l2m2m": "rpi",
+            "": None,
+            "-hwaccel none": None,
+        }
+
+        for args, expected in cases.items():
+            self.assertEqual(_hwaccel_hardware(args), expected, args)
+
+    def test_vaapi_driver_disambiguation(self):
+        with patch("frigate.stats.hardware.is_vaapi_amd_driver", return_value=True):
+            self.assertEqual(_hwaccel_hardware("preset-vaapi"), "amd_gpu")
+
+        hardware._is_amd_vaapi.cache_clear()
+
+        with patch("frigate.stats.hardware.is_vaapi_amd_driver", return_value=False):
+            self.assertEqual(_hwaccel_hardware("-hwaccel vaapi"), "intel_gpu")
+
+    def test_vulkan_resolves_to_present_gpu(self):
+        with patch.object(
+            hardware.hardware_prober,
+            "probe",
+            return_value=[found("cpu"), found("onnx:amd")],
+        ):
+            self.assertEqual(_hwaccel_hardware("preset-vulkan"), "amd_gpu")
+
+        with patch.object(
+            hardware.hardware_prober, "probe", return_value=[found("cpu")]
+        ):
+            self.assertIsNone(_hwaccel_hardware("preset-vulkan"))
+
+
+class TestGpuDeviceNodes(HardwareStatsTestCase):
+    def test_host_gpu_without_device_nodes_is_skipped(self):
+        # /proc and /sys show the host's nvidia card even when it is not
+        # passed into the container, so resolution falls to the intel GPU
+        self.device_nodes.side_effect = lambda name: name != "nvidia"
+
+        with patch.object(
+            hardware.hardware_prober,
+            "probe",
+            return_value=[found("onnx:nvidia"), found("openvino:GPU")],
+        ):
+            self.assertEqual(hardware._present_gpu(), "intel_gpu")
+
+    def test_device_node_globs(self):
+        with patch("frigate.stats.hardware.glob", return_value=[]):
+            self.assertFalse(_gpu_device_nodes_exist("nvidia"))
+            self.assertFalse(_gpu_device_nodes_exist("intel_gpu"))
+            self.assertFalse(_gpu_device_nodes_exist("amd_gpu"))
+            self.assertTrue(_gpu_device_nodes_exist("jetson"))
+
+        with patch(
+            "frigate.stats.hardware.glob", return_value=["/dev/nvidia0"]
+        ) as dev_glob:
+            self.assertTrue(_gpu_device_nodes_exist("nvidia"))
+            dev_glob.assert_called_with("/dev/nvidia*")
+
+
+class TestScanFfmpeg(HardwareStatsTestCase):
+    def test_camera_and_input_args(self):
+        config = self.make_config("preset-nvidia")
+        config.cameras["back"].ffmpeg.inputs[0].hwaccel_args = "-hwaccel qsv"
+        stats = self.make_stats(config)
+
+        self.assertEqual(stats._scan_ffmpeg(), {"nvidia", "intel_gpu"})
+
+    def test_no_hwaccel(self):
+        stats = self.make_stats(self.make_config())
+        self.assertEqual(stats._scan_ffmpeg(), set())
+
+
+class TestScanDetectors(HardwareStatsTestCase):
+    def scan(self, specs: list[DeviceSpec], present: tuple[str, ...] = ()) -> set[str]:
+        stats = self.make_stats()
+        model = SimpleNamespace()
+        stats.config = SimpleNamespace(
+            models=[model], devices_for_model=lambda m: specs
+        )
+
+        with patch.object(
+            hardware.hardware_prober,
+            "probe",
+            return_value=[found(key) for key in present],
+        ):
+            return stats._scan_detectors()
+
+    def test_detector_mapping(self):
+        self.assertEqual(self.scan([DeviceSpec("rknn", "rknn", None)]), {"rockchip"})
+        self.assertEqual(
+            self.scan([DeviceSpec("axengine", "axengine", None)]), {"axengine"}
+        )
+        self.assertEqual(
+            self.scan([DeviceSpec("tensorrt:0", "tensorrt", "0")]), {"jetson"}
+        )
+        self.assertEqual(
+            self.scan([DeviceSpec("openvino:NPU", "openvino", "NPU")]), {"intel_npu"}
+        )
+        self.assertEqual(
+            self.scan([DeviceSpec("openvino:GPU", "openvino", "GPU")]), {"intel_gpu"}
+        )
+        self.assertEqual(
+            self.scan([DeviceSpec("openvino", "openvino", None)]), {"intel_gpu"}
+        )
+        self.assertEqual(
+            self.scan([DeviceSpec("openvino:CPU", "openvino", "CPU")]), set()
+        )
+        self.assertEqual(self.scan([DeviceSpec("cpu", "cpu", None)]), set())
+        self.assertEqual(self.scan([DeviceSpec("hailo8l", "hailo8l", None)]), set())
+
+    def test_onnx_resolves_to_present_gpu(self):
+        spec = DeviceSpec("onnx", "onnx", None)
+        self.assertEqual(self.scan([spec], ("onnx:nvidia",)), {"nvidia"})
+        self.assertEqual(self.scan([spec], ("onnx:amd",)), {"amd_gpu"})
+        self.assertEqual(self.scan([spec], ("cpu",)), set())
+
+
+class TestScanEnrichments(HardwareStatsTestCase):
+    def scan(self, config: FrigateConfig, present: tuple[str, ...]) -> set[str]:
+        stats = self.make_stats(config)
+
+        with patch.object(
+            hardware.hardware_prober,
+            "probe",
+            return_value=[found(key) for key in present],
+        ):
+            return stats._scan_enrichments()
+
+    def test_semantic_search_large_demands_gpu(self):
+        config = self.make_config(
+            semantic_search={"enabled": True, "model_size": "large"}
+        )
+        self.assertEqual(self.scan(config, ("onnx:nvidia",)), {"nvidia"})
+
+    def test_semantic_search_small_stays_cpu(self):
+        config = self.make_config(
+            semantic_search={"enabled": True, "model_size": "small"}
+        )
+        self.assertEqual(self.scan(config, ("onnx:nvidia",)), set())
+
+    def test_semantic_search_explicit_cpu_device(self):
+        config = self.make_config(
+            semantic_search={"enabled": True, "model_size": "large", "device": "CPU"}
+        )
+        self.assertEqual(self.scan(config, ("onnx:nvidia",)), set())
+
+    def test_face_recognition_defaults_to_gpu(self):
+        config = self.make_config(face_recognition={"enabled": True})
+        self.assertEqual(self.scan(config, ("onnx:amd",)), {"amd_gpu"})
+
+    def test_lpr_defaults_to_auto(self):
+        config = self.make_config(lpr={"enabled": True})
+        self.assertEqual(self.scan(config, ("openvino:GPU",)), {"intel_gpu"})
+
+    def test_no_gpu_present_means_no_demand(self):
+        config = self.make_config(
+            semantic_search={"enabled": True, "model_size": "large"}
+        )
+        self.assertEqual(self.scan(config, ("cpu",)), set())
+
+    def test_audio_transcription_gpu_is_cuda_only(self):
+        config = self.make_config()
+        config.audio_transcription.enabled = True
+        config.audio_transcription.device = "GPU"
+        self.assertEqual(self.scan(config, ("onnx:nvidia",)), {"nvidia"})
+        self.assertEqual(self.scan(config, ("onnx:amd",)), set())
+
+
+class TestUpdateConfig(HardwareStatsTestCase):
+    def test_catalog_from_hwaccel(self):
+        stats = self.make_stats(self.make_config("preset-nvidia"))
+        self.assertEqual(set(stats._monitored), {"nvidia"})
+
+    def test_telemetry_gates(self):
+        config = self.make_config(
+            "preset-intel-qsv-h264",
+            telemetry={"stats": {"intel_gpu_stats": False}},
+        )
+        stats = self.make_stats(config)
+        self.assertEqual(set(stats._monitored), set())
+
+    def test_recalculated_on_config_update(self):
+        config = self.make_config()
+        stats = self.make_stats(config)
+        self.assertEqual(set(stats._monitored), set())
+
+        config.cameras["back"].ffmpeg.hwaccel_args = "preset-rk-h264"
+        self.subscriber.return_value.check_for_updates.return_value = {
+            "ffmpeg": ["back"]
+        }
+
+        with patch("frigate.stats.hardware.get_cpu_stats", return_value={}):
+            stats.update_stats({})
+
+        self.assertEqual(set(stats._monitored), {"rockchip"})
+
+
+class TestUpdateStats(HardwareStatsTestCase):
+    def run_stats(self, stats: HardwareStats) -> dict:
+        all_stats: dict = {}
+
+        with patch("frigate.stats.hardware.get_cpu_stats", return_value={}):
+            stats.update_stats(all_stats)
+
+        return all_stats
+
+    def test_polls_once_and_formats_legacy_shape(self):
+        stats = self.make_stats(self.make_config("preset-nvidia"))
+        usage = {
+            0: {"name": "RTX", "gpu": 12.5, "mem": 25.0, "enc": 1, "dec": 2, "temp": 60}
+        }
+
+        with patch(
+            "frigate.stats.hardware.get_nvidia_gpu_stats", return_value=usage
+        ) as nvidia:
+            all_stats = self.run_stats(stats)
+
+        nvidia.assert_called_once()
+        self.assertEqual(
+            all_stats["gpu_usages"],
+            {
+                "RTX": {
+                    "vendor": "nvidia",
+                    "gpu": "12.5%",
+                    "mem": "25.0%",
+                    "enc": "1.0%",
+                    "dec": "2.0%",
+                    "temp": "60",
+                }
+            },
+        )
+
+    def test_failure_emits_fallback_and_latches_cooldown(self):
+        stats = self.make_stats(self.make_config("preset-nvidia"))
+
+        with patch(
+            "frigate.stats.hardware.get_nvidia_gpu_stats", return_value={}
+        ) as nvidia:
+            all_stats = self.run_stats(stats)
+
+            self.assertEqual(
+                all_stats["gpu_usages"],
+                {"nvidia-gpu": {"vendor": "nvidia", "gpu": "", "mem": ""}},
+            )
+            self.assertIn("nvidia", stats._errors)
+
+            # in cooldown, the second cycle skips the poll
+            all_stats = self.run_stats(stats)
+            nvidia.assert_called_once()
+            self.assertNotIn("gpu_usages", all_stats)
+
+            # an expired cooldown is retried
+            stats._errors["nvidia"] -= hardware.HARDWARE_ERROR_COOLDOWN_SECONDS + 1
+            self.run_stats(stats)
+            self.assertEqual(nvidia.call_count, 2)
+
+    def test_npu_failure_latches_cooldown(self):
+        stats = self.make_stats(self.make_config())
+        stats._monitored = {"axengine": lambda: hardware._poll_axengine(stats.config)}
+
+        with patch(
+            "frigate.stats.hardware.get_axcl_npu_stats", return_value=None
+        ) as axcl:
+            all_stats = self.run_stats(stats)
+            self.assertNotIn("npu_usages", all_stats)
+            self.assertIn("axengine", stats._errors)
+
+            self.run_stats(stats)
+            axcl.assert_called_once()
+
+    def test_rockchip_fills_both_buckets(self):
+        stats = self.make_stats(self.make_config("preset-rk-h264"))
+
+        with (
+            patch(
+                "frigate.stats.hardware.get_rockchip_gpu_stats",
+                return_value={"gpu": "12.5%", "mem": "-%"},
+            ),
+            patch(
+                "frigate.stats.hardware.get_rockchip_npu_stats",
+                return_value={"npu": 42.0, "mem": "-%"},
+            ),
+        ):
+            all_stats = self.run_stats(stats)
+
+        self.assertEqual(
+            all_stats["gpu_usages"],
+            {"rockchip": {"vendor": "rockchip", "gpu": "12.5%", "mem": "-%"}},
+        )
+        self.assertEqual(
+            all_stats["npu_usages"], {"rockchip": {"npu": 42.0, "mem": "-%"}}
+        )
+
+    def test_pollers_run_concurrently(self):
+        stats = self.make_stats(self.make_config())
+
+        def slow_poll() -> HardwarePollResult:
+            time.sleep(0.2)
+            return HardwarePollResult()
+
+        stats._monitored = {"nvidia": slow_poll, "intel_gpu": slow_poll}
+        start = time.monotonic()
+        self.run_stats(stats)
+
+        self.assertLess(time.monotonic() - start, 0.35)
+
+
+class TestHardwareTemperatures(unittest.TestCase):
+    @patch("frigate.stats.hardware.read_temperature", side_effect=[45.0, 55.0])
+    @patch("frigate.stats.hardware.os.listdir", return_value=["apex_1", "apex_0"])
+    @patch("frigate.stats.hardware.os.path.isdir", return_value=True)
+    def test_edgetpu_sorted_by_device(self, isdir, listdir, read):
+        self.assertEqual(get_hardware_temperatures("edgetpu"), [45.0, 55.0])
+        read.assert_any_call("/sys/class/apex/apex_0/temp")
+
+    @patch(
+        "frigate.stats.hardware.get_hailo_temps",
+        return_value={"hailo8l-1": 52.0, "hailo8l-0": 51.0},
+    )
+    def test_hailo_sorted_by_name(self, temps):
+        self.assertEqual(get_hardware_temperatures("hailo8l"), [51.0, 52.0])
+
+    def test_unsupported_type(self):
+        self.assertEqual(get_hardware_temperatures("rknn"), [])

frigate/util/services.py

@@ -911,8 +911,8 @@ def get_nvidia_gpu_stats() -> dict[int, dict]:
         return results
 
 
-def get_jetson_stats() -> dict[int, dict] | None:
-    results = {}
+def get_jetson_stats() -> dict[str, str] | None:
+    results: dict[str, str] = {}
 
     try:
         results["mem"] = "-"  # no discrete gpu memory