Implement UI for managing multiple models (#24023)

* Implement hardware detection and UI management

* Cleanup Frigate+ detection

* Don't count model as changed

* Fixes for audio map error

* Add descriptions

* Enforce that all model must exist

* Fix hardware picking

* Docs fixes

* WebUI cleanup

* Cleanup handling of scenes

* UI refinement

* Cleanup recommended UI

* test fixews

frigate/api/app.py

@@ -1330,8 +1330,18 @@ def categorized_object_names(
 
 
 @router.get("/audio_labels", dependencies=[Depends(allow_any_authenticated())])
-def get_audio_labels():
+def get_audio_labels(request: Request):
     labels = load_labels("/audio-labelmap.txt", prefill=521)
+
+    # configured overrides group several audio classes under one label, and the
+    # detector merges them over the defaults at runtime. Offer them here too, or
+    # a grouped label could never be picked in the UI.
+    config: FrigateConfig = request.app.frigate_config
+    labels.update(config.audio.labelmap)
+
+    for camera in config.cameras.values():
+        labels.update(camera.audio.labelmap)
+
     return JSONResponse(content=labels)
 
 

frigate/api/defs/tags.py

@@ -8,6 +8,7 @@ class Tags(Enum):
     chat = "Chat"
     events = "Events"
     export = "Export"
+    hardware = "Hardware"
     classification = "Classification"
     logs = "Logs"
     media = "Media"

frigate/api/fastapi_app.py

@@ -21,6 +21,7 @@ from frigate.api import (
     debug_replay,
     event,
     export,
+    hardware,
     media,
     motion_search,
     notification,
@@ -145,6 +146,7 @@ def create_fastapi_app(
     app.include_router(preview.router)
     app.include_router(notification.router)
     app.include_router(export.router)
+    app.include_router(hardware.router)
     app.include_router(event.router)
     app.include_router(media.router)
     app.include_router(motion_search.router)

frigate/api/hardware.py

@@ -0,0 +1,30 @@
+"""Hardware discovery APIs."""
+
+import logging
+
+from fastapi import APIRouter, Depends
+
+from frigate.api.auth import require_role
+from frigate.api.defs.tags import Tags
+from frigate.detectors.hardware import DetectionHardware, hardware_prober
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=[Tags.hardware])
+
+
+@router.get(
+    "/hardware/probe",
+    response_model=list[DetectionHardware],
+    dependencies=[Depends(require_role(["admin"]))],
+)
+def probe_hardware(refresh: bool = False) -> list[DetectionHardware]:
+    """Get the object detection hardware attached to this system.
+
+    Args:
+        refresh: Probe again instead of returning the cached result
+
+    Returns:
+        Every kind of detection hardware that was found
+    """
+    return hardware_prober.probe(refresh=refresh)

frigate/config/camera/detect.py

@@ -62,10 +62,10 @@ class DetectConfig(FrigateBaseModel):
         title="Detect width",
         description="Width (pixels) of frames used for the detect stream; leave empty to use the native stream resolution.",
     )
-    scene: SceneEnum | None = Field(
-        default=None,
+    scene: SceneEnum = Field(
+        default=SceneEnum.all,
         title="Detect scene",
-        description="The environment this camera looks at, used to pick which of the configured models runs on it. Defaults to the model with a scene of 'all'.",
+        description="The environment this camera looks at, used to pick which of the configured models runs on it. Cameras left on 'all' run the model configured with a scene of 'all'.",
     )
     fps: int = Field(
         default=5,

frigate/config/config.py

@@ -807,37 +807,37 @@ class FrigateConfig(FrigateBaseModel):
         }
         self._all_labels = labels
 
-    def _resolve_camera_model(self, name: str, scene: SceneEnum | None) -> ModelConfig:
+    def _resolve_camera_model(self, name: str, scene: SceneEnum) -> ModelConfig:
         """Resolve which model a camera runs on.
 
+        A camera may name a scene no model is configured for, which is valid as
+        long as an 'all' model is there to fall back to.
+
         Args:
             name: Name of the camera
-            scene: The camera's configured detect scene, if any
+            scene: The camera's detect scene, which defaults to 'all'
 
         Returns:
             The model the camera runs on
         """
         by_scene = {model.scene: model for model in self.models}
+        model = by_scene.get(scene)
 
-        if scene is not None:
-            model = by_scene.get(scene)
-
-            if model is None:
-                raise ValueError(
-                    f"Camera '{name}' has a detect scene of '{scene.value}', but no model is configured for that scene."
-                )
-
+        if model is not None:
             return model
 
-        default = by_scene.get(SceneEnum.all) or (
-            self.models[0] if len(self.models) == 1 else None
-        )
+        default = by_scene.get(SceneEnum.all)
 
         if default is None:
             raise ValueError(
-                f"Camera '{name}' must set detect -> scene, because more than one model is configured and none of them uses a scene of 'all'."
+                f"Camera '{name}' has a detect scene of '{scene.value}', but no model is configured for that scene or for 'all'."
             )
 
+        logger.warning(
+            "Camera '%s' has a detect scene of '%s', but no model is configured for that scene, so the 'all' model is used",
+            name,
+            scene.value,
+        )
         return default
 
     @model_validator(mode="after")

frigate/detectors/hardware.py

@@ -0,0 +1,368 @@
+"""Discovery of object detection hardware attached to the system.
+
+Every probe here is a filesystem read. Nothing shells out, initializes a
+runtime, or opens a device, so this is cheap enough to run from the API process
+while detector children hold the hardware.
+
+Hardware is reported whether or not this image ships a detector that can drive
+it. Matching hardware to an image is a separate concern.
+"""
+
+import logging
+import os
+from glob import glob
+
+from pydantic import BaseModel, Field
+
+from frigate.const import SUPPORTED_RK_SOCS
+from frigate.detectors.detector_types import config_types
+from frigate.util.services import enumerate_drm_devices
+
+logger = logging.getLogger(__name__)
+
+# roots the probes read from, so tests can point them at a fixture tree
+SYS_ROOT = "/sys"
+DEV_ROOT = "/dev"
+PROC_ROOT = "/proc"
+ETC_ROOT = "/etc"
+
+# a Coral reports as Global Unichip until its firmware is loaded, then as Google
+CORAL_USB_IDS = {("1a6e", "089a"), ("18d1", "9302")}
+
+INTEL_DRM_DRIVERS = ("i915", "xe")
+AMD_DRM_DRIVERS = ("amdgpu",)
+
+
+class HardwareUnit(BaseModel):
+    """One physical piece of hardware."""
+
+    device: str = Field(
+        title="Device string",
+        description="The value to put in a model's devices list, for example 'edgetpu:pci:1'.",
+    )
+    label: str = Field(
+        title="Unit label",
+        description="How to identify this unit among others of the same kind, for example 'PCIe 1'.",
+    )
+
+
+class DetectionHardware(BaseModel):
+    """A kind of detection hardware, and every unit of it that was found."""
+
+    key: str = Field(
+        title="Hardware key",
+        description="Stable identifier for this kind of hardware.",
+    )
+    detector: str = Field(
+        title="Detector type",
+        description="The detector that drives this hardware.",
+    )
+    name: str = Field(
+        title="Hardware name",
+        description="Human readable name for this kind of hardware.",
+    )
+    units: list[HardwareUnit] = Field(
+        title="Units",
+        description="Each physical piece of this hardware that was found.",
+    )
+    count: int = Field(
+        title="Unit count",
+        description="How many units were found.",
+    )
+    unlimited: bool = Field(
+        title="Unlimited detectors",
+        description="Whether this hardware can run more inference processes than there are units.",
+    )
+
+
+def _read(path: str) -> str | None:
+    """Read a small file, returning None if it cannot be read."""
+    try:
+        with open(path) as f:
+            return f.read().strip()
+    except OSError:
+        return None
+
+
+def _is_shareable(detector: str) -> bool:
+    """Whether a detector lets the same device run more than one process."""
+    config_class = config_types.get(detector)
+
+    # a detector missing from this image is assumed to behave like most of them
+    return config_class.shareable if config_class else True
+
+
+def _hardware(
+    key: str, detector: str, name: str, units: list[HardwareUnit]
+) -> DetectionHardware:
+    return DetectionHardware(
+        key=key,
+        detector=detector,
+        name=name,
+        units=units,
+        count=len(units),
+        unlimited=_is_shareable(detector),
+    )
+
+
+def detect_coral_pci() -> DetectionHardware | None:
+    """Find PCIe and M.2 Coral accelerators, which register as apex devices."""
+    names = sorted(
+        os.path.basename(path) for path in glob(f"{SYS_ROOT}/class/apex/apex_*")
+    )
+
+    if not names:
+        return None
+
+    units = [
+        HardwareUnit(device=f"edgetpu:pci:{index}", label=f"PCIe {index}")
+        for index in range(len(names))
+    ]
+    return _hardware("edgetpu:pci", "edgetpu", "Coral EdgeTPU (PCIe)", units)
+
+
+def detect_coral_usb() -> DetectionHardware | None:
+    """Find USB Coral accelerators by their USB vendor and product ids."""
+    found = 0
+
+    for device_dir in sorted(glob(f"{SYS_ROOT}/bus/usb/devices/*")):
+        vendor = _read(os.path.join(device_dir, "idVendor"))
+        product = _read(os.path.join(device_dir, "idProduct"))
+
+        if vendor and product and (vendor.lower(), product.lower()) in CORAL_USB_IDS:
+            found += 1
+
+    if not found:
+        return None
+
+    units = [
+        HardwareUnit(device=f"edgetpu:usb:{index}", label=f"USB {index}")
+        for index in range(found)
+    ]
+    return _hardware("edgetpu:usb", "edgetpu", "Coral EdgeTPU (USB)", units)
+
+
+def _drm_devices(drivers: tuple[str, ...]) -> list[str]:
+    """PCI addresses of DRM devices bound to one of the given drivers."""
+    return sorted(
+        pdev for pdev, driver in enumerate_drm_devices().items() if driver in drivers
+    )
+
+
+def detect_intel_gpu() -> DetectionHardware | None:
+    """Find Intel GPUs through their DRM driver."""
+    pdevs = _drm_devices(INTEL_DRM_DRIVERS)
+
+    if not pdevs:
+        return None
+
+    # OpenVINO reports a lone GPU as "GPU" and enumerates them as GPU.0, GPU.1
+    # only when there is more than one
+    if len(pdevs) == 1:
+        units = [HardwareUnit(device="openvino:GPU", label=pdevs[0])]
+    else:
+        units = [
+            HardwareUnit(device=f"openvino:GPU.{index}", label=pdev)
+            for index, pdev in enumerate(pdevs)
+        ]
+
+    return _hardware("openvino:GPU", "openvino", "Intel GPU", units)
+
+
+def detect_intel_npu() -> DetectionHardware | None:
+    """Find Intel NPUs, which register as accel devices bound to intel_vpu."""
+    units = []
+
+    for accel_path in sorted(glob(f"{SYS_ROOT}/class/accel/accel*")):
+        try:
+            driver = os.path.basename(os.readlink(f"{accel_path}/device/driver"))
+        except OSError:
+            continue
+
+        if driver != "intel_vpu":
+            continue
+
+        units.append(
+            HardwareUnit(device="openvino:NPU", label=os.path.basename(accel_path))
+        )
+
+    if not units:
+        return None
+
+    # OpenVINO has no way to address a specific NPU, so only the first is usable
+    return _hardware("openvino:NPU", "openvino", "Intel NPU", units[:1])
+
+
+def detect_amd_gpu() -> DetectionHardware | None:
+    """Find AMD GPUs through their DRM driver."""
+    pdevs = _drm_devices(AMD_DRM_DRIVERS)
+
+    if not pdevs:
+        return None
+
+    # ROCm runs through onnx, whose MIGraphX provider takes no device index, so
+    # only one is addressable
+    units = [HardwareUnit(device="onnx", label=pdevs[0])]
+    return _hardware("onnx:amd", "onnx", "AMD GPU", units)
+
+
+def detect_nvidia_gpu() -> DetectionHardware | None:
+    """Find discrete Nvidia GPUs through the nvidia driver's proc entries."""
+    units = []
+
+    for index, gpu_dir in enumerate(sorted(glob(f"{PROC_ROOT}/driver/nvidia/gpus/*"))):
+        information = _read(os.path.join(gpu_dir, "information")) or ""
+        name = f"GPU {index}"
+
+        for line in information.splitlines():
+            if line.startswith("Model:"):
+                name = line.split(":", 1)[1].strip()
+                break
+
+        units.append(HardwareUnit(device=f"onnx:{index}", label=name))
+
+    if not units:
+        return None
+
+    # the model name is more useful as the hardware name when there is only one
+    name = units[0].label if len(units) == 1 else "NVIDIA GPU"
+    return _hardware("onnx:nvidia", "onnx", name, units)
+
+
+def detect_jetson() -> DetectionHardware | None:
+    """Find an Nvidia Jetson, whose integrated GPU runs through tensorrt."""
+    is_jetson = os.path.isfile(f"{ETC_ROOT}/nv_tegra_release") or os.path.exists(
+        f"{SYS_ROOT}/devices/gpu.0/load"
+    )
+
+    if not is_jetson:
+        return None
+
+    units = [HardwareUnit(device="tensorrt:0", label="Integrated GPU")]
+    return _hardware("tensorrt", "tensorrt", "NVIDIA Jetson", units)
+
+
+def _dev_units(pattern: str, device: str, label: str) -> list[HardwareUnit]:
+    """Build units from device nodes matching a glob."""
+    return [
+        HardwareUnit(device=device.format(index=index), label=f"{label} {index}")
+        for index in range(len(glob(f"{DEV_ROOT}/{pattern}")))
+    ]
+
+
+def detect_hailo() -> DetectionHardware | None:
+    """Find Hailo accelerators by their device nodes."""
+    nodes = sorted(glob(f"{DEV_ROOT}/hailo*"))
+
+    if not nodes:
+        return None
+
+    # the hailo runtime schedules across every attached device itself, so there
+    # is nothing to address individually
+    units = [HardwareUnit(device="hailo8l:PCIe", label=os.path.basename(nodes[0]))]
+    return _hardware("hailo8l", "hailo8l", "Hailo", units)
+
+
+def detect_memryx() -> DetectionHardware | None:
+    """Find MemryX accelerators by their device nodes."""
+    units = _dev_units("memx*", "memryx:PCIe:{index}", "PCIe")
+
+    if not units:
+        return None
+
+    return _hardware("memryx", "memryx", "MemryX MX3", units)
+
+
+def detect_rockchip() -> DetectionHardware | None:
+    """Find a Rockchip NPU by reading the SoC from the device tree."""
+    compatible = _read(f"{PROC_ROOT}/device-tree/compatible")
+
+    if not compatible:
+        return None
+
+    soc = compatible.split(",")[-1].strip("\x00")
+
+    if soc not in SUPPORTED_RK_SOCS:
+        return None
+
+    units = [HardwareUnit(device="rknn", label=soc.upper())]
+    return _hardware("rknn", "rknn", f"Rockchip NPU ({soc.upper()})", units)
+
+
+def detect_axengine() -> DetectionHardware | None:
+    """Find an AXERA accelerator by its control device node."""
+    if not os.path.exists(f"{DEV_ROOT}/axcl_host"):
+        return None
+
+    units = [HardwareUnit(device="axengine", label="AXERA")]
+    return _hardware("axengine", "axengine", "AXERA NPU", units)
+
+
+def detect_synaptics() -> DetectionHardware | None:
+    """Find a Synaptics NPU by its device node."""
+    if not os.path.exists(f"{DEV_ROOT}/synap"):
+        return None
+
+    units = [HardwareUnit(device="synaptics", label="Synaptics")]
+    return _hardware("synaptics", "synaptics", "Synaptics NPU", units)
+
+
+def detect_cpu() -> DetectionHardware:
+    """The CPU, which is always available."""
+    units = [HardwareUnit(device="cpu", label="CPU")]
+    return _hardware("cpu", "cpu", "CPU", units)
+
+
+# ordered so accelerators are offered ahead of the CPU fallback
+PROBES = (
+    detect_coral_pci,
+    detect_coral_usb,
+    detect_hailo,
+    detect_memryx,
+    detect_intel_npu,
+    detect_intel_gpu,
+    detect_nvidia_gpu,
+    detect_jetson,
+    detect_amd_gpu,
+    detect_rockchip,
+    detect_axengine,
+    detect_synaptics,
+    detect_cpu,
+)
+
+
+class HardwareProber:
+    """Probes for detection hardware, caching the result for the process."""
+
+    _hardware: list[DetectionHardware] | None = None
+
+    def probe(self, refresh: bool = False) -> list[DetectionHardware]:
+        """Get the detection hardware attached to this system.
+
+        Args:
+            refresh: Probe again instead of using the cached result
+
+        Returns:
+            Every kind of detection hardware that was found
+        """
+        if self._hardware is not None and not refresh:
+            return self._hardware
+
+        found = []
+
+        for probe in PROBES:
+            try:
+                hardware = probe()
+            except Exception:
+                logger.warning("Failed to probe for %s", probe.__name__, exc_info=True)
+                continue
+
+            if hardware is not None:
+                found.append(hardware)
+
+        logger.debug("Detected hardware: %s", [h.key for h in found])
+        self._hardware = found
+        return found
+
+
+hardware_prober = HardwareProber()

frigate/test/http_api/test_http_audio_labels.py

@@ -0,0 +1,55 @@
+"""Tests for the audio labels API."""
+
+import unittest
+from unittest.mock import patch
+
+from frigate.models import Event
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+
+
+class TestHttpAudioLabels(BaseTestHttp):
+    def setUp(self):
+        super().setUp([Event])
+
+    def _labels(self, config: dict | None = None) -> dict[str, str]:
+        if config:
+            self.minimal_config.update(config)
+
+        app = self.create_app()
+
+        with patch(
+            "frigate.api.app.load_labels", return_value={0: "speech", 1: "bark"}
+        ):
+            with AuthTestClient(app) as client:
+                response = client.get("/audio_labels")
+
+        self.assertEqual(response.status_code, 200)
+        return response.json()
+
+    def test_the_default_labels_are_returned(self):
+        self.assertEqual(self._labels(), {"0": "speech", "1": "bark"})
+
+    def test_a_global_labelmap_override_is_offered(self):
+        # grouping several classes under one label makes that label selectable
+        labels = self._labels({"audio": {"labelmap": {0: "noise", 1: "noise"}}})
+
+        self.assertEqual(set(labels.values()), {"noise"})
+
+    def test_a_camera_labelmap_override_is_offered(self):
+        labels = self._labels(
+            {
+                "cameras": {
+                    "front_door": {
+                        **self.minimal_config["cameras"]["front_door"],
+                        "audio": {"labelmap": {1: "dogs"}},
+                    }
+                }
+            }
+        )
+
+        self.assertEqual(labels["1"], "dogs")
+        self.assertEqual(labels["0"], "speech")
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

frigate/test/test_config.py

@@ -159,7 +159,7 @@ class TestConfig(unittest.TestCase):
             FrigateConfig(**(deep_merge(config, self.minimal)))
 
     @patch("frigate.detectors.detector_config.load_labels")
-    def test_camera_scene_must_match_a_model(self, mock_labels):
+    def test_camera_scene_without_a_model_falls_back_to_all(self, mock_labels):
         mock_labels.return_value = {}
         config = {
             "models": [{"devices": ["cpu"]}],
@@ -175,6 +175,27 @@ class TestConfig(unittest.TestCase):
             },
         }
 
+        frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
+
+        assert frigate_config.model_for_camera("back").scene == SceneEnum.all
+
+    @patch("frigate.detectors.detector_config.load_labels")
+    def test_camera_scene_without_a_model_or_a_default(self, mock_labels):
+        mock_labels.return_value = {}
+        config = {
+            "models": [{"scene": "indoor", "devices": ["cpu"]}],
+            "cameras": {
+                "back": {
+                    "detect": {"scene": "outdoor"},
+                    "ffmpeg": {
+                        "inputs": [
+                            {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]},
+                        ]
+                    },
+                },
+            },
+        }
+
         with self.assertRaises(ValidationError):
             FrigateConfig(**(deep_merge(config, self.minimal)))
 
@@ -1298,7 +1319,7 @@ class TestConfig(unittest.TestCase):
             70: "dogs",
             75: "dogs",
         }
