From 6ae8050974b5c89904cf6f1a19f21715f580f6f9 Mon Sep 17 00:00:00 2001 From: Nicolas Mowen Date: Mon, 31 Aug 2026 15:56:37 -0600 Subject: [PATCH] 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 | 8 +- frigate/stats/hardware.py | 472 ++++++++++++++++++++++++++++ frigate/stats/util.py | 260 +-------------- frigate/test/test_hardware_stats.py | 418 ++++++++++++++++++++++++ frigate/util/services.py | 4 +- 5 files changed, 909 insertions(+), 253 deletions(-) create mode 100644 frigate/stats/hardware.py create mode 100644 frigate/test/test_hardware_stats.py diff --git a/frigate/stats/emitter.py b/frigate/stats/emitter.py index 7c2e328c73..6072314118 100644 --- a/frigate/stats/emitter.py +++ b/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...") diff --git a/frigate/stats/hardware.py b/frigate/stats/hardware.py new file mode 100644 index 0000000000..70062152b2 --- /dev/null +++ b/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 [] diff --git a/frigate/stats/util.py b/frigate/stats/util.py index 502769686e..575903e8f7 100644 --- a/frigate/stats/util.py +++ b/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"]), diff --git a/frigate/test/test_hardware_stats.py b/frigate/test/test_hardware_stats.py new file mode 100644 index 0000000000..4069455238 --- /dev/null +++ b/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"), []) diff --git a/frigate/util/services.py b/frigate/util/services.py index 3261f3d28f..7f39601851 100644 --- a/frigate/util/services.py +++ b/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