-        assert frigate_config.model.merged_labelmap[69] != "dogs"
+        assert frigate_config.primary_model.merged_labelmap[69] != "dogs"
 
     def test_default_labelmap_empty(self):
         config = {

frigate/test/test_detector_hardware.py

@@ -0,0 +1,247 @@
+"""Tests for detection hardware discovery."""
+
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from frigate.detectors import hardware
+from frigate.detectors.detector_types import config_types
+from frigate.detectors.hardware import HardwareProber
+
+
+def write(path: str, content: str = "") -> None:
+    """Create a file and any parent directories."""
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+
+    with open(path, "w") as f:
+        f.write(content)
+
+
+class HardwareProbeTestCase(unittest.TestCase):
+    """Points every probe at an empty fixture tree, so nothing is found by default."""
+
+    def setUp(self):
+        self.root = tempfile.TemporaryDirectory()
+        self.addCleanup(self.root.cleanup)
+
+        for name in ("SYS_ROOT", "DEV_ROOT", "PROC_ROOT", "ETC_ROOT"):
+            sub = os.path.join(self.root.name, name.split("_")[0].lower())
+            os.makedirs(sub, exist_ok=True)
+            patcher = patch.object(hardware, name, sub)
+            patcher.start()
+            self.addCleanup(patcher.stop)
+            setattr(self, name.lower(), sub)
+
+        drm = patch.object(hardware, "enumerate_drm_devices", return_value={})
+        self.drm = drm.start()
+        self.addCleanup(drm.stop)
+
+    def probe(self) -> dict[str, hardware.DetectionHardware]:
+        return {found.key: found for found in HardwareProber().probe()}
+
+
+class TestNoHardware(HardwareProbeTestCase):
+    def test_only_the_cpu_is_reported(self):
+        self.assertEqual(list(self.probe()), ["cpu"])
+
+    def test_the_cpu_is_unlimited(self):
+        self.assertTrue(self.probe()["cpu"].unlimited)
+
+
+class TestCoral(HardwareProbeTestCase):
+    def test_each_apex_device_is_a_unit(self):
+        for name in ("apex_0", "apex_1"):
+            os.makedirs(os.path.join(self.sys_root, "class", "apex", name))
+
+        coral = self.probe()["edgetpu:pci"]
+
+        self.assertEqual(coral.count, 2)
+        self.assertEqual(
+            [unit.device for unit in coral.units],
+            ["edgetpu:pci:0", "edgetpu:pci:1"],
+        )
+
+    def test_a_coral_is_not_unlimited(self):
+        os.makedirs(os.path.join(self.sys_root, "class", "apex", "apex_0"))
+
+        self.assertFalse(self.probe()["edgetpu:pci"].unlimited)
+
+    def test_usb_corals_are_found_by_their_usb_ids(self):
+        usb = os.path.join(self.sys_root, "bus", "usb", "devices")
+        # a Coral reports as Global Unichip before its firmware loads
+        write(os.path.join(usb, "1-1", "idVendor"), "1a6e")
+        write(os.path.join(usb, "1-1", "idProduct"), "089a")
+        # and as Google afterwards
+        write(os.path.join(usb, "1-2", "idVendor"), "18d1")
+        write(os.path.join(usb, "1-2", "idProduct"), "9302")
+
+        coral = self.probe()["edgetpu:usb"]
+
+        self.assertEqual(coral.count, 2)
+        self.assertEqual(coral.units[0].device, "edgetpu:usb:0")
+
+    def test_other_usb_devices_are_ignored(self):
+        usb = os.path.join(self.sys_root, "bus", "usb", "devices")
+        write(os.path.join(usb, "1-1", "idVendor"), "046d")
+        write(os.path.join(usb, "1-1", "idProduct"), "0825")
+
+        self.assertNotIn("edgetpu:usb", self.probe())
+
+
+class TestGpus(HardwareProbeTestCase):
+    def test_a_single_intel_gpu_is_the_unnumbered_device(self):
+        self.drm.return_value = {"0000:00:02.0": "i915"}
+
+        gpu = self.probe()["openvino:GPU"]
+
+        self.assertEqual([unit.device for unit in gpu.units], ["openvino:GPU"])
+        self.assertTrue(gpu.unlimited)
+
+    def test_multiple_intel_gpus_are_numbered(self):
+        self.drm.return_value = {"0000:00:02.0": "i915", "0000:03:00.0": "xe"}
+
+        gpu = self.probe()["openvino:GPU"]
+
+        self.assertEqual(
+            [unit.device for unit in gpu.units],
+            ["openvino:GPU.0", "openvino:GPU.1"],
+        )
+
+    def test_non_gpu_drm_devices_are_ignored(self):
+        self.drm.return_value = {"0000:00:02.0": "virtio-mmio"}
+
+        self.assertNotIn("openvino:GPU", self.probe())
+
+    def test_amd_gpus_run_through_onnx(self):
+        self.drm.return_value = {"0000:03:00.0": "amdgpu"}
+
+        self.assertEqual(self.probe()["onnx:amd"].units[0].device, "onnx")
+
+    def test_an_intel_npu_is_found_by_its_driver(self):
+        accel = os.path.join(self.sys_root, "class", "accel", "accel0", "device")
+        os.makedirs(accel)
+        os.symlink("/drivers/intel_vpu", os.path.join(accel, "driver"))
+
+        self.assertEqual(self.probe()["openvino:NPU"].units[0].device, "openvino:NPU")
+
+    def test_other_accel_devices_are_ignored(self):
+        accel = os.path.join(self.sys_root, "class", "accel", "accel0", "device")
+        os.makedirs(accel)
+        os.symlink("/drivers/something_else", os.path.join(accel, "driver"))
+
+        self.assertNotIn("openvino:NPU", self.probe())
+
+
+class TestNvidia(HardwareProbeTestCase):
+    def _add_gpu(self, address: str, model: str) -> None:
+        write(
+            os.path.join(
+                self.proc_root, "driver", "nvidia", "gpus", address, "information"
+            ),
+            f"Model: \t {model}\nIRQ:   \t 62\n",
+        )
+
+    def test_the_model_name_is_read_from_proc(self):
+        self._add_gpu("0000:01:00.0", "NVIDIA GeForce RTX 3060")
+
+        gpu = self.probe()["onnx:nvidia"]
+
+        self.assertEqual(gpu.name, "NVIDIA GeForce RTX 3060")
+        self.assertEqual(gpu.units[0].device, "onnx:0")
+
+    def test_multiple_gpus_are_indexed(self):
+        self._add_gpu("0000:01:00.0", "NVIDIA GeForce RTX 3060")
+        self._add_gpu("0000:02:00.0", "NVIDIA GeForce RTX 4090")
+
+        gpu = self.probe()["onnx:nvidia"]
+
+        self.assertEqual(gpu.name, "NVIDIA GPU")
+        self.assertEqual([unit.device for unit in gpu.units], ["onnx:0", "onnx:1"])
+        self.assertEqual(gpu.units[1].label, "NVIDIA GeForce RTX 4090")
+
+    def test_a_jetson_runs_through_tensorrt(self):
+        write(os.path.join(self.etc_root, "nv_tegra_release"), "# R36 (release)")
+
+        self.assertEqual(self.probe()["tensorrt"].units[0].device, "tensorrt:0")
+
+
+class TestAccelerators(HardwareProbeTestCase):
+    def test_hailo_is_found_by_its_device_node(self):
+        write(os.path.join(self.dev_root, "hailo0"))
+
+        self.assertEqual(self.probe()["hailo8l"].units[0].device, "hailo8l:PCIe")
+
+    def test_each_memryx_node_is_a_unit(self):
+        write(os.path.join(self.dev_root, "memx0"))
+        write(os.path.join(self.dev_root, "memx1"))
+
+        memryx = self.probe()["memryx"]
+
+        self.assertEqual(
+            [unit.device for unit in memryx.units],
+            ["memryx:PCIe:0", "memryx:PCIe:1"],
+        )
+        self.assertFalse(memryx.unlimited)
+
+    def test_a_supported_rockchip_soc_is_reported(self):
+        write(
+            os.path.join(self.proc_root, "device-tree", "compatible"),
+            "rockchip,rk3588\x00",
+        )
+
+        self.assertEqual(self.probe()["rknn"].units[0].device, "rknn")
+
+    def test_an_unsupported_soc_is_ignored(self):
+        write(
+            os.path.join(self.proc_root, "device-tree", "compatible"),
+            "nvidia,tegra\x00",
+        )
+
+        self.assertNotIn("rknn", self.probe())
+
+    def test_axengine_is_found_by_its_control_node(self):
+        write(os.path.join(self.dev_root, "axcl_host"))
+
+        self.assertEqual(self.probe()["axengine"].units[0].device, "axengine")
+
+    def test_synaptics_is_found_by_its_device_node(self):
+        write(os.path.join(self.dev_root, "synap"))
+
+        self.assertEqual(self.probe()["synaptics"].units[0].device, "synaptics")
+
+
+class TestProber(HardwareProbeTestCase):
+    def test_the_result_is_cached_until_refreshed(self):
+        prober = HardwareProber()
+        self.assertNotIn("edgetpu:pci", {found.key for found in prober.probe()})
+
+        os.makedirs(os.path.join(self.sys_root, "class", "apex", "apex_0"))
+
+        self.assertNotIn("edgetpu:pci", {found.key for found in prober.probe()})
+        self.assertIn(
+            "edgetpu:pci", {found.key for found in prober.probe(refresh=True)}
+        )
+
+    def test_a_failing_probe_does_not_break_the_rest(self):
+        with patch.object(hardware, "detect_hailo", side_effect=OSError("boom")):
+            self.assertIn("cpu", self.probe())
+
+    def test_unlimited_tracks_the_detector_shareable_flag(self):
+        for name in ("apex_0",):
+            os.makedirs(os.path.join(self.sys_root, "class", "apex", name))
+        write(os.path.join(self.dev_root, "memx0"))
+        self.drm.return_value = {"0000:00:02.0": "i915"}
+
+        for found in self.probe().values():
+            config_class = config_types.get(found.detector)
+
+            if config_class is None:
+                continue
+
+            with self.subTest(hardware=found.key):
+                self.assertEqual(found.unlimited, config_class.shareable)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

frigate/test/test_gpu_stats.py

@@ -108,7 +108,7 @@ class TestGpuStats(unittest.TestCase):
     @patch("frigate.util.services.time.sleep")
     @patch("frigate.util.services.time.monotonic")
     @patch("frigate.util.services._read_intel_drm_fdinfo")
-    @patch("frigate.util.services._enumerate_drm_devices")
+    @patch("frigate.util.services.enumerate_drm_devices")
     def test_intel_gpu_stats_fdinfo(
         self, drm_devices, read_fdinfo, monotonic, sleep, get_names
     ):
@@ -187,7 +187,7 @@ class TestGpuStats(unittest.TestCase):
     @patch("frigate.util.services.time.sleep")
     @patch("frigate.util.services.time.monotonic")
     @patch("frigate.util.services._read_intel_drm_fdinfo")
-    @patch("frigate.util.services._enumerate_drm_devices")
+    @patch("frigate.util.services.enumerate_drm_devices")
     def test_intel_gpu_stats_xe_capacity(
         self, drm_devices, read_fdinfo, monotonic, sleep, get_names
     ):
@@ -246,7 +246,7 @@ class TestGpuStats(unittest.TestCase):
     @patch("frigate.stats.intel_gpu_info.intel_gpu_name_resolver.get_names")
     @patch("frigate.util.services.time.sleep")
     @patch("frigate.util.services._read_intel_drm_fdinfo")
-    @patch("frigate.util.services._enumerate_drm_devices")
+    @patch("frigate.util.services.enumerate_drm_devices")
     def test_intel_gpu_stats_no_clients_reports_idle(
         self, drm_devices, read_fdinfo, sleep, get_names
     ):
@@ -274,7 +274,7 @@ class TestGpuStats(unittest.TestCase):
 
     @patch("frigate.util.services.time.sleep")
     @patch("frigate.util.services._read_intel_drm_fdinfo")
-    @patch("frigate.util.services._enumerate_drm_devices")
+    @patch("frigate.util.services.enumerate_drm_devices")
     def test_intel_gpu_stats_clients_without_engine_counters(
         self, drm_devices, read_fdinfo, sleep
     ):
@@ -301,7 +301,7 @@ class TestGpuStats(unittest.TestCase):
         read_fdinfo.assert_called_once()
 
     @patch("frigate.util.services._read_intel_drm_fdinfo")
-    @patch("frigate.util.services._enumerate_drm_devices")
+    @patch("frigate.util.services.enumerate_drm_devices")
     def test_intel_gpu_stats_no_intel_device(self, drm_devices, read_fdinfo):
         # Only a non-Intel GPU is visible in sysfs; /proc is never scanned
         drm_devices.return_value = {"0000:01:00.0": "nvidia"}
@@ -310,7 +310,7 @@ class TestGpuStats(unittest.TestCase):
         read_fdinfo.assert_not_called()
 
     @patch("frigate.util.services._read_intel_drm_fdinfo")
-    @patch("frigate.util.services._enumerate_drm_devices")
+    @patch("frigate.util.services.enumerate_drm_devices")
     @patch("frigate.util.services._resolve_intel_gpu_pdev")
     def test_intel_gpu_stats_unresolvable_device_hint(
         self, resolve_pdev, drm_devices, read_fdinfo
@@ -324,7 +324,7 @@ class TestGpuStats(unittest.TestCase):
         read_fdinfo.assert_not_called()
 
     @patch("frigate.util.services._read_intel_drm_fdinfo")
-    @patch("frigate.util.services._enumerate_drm_devices")
+    @patch("frigate.util.services.enumerate_drm_devices")
     @patch("frigate.util.services._resolve_intel_gpu_pdev")
     def test_intel_gpu_stats_hint_resolves_to_non_intel_gpu(
         self, resolve_pdev, drm_devices, read_fdinfo
@@ -342,7 +342,7 @@ class TestGpuStats(unittest.TestCase):
         read_fdinfo.assert_not_called()
 
     @patch("frigate.util.services._read_intel_drm_fdinfo")
-    @patch("frigate.util.services._enumerate_drm_devices")
+    @patch("frigate.util.services.enumerate_drm_devices")
     def test_intel_gpu_stats_unreadable_proc(self, drm_devices, read_fdinfo):
         # A scan failure (None) is a different condition than a scan that
         # finds no clients ({}) and must not report idle
@@ -355,7 +355,7 @@ class TestGpuStats(unittest.TestCase):
     @patch("frigate.util.services.time.sleep")
     @patch("frigate.util.services.time.monotonic")
     @patch("frigate.util.services._read_intel_drm_fdinfo")
-    @patch("frigate.util.services._enumerate_drm_devices")
+    @patch("frigate.util.services.enumerate_drm_devices")
     def test_intel_gpu_stats_clients_lost_between_samples(
         self, drm_devices, read_fdinfo, monotonic, sleep, get_names
     ):

frigate/util/services.py

@@ -315,7 +315,7 @@ def _resolve_intel_gpu_pdev(device: str | None) -> str | None:
     return pdev if _PCI_ADDRESS_RE.match(pdev) else None
 
 
-def _enumerate_drm_devices() -> dict[str, str]:
+def enumerate_drm_devices() -> dict[str, str]:
     """Map each PCI-attached DRM device to its bound kernel driver.
 
     Reads /sys/class/drm, which reflects every GPU on the host even when only
@@ -517,7 +517,7 @@ def get_intel_gpu_stats(
         )
         return None
 
-    drm_devices = _enumerate_drm_devices()
+    drm_devices = enumerate_drm_devices()
     intel_pdevs = {
         pdev: driver
         for pdev, driver in drm_devices.items()