diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/init-devices/run b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/init-devices/run index 7b20bf3914..81e47f973a 100755 --- a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/init-devices/run +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/init-devices/run @@ -38,6 +38,7 @@ device_globs=( "/dev/nvidia*" "/dev/memx*" "/dev/dxrt*" + "/dev/metis*" ) IFS=',' read -ra extra_globs <<< "${DEVICE_ACL_PATHS:-}" diff --git a/docs/data/object_detectors_models.yaml b/docs/data/object_detectors_models.yaml index d105c7181f..fabd637d81 100644 --- a/docs/data/object_detectors_models.yaml +++ b/docs/data/object_detectors_models.yaml @@ -1278,3 +1278,34 @@ axengine: input_dtype: int input_pixel_format: bgr labelmap_path: /labelmap/coco-80.txt +axelera: + title: Axelera Metis + models: + - key: yolox + label: YOLOX + recommended: true + download: |- + Axelera models must be compiled for the Metis NPU with the Voyager SDK before Frigate can run them (compilation is not supported inside the container). Compile a model with the SDK, zip the compiled output, and point the model's `path` at the zip. The default recommendation is the Apache-2.0 licensed YOLOX (Megvii YOLOX, e.g. the `yolox-s-coco-onnx` model-zoo network). + ui: |- + Navigate to **Settings > System > Detection models** and select **Axelera Metis** from the **Hardware** dropdown. Then, on the same model, open the **Custom Model** tab and configure: + + | Field | Value | + | ---------------------------------------- | ------------------------------------------------------------ | + | **Custom object detector model path** | `/config/model_cache/axelera/yolox-s-coco-onnx.zip` | + | **Label map for custom object detector** | `/labelmap/coco-80.txt` | + | **Object detection model input width** | `640` (must equal the compiled input size) | + | **Object detection model input height** | `640` | + | **Model Input Pixel Color Format** | `bgr` (Frigate delivers BGR; the plugin hands RGB planes to the compiled input) | + | **Model Input Tensor Shape** | `nhwc` (Frigate's default value) | + | **Model Input D Type** | `int` (Frigate's default value) | + | **Object Detection Model Type** | `yolox` | + yaml: |- + models: + - devices: + - axelera:PCIe + path: /config/model_cache/axelera/yolox-s-coco-onnx.zip + model_type: yolox + width: 640 + height: 640 + input_pixel_format: bgr + labelmap_path: /labelmap/coco-80.txt diff --git a/docs/docs/configuration/object_detectors.md b/docs/docs/configuration/object_detectors.md index 1ba6fd0003..d8664c0a4a 100644 --- a/docs/docs/configuration/object_detectors.md +++ b/docs/docs/configuration/object_detectors.md @@ -63,6 +63,10 @@ Frigate supports multiple different detectors that work on different types of ha - [AXEngine](#axera): axmodels can run on AXERA AI acceleration. +**Axelera** + +- [Axelera Metis](#axelera-metis): The Metis AI Accelerator is available in M.2 and PCIe formats, running compiled YOLO models through the vendor runtime. + **For Testing** - [CPU Detector (not recommended for actual use](#cpu-detector-not-recommended): Use a CPU to run tflite model, this is not recommended and in most cases OpenVINO can be used in CPU mode with better results. @@ -100,7 +104,7 @@ models: - openvino:GPU ``` -Coral EdgeTPU and MemryX accelerators can only be opened by one process, so those devices can not be repeated. +Coral EdgeTPU, MemryX, and Axelera Metis accelerators can only be opened by one process, so those devices can not be repeated. ### Running more than one model @@ -940,3 +944,41 @@ The AXEngine python package is not part of the Frigate image. It is downloaded a When configuring the AXEngine detector, you have to specify the model name. + +--- + +## Axelera Metis + + + +Hardware accelerated object detection is supported on the Axelera Metis AI Accelerator, available in M.2 and PCIe card formats. Inference runs through Axelera's low-level runtime (`axelera.runtime`), which executes compiled YOLO-family models with a flat per-frame latency. + +See the [installation docs](../frigate/installation.md#axelera-metis) for information on configuring the Metis hardware. + +:::info + +The Axelera runtime is not part of the Frigate image. It is downloaded and installed into `/config/.local` the first time an Axelera detector is configured, verified against pinned checksums, and updated automatically when a Frigate release pins a new version. If the container has no internet access, see [Detector runtimes](/frigate/network_requirements#detector-runtimes) for how to provide the files yourself. + +::: + +### Configuration {#configuration-axelera} + + + +#### Using a Custom Model + +Models must be compiled for the Metis NPU with the Axelera Voyager SDK before Frigate can run them. The compiled output is a directory containing `1/model.json`, `1/manifest.json`, and the baked `1/postprocess_graph.onnx`. + +Compile the model on a machine with the Voyager SDK installed (compilation is not supported inside the Frigate container), for example with the SDK's `deploy.py` against a model-zoo yaml. Then zip the compiled output and point the model's `path` at the zip (bind-mounted, or at an http(s) URL); the archive extracts fresh on every start. + +```yaml +models: + - devices: + - axelera:PCIe + path: /config/model_cache/axelera/yolox-s-coco-onnx.zip + width: 640 # must equal the compiled input size + height: 640 + input_pixel_format: bgr # Frigate delivers BGR; the plugin feeds RGB planes + model_type: yolox + labelmap_path: /labelmap/coco-80.txt +``` diff --git a/docs/docs/frigate/installation.md b/docs/docs/frigate/installation.md index 605c8afd2c..27c862c845 100644 --- a/docs/docs/frigate/installation.md +++ b/docs/docs/frigate/installation.md @@ -596,6 +596,60 @@ If you are using `docker run`, add this option to your command `--device /dev/ax Finally, configure [hardware object detection](/configuration/object_detectors#axera) to complete the setup. +### Axelera Metis + +The Axelera Metis AI Accelerator is available in the M.2 and PCIe card form factors and is supported on x86 (Intel/AMD) PCs and Rock 5B class arm64 boards. + +The Axelera runtime used inside the container is not part of the Frigate image; Frigate downloads and installs it at first start once an Axelera detector is configured. Containers without internet access can provide the files themselves, see [Detector runtimes](/frigate/network_requirements#detector-runtimes). The host side kernel driver still has to be installed as described below. + +#### Installation + +Install the Metis PCIe kernel driver and firmware on the **host** following [Axelera's driver installation guide](https://developer.axelera.ai/axelera-documentation/Getting-Started/GettingStartedWithMetis/InstallingAxeleraMetisDriver). The driver creates the `/dev/metis*` device nodes. Reboot after installing it. + +#### Setup + +To set up Frigate, follow the default installation instructions, for example: `ghcr.io/blakeblackshear/frigate:stable` + +The runtime enumerates the card by the colon-separated name the host udev rules create as a symlink (for example `/dev/metis-0:1:0`). Docker cannot pass that name as a `--device` argument, and a bind mount resolves the symlink to the raw node, which the runtime does not enumerate. A CDI device spec defines the container path and host path as separate fields, and a `startContainer` hook recreates the symlink inside the container. + +Create `/etc/cdi/axelera.yaml` on the host, matching the node `ls /dev/metis*` prints: + +```yaml +cdiVersion: "0.6.0" +kind: axelera.ai/metis +devices: + - name: "0" + containerEdits: + deviceNodes: + - path: /dev/metis-0-1-0 + hostPath: /dev/metis-0-1-0 + - path: /dev/dma_heap/system + hooks: + - hookName: startContainer + path: /bin/ln + args: ["/bin/ln", "-sf", "metis-0-1-0", "/dev/metis-0:1:0"] + timeout: 5 +``` + +Then grant the device in your `docker-compose.yml` file: + +```yaml +devices: + - axelera.ai/metis=0 +``` + +If you are using `docker run`, add `--device axelera.ai/metis=0` to your command. CDI is enabled by default in Docker 28 and later; older versions need `"features": {"cdi": true}` in `daemon.json`. The hook runs as the container user, so this setup needs the container to start as root and does not work under Docker's `user:` mode. + +:::warning + +DMA-BUF must be available on the host kernel. On Debian/Ubuntu this is provided by the default kernel; on other systems it may need enabling (`CONFIG_DMABUF_HEAPS_SYSTEM`). If Frigate's log reports that no DMA heap can be allocated, check that `/dev/dma_heap/system` exists on the host and is passed into the container. + +::: + +#### Configuration + +Finally, configure [hardware object detection](/configuration/object_detectors#axelera-metis) to complete the setup. + ## Docker Running through Docker with Docker Compose is the recommended install method. diff --git a/docs/docs/frigate/network_requirements.md b/docs/docs/frigate/network_requirements.md index f1b686a4ac..f28ecac65e 100644 --- a/docs/docs/frigate/network_requirements.md +++ b/docs/docs/frigate/network_requirements.md @@ -65,11 +65,12 @@ The SDKs for a few hardware detectors are not shipped in the Frigate image. They | [Hailo 8 / 8L / 8R](/configuration/object_detectors#hailo) | 4.21.0 | `hailort-debian12-amd64.tar.gz` and `hailort-4.21.0-cp311-cp311-linux_x86_64.whl` on x86, `hailort-debian12-arm64.tar.gz` and `hailort-4.21.0-cp311-cp311-linux_aarch64.whl` on arm64 | [GitHub release](https://github.com/frigate-nvr/hailort/releases/tag/v4.21.0) | | [MemryX MX3](/configuration/object_detectors#memryx-mx3) | 2.1.0 | `mx_accl_frigate-2.1.0.zip` (the release source archive, renamed) | [GitHub archive](https://github.com/memryx/mx_accl_frigate/archive/refs/tags/v2.1.0.zip) | | [AXERA AXEngine](/configuration/object_detectors#axera) | 0.1.3 | `axengine-0.1.3-py3-none-any.whl` | [GitHub release](https://github.com/AXERA-TECH/pyaxengine/releases/tag/0.1.3-frigate) | +| [Axelera Metis](/configuration/object_detectors#axelera-metis) | 1.8.0 | `axelera_runtime-1.8.0-cp311-cp311-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl` on x86, `axelera_runtime-1.8.0-cp311-cp311-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl` on arm64, plus `axelera_firmware-1.8.0-py3-none-any.whl` on both | [Axelera Artifactory PyPI](https://software.axelera.ai/artifactory/#web/ui/redirect/pypi/axelera-pypi/simple) | If the container cannot reach GitHub, provide the files yourself: 1. Download the files for your architecture on a machine with internet access. -2. Place them, with exactly the file names listed above, in `/config/model_cache/runtimes//`, where `` is the detector named in your config's `devices` (`hailo`, `memryx`, or `axengine`). +2. Place them, with exactly the file names listed above, in `/config/model_cache/runtimes//`, where `` is the detector named in your config's `devices` (`hailo`, `memryx`, `axengine`, or `axelera`). 3. Start Frigate. Files whose checksum matches are installed without any download; a file with the wrong checksum is discarded and downloaded again, so a failed startup log names the file to replace. The `GITHUB_ENDPOINT` mirror variable below applies to these downloads as well. diff --git a/frigate/detectors/hardware.py b/frigate/detectors/hardware.py index 10becfb219..1336ae6de8 100644 --- a/frigate/detectors/hardware.py +++ b/frigate/detectors/hardware.py @@ -284,6 +284,19 @@ def detect_deepx() -> DetectionHardware | None: return _hardware("deepx", "deepx", "DEEPX NPU", units) +def detect_axelera() -> DetectionHardware | None: + """Find Axelera Metis accelerators by their device nodes.""" + nodes = sorted(glob(f"{DEV_ROOT}/metis*")) + + if not nodes: + return None + + # the runtime addresses the card through the driver, not per node, so one + # unit covers the accelerator + units = [HardwareUnit(device="axelera:PCIe", label=os.path.basename(nodes[0]))] + return _hardware("axelera", "axelera", "Axelera Metis", 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") @@ -344,6 +357,7 @@ PROBES = ( detect_hailo, detect_memryx, detect_deepx, + detect_axelera, detect_intel_npu, detect_intel_gpu, detect_nvidia_gpu, diff --git a/frigate/detectors/plugins/axelera.py b/frigate/detectors/plugins/axelera.py new file mode 100644 index 0000000000..9510ef78d9 --- /dev/null +++ b/frigate/detectors/plugins/axelera.py @@ -0,0 +1,972 @@ +"""Frigate object detector plugin for the Axelera Metis NPU (`axelera`). + +Axelera AI makes the Metis PCIe/M.2 NPU accelerator. Inference runs through the +vendor's LOW-LEVEL runtime API (``axelera.runtime``, the same direct +"ModelInstance.run" interface used for hailo8l/memryx), NOT through the +GStreamer-based ``create_inference_stream`` app pipeline. + +Why the low-level API instead of the stream: the video pipeline is depth-4 and +clocked by the input arrival interval, so a frame's result round-trips in +~4x the input gap (~150 ms at a ~35 fps aggregate) even though the accelerator +computes a frame in ~21 ms back-to-back. That latency is what Frigate's async +runner reports and what a camera waits on. The direct runtime API is a +synchronous ``ModelInstance.run(inputs, outputs)`` call with no video pipeline, +so per-frame latency is the flat model time (~21 ms for YOLO-NAS M 704x576, +measured), independent of scene activity. + +Model configuration: + - ``path`` = a .zip of the COMPILED model output (containing ``1/model.json`` + + ``1/manifest.json`` + ``1/postprocess_graph.onnx``), local or http(s) + URL (the memryx custom-model pattern); the selected archive extracts + fresh on every start. + - ``labelmap_path`` / ``labelmap``: Frigate's labelmap names the class + index the decode returns, so renames and the coco-80 bus/truck->car + mapping need no plugin-side code. + - ``width`` / ``height``: must equal the compiled model input size; the + plugin raises at init otherwise (no letterboxing). + +The SDK's own baked postprocess graph (``1/postprocess_graph.onnx``) converts +the raw int8 AIPU tensors into boxes + class scores, so the plugin only +dequantizes the heads and runs a thin class-max + NMS. +""" + +import json +import logging +import os +import queue +import shutil +import threading +import zipfile +from typing import ClassVar, Literal + +import cv2 +import numpy as np +from pydantic import ConfigDict, Field + +from frigate.const import MODEL_CACHE_DIR +from frigate.detectors.detection_api import DetectionApi +from frigate.detectors.detector_config import BaseDetectorConfig, ModelTypeEnum +from frigate.util.file import FileLock +from frigate.util.model import xyxy_to_xywh_for_nms +from frigate.util.runtime_deps import ( + Artifact, + ArtifactKind, + RuntimeManifest, + user_site, +) + +logger = logging.getLogger(__name__) + +DETECTOR_KEY = "axelera" + +# The Axelera runtime SDK is installed at first start rather than shipped in +# the image. The wheels come from Axelera's public Artifactory PyPI index +# (readable without credentials) and are pinned by sha256. The runtime wheel +# is auditwheel-repaired (all native deps bundled under axelera_runtime.libs +# and resolved through $ORIGIN RPATH) and the firmware wheel ships the AIPU +# device firmware under axelera/omega so the pip install is fully self +# contained. axelera.runtime imports axelera.runtime2 only inside a try for +# its separate "new runtime API"; the low-level API this plugin uses lives in +# axelera.runtime.objects and works without it, so runtime2 is not installed. +AXELERA_RUNTIME_VERSION = "1.8.0" +AXELERA_FIRMWARE_VERSION = "1.8.0" +AXELERA_PYPI = "https://software.axelera.ai/artifactory/axelera-pypi" + +AXELERA_MANIFEST = RuntimeManifest( + name=DETECTOR_KEY, + version=AXELERA_RUNTIME_VERSION, + artifacts=( + Artifact( + url=( + f"{AXELERA_PYPI}/axelera-runtime/{AXELERA_RUNTIME_VERSION}/" + "axelera_runtime-1.8.0-cp311-cp311-manylinux_2_27_x86_64" + ".manylinux_2_28_x86_64.whl" + ), + sha256="c97b41f427a61ca0fe22dcab38bbf885d12e0a04a0c4f55e87f1f4036a39c869", + kind=ArtifactKind.wheel, + machines=("x86_64",), + ), + Artifact( + url=( + f"{AXELERA_PYPI}/axelera-runtime/{AXELERA_RUNTIME_VERSION}/" + "axelera_runtime-1.8.0-cp311-cp311-manylinux_2_27_aarch64" + ".manylinux_2_28_aarch64.whl" + ), + sha256="5ec8a048fa94d261080c11b840044f5f7b62fce1f0a43ba8733fc880f40d86ef", + kind=ArtifactKind.wheel, + machines=("aarch64",), + ), + Artifact( + url=( + f"{AXELERA_PYPI}/axelera-firmware/{AXELERA_FIRMWARE_VERSION}/" + "axelera_firmware-1.8.0-py3-none-any.whl" + ), + sha256="b926c16383e696c6e2b200b9207a6fa8536e9b1f8a3d261987d3f829c0b37d36", + kind=ArtifactKind.wheel, + ), + ), + import_check="axelera.runtime", +) + +AX_DEFAULT_NETWORK = "yolox-s-coco-onnx" +MAX_ROWS = 20 +# int8 zero-point the compiled inputs are offset by (px - 128); also the +# filler for the padded border. Everything else about a specific compiled +# model (padded input size, content offset, per-output padded/real channels +# and dequant scale/zero-point) is read at init from 1/manifest.json, so the +# decoder works for any compiled YOLO-family model that ships a baked +# postprocess_graph.onnx (verified: YOLO-NAS M and YOLOX-S). +INPUT_Q_ZERO = -128 +# confidence / NMS defaults mirror the SDK DecodeYolo operator defaults +CONF_THRESHOLD = 0.25 +NMS_IOU_THRESHOLD = 0.45 +QUEUE_MAXSIZE = 32 +# ping-pong buffer sets for the producer/consumer pipeline: the producer +# writes one set while the decoder reads a previous one, so the NPU stays +# saturated. 3 sets bound the in-flight frames (pool size must equal the +# decode queue bound). +BUFFER_SETS = 3 + + +# ----------------- Utility Functions ----------------- # + + +def is_url(value: str) -> bool: + """Return True for http(s) URLs.""" + return value.startswith(("http://", "https://")) + + +def _extract_model_zip(zip_path: str, dest_dir: str) -> None: + """Unpack a compiled-model zip into ``dest_dir`` (caller wipes it first).""" + with zipfile.ZipFile(zip_path) as archive: + for member in archive.infolist(): + if member.is_dir(): + continue + target = os.path.realpath(os.path.join(dest_dir, member.filename)) + if not target.startswith(os.path.realpath(dest_dir) + os.sep): + raise ValueError( + f"model archive member escapes target: {member.filename}" + ) + os.makedirs(os.path.dirname(target), exist_ok=True) + with archive.open(member) as source, open(target, "wb") as out: + out.write(source.read()) + + +def _resolve_model_json(path) -> str: + """Resolve ``path`` (a .zip of the compiled output) to its ``model.json``. + + Like the memryx custom-model flow, ``path`` is always a zip of the + compiled output: an http(s) URL downloads into the model cache, a local + zip is verified, and the selected archive extracts fresh on every start + (under a lock, as several processes may start at once), so a replaced + archive is always the model that runs. Anything else raises. + """ + if not path: + raise ValueError( + "axelera: the model path is required (a .zip of the compiled " + "model, local or http(s) URL)." + ) + if is_url(path): + name = os.path.splitext(os.path.basename(path.split("?")[0]))[0] + name = name or AX_DEFAULT_NETWORK + cache = os.path.join(MODEL_CACHE_DIR, "axelera") + zip_path = os.path.join(cache, f"{name}.zip") + if not os.path.isfile(zip_path): + os.makedirs(cache, exist_ok=True) + try: + # function level like runtime_deps: keeps the detector API + # import (and this module) free of the downloader's IPC stack + from frigate.util.downloader import ModelDownloader + + ModelDownloader.download_from_url(path, zip_path) + except OSError as exc: + raise RuntimeError( + f"Failed to download model from {path}: {exc}" + ) from exc + else: + if not path.endswith(".zip"): + raise ValueError( + f"Invalid model path: {path}. Only .zip files are supported. " + "Please provide a .zip model archive." + ) + if not os.path.isfile(path): + raise FileNotFoundError(f"Custom model zip not found: {path}") + zip_path = path + name = os.path.splitext(os.path.basename(path))[0] or AX_DEFAULT_NETWORK + + if not zipfile.is_zipfile(zip_path): + raise ValueError(f"model archive {zip_path} is not a zip file") + cache = os.path.join(MODEL_CACHE_DIR, "axelera") + os.makedirs(cache, exist_ok=True) + extract_dir = os.path.join(cache, f"{name}-extracted") + with FileLock(os.path.join(cache, f".{name}.lock")): + if os.path.isdir(extract_dir): + shutil.rmtree(extract_dir) + _extract_model_zip(zip_path, extract_dir) + return _find_model_json(extract_dir) + + +def _find_model_json(where: str) -> str: + """Locate the runtime's model.json under an extracted model zip. + + ``where`` is the extract directory, so model.json sits either at its root + (``1/model.json`` or ``model.json``) or under the single top-level folder + the archive was wrapped in. Only an unambiguous single match is accepted: + several candidate subdirectories raise instead of returning whichever + sorts first, because silently picking one would run a model the user + never selected. + """ + if not os.path.isdir(where): + raise FileNotFoundError( + f"model.json not found under {where}. The model zip must contain " + "1/model.json at its root or under one top-level folder." + ) + for candidate in ( + os.path.join(where, "1", "model.json"), + os.path.join(where, "model.json"), + ): + if os.path.isfile(candidate): + return candidate + matches: list[str] = [] + for entry in sorted(os.listdir(where)): + sub = os.path.join(where, entry) + if not os.path.isdir(sub): + continue + for candidate in ( + os.path.join(sub, "1", "model.json"), + os.path.join(sub, "model.json"), + ): + if os.path.isfile(candidate): + matches.append(candidate) + break + if len(matches) == 1: + return matches[0] + if len(matches) > 1: + listed = ", ".join(os.path.relpath(m, where) for m in matches) + raise ValueError( + f"multiple compiled models found under {where}: {listed}. The " + "model zip must contain exactly one model.json, at its root or " + "under one top-level folder." + ) + raise FileNotFoundError( + f"model.json not found under {where}. The model zip must contain " + "1/model.json at its root or under one top-level folder." + ) + + +def _nms_per_class(boxes, scores, class_ids, iou_threshold: float) -> np.ndarray: + """Per-class greedy NMS via cv2.dnn.NMSBoxes (C++, GIL-free). + + Boxes are xyxy; xyxy_to_xywh_for_nms converts them to the top-left+size + format NMSBoxes expects. Running NMS per class keeps a confident car from + suppressing an overlapping pedestrian, which class-agnostic NMS does. + + Returns kept indices sorted by descending score. + """ + if len(boxes) == 0: + return np.array([], dtype=np.int64) + + order = np.argsort(-scores) + kept: list[int] = [] + for cls in np.unique(class_ids): + idx = order[class_ids[order] == cls] + if len(idx) == 0: + continue + indices = cv2.dnn.NMSBoxes( + xyxy_to_xywh_for_nms(boxes[idx]).tolist(), + [float(s) for s in scores[idx]], + # inputs are already filtered above CONF_THRESHOLD by the caller + score_threshold=0.0, + nms_threshold=float(iou_threshold), + ) + if isinstance(indices, np.ndarray): + kept.extend(idx[indices.reshape(-1)].tolist()) + elif len(indices) > 0: + kept.extend(idx[np.asarray(indices, dtype=np.int64).reshape(-1)].tolist()) + + if not kept: + return np.array([], dtype=np.int64) + kept_arr = np.asarray(kept, dtype=np.int64) + return kept_arr[np.argsort(-scores[kept_arr])] + + +def _read_model_manifest(model_dir: str) -> dict: + """Read the compiled model's manifest (geometry + dequantize params). + + The compiler writes two JSONs next to the kernel: ``1/model.json`` (padded + kernel I/O only) and ``1/manifest.json`` (original/padded shapes, the exact + content offset inside the padded input, per-output padded channel counts + and dequantize (scale, zero-point) for every head). + """ + candidate = os.path.join(model_dir, "manifest.json") + if not os.path.isfile(candidate): + raise FileNotFoundError( + "compiled manifest not found next to the model kernel. Point " + "path at a zip of a directory compiled by the Voyager SDK that " + "contains 1/model.json + 1/manifest.json + 1/postprocess_graph.onnx." + ) + try: + with open(candidate) as f: + manifest = json.load(f) + except Exception as exc: + raise ValueError( + f"could not parse compiled manifest {candidate}: {exc}" + ) from exc + required = ( + "input_shapes", + "input_shapes_original", + "output_shapes", + "output_shapes_original", + "dequantize_params", + ) + missing = [k for k in required if k not in manifest] + if missing: + raise ValueError( + f"compiled manifest {candidate} missing required keys: {missing}" + ) + return manifest + + +# ----------------- AxeleraRuntimeInference (engine) ----------------- # + + +class _AxeleraRuntimeInference: + """Runs a compiled model via ``axelera.runtime`` on a two-thread pipeline. + + Frigate's AsyncDetectorRunner calls ``submit(connection_id, frame)`` from + the detect thread and ``collect()`` from the result thread. Two engine + threads process frames strictly in order: a producer resizes + + quantizes + runs the model (``ModelInstance.run``, synchronous and flat: + ~21 ms measured), and a consumer feeds the int8 heads through the SDK's + baked ``postprocess_graph.onnx`` while the NPU is already busy on the next + frame (measured throughput ~25 ms/frame, ~40 fps, vs ~29 ms serial). All + ``run`` calls stay on the single producer thread, so in-flight frames never + execute the model in parallel (a known card-wedge condition). Because both + threads consume their queues FIFO, results are paired to requests by + construction - there is no pipeline reorder to guard against. + """ + + def __init__( + self, + model_json, + model_type: ModelTypeEnum, + width: int, + height: int, + pixel_format: str = "bgr", + runtime_factory=None, + onnx_factory=None, + ) -> None: + self._model_type = model_type + self._width, self._height = int(width), int(height) + # the compiled model consumes RGB planes; Frigate delivers BGR when + # input_pixel_format is bgr (flip below), RGB when rgb (no flip) + self._flip_channels = pixel_format == "bgr" + self._frame_q: queue.Queue = queue.Queue(maxsize=QUEUE_MAXSIZE) + self._result_q: queue.Queue = queue.Queue(maxsize=QUEUE_MAXSIZE) + self._stop = threading.Event() + self._producer_thread: threading.Thread | None = None + self._consumer_thread: threading.Thread | None = None + + if runtime_factory is None: + try: + from axelera.runtime import Context + except (ImportError, ModuleNotFoundError) as exc: + raise ImportError( + "Axelera runtime (axelera.runtime) is not available in this " + "environment. Frigate installs it at startup when an Axelera " + "detector is configured; check the startup log for errors." + ) from exc + runtime_factory = Context + if onnx_factory is None: + import onnxruntime as _ort + + onnx_factory = _ort + + model_dir = os.path.dirname(model_json) + self._ctx = runtime_factory() + model = self._ctx.load_model(model_json) + # enumerate and connect to the DeviceInfo explicitly rather than + # letting the runtime pick: it surfaces the lspci/device-node + # mismatch as an empty list instead of a late connect failure + devices = self._ctx.list_devices() + if not devices: + raise RuntimeError( + "axelera: no Metis device enumerated; the runtime needs the " + "raw /dev/metis--- node in the container and " + "the colon-separated name (e.g. /dev/metis-0:1:0) present " + "as a symlink to it. See the install docs for the CDI " + "device spec that provides both, and check that the metis " + "driver is loaded on the host" + ) + # hold the device connection on self: the runtime releases the device + # when the connection object is garbage collected, so it must outlive + # init for as long as the model instance uses it + self._conn = self._ctx.device_connect(devices[0], num_sub_devices=1) + self._instance = self._conn.load_model_instance( + model, num_sub_devices=1, aipu_cores=1 + ) + + manifest = _read_model_manifest(model_dir) + + # input: padded NHWC (1, H, W, 4) exactly as the runtime expects it, + # plus the original NCHW (1, C, H, W) workspace the resize targets + in_padded = [int(x) for x in manifest["input_shapes"][0]] + in_orig = [int(x) for x in manifest["input_shapes_original"][0]] + if len(in_padded) != 4 or in_padded[3] != 4 or in_orig[1] != 3: + raise ValueError( + f"unsupported compiled input: padded {in_padded}, original {in_orig}" + ) + self._padded_h, self._padded_w = in_padded[1], in_padded[2] + self._model_h, self._model_w = in_orig[2], in_orig[3] + + # where the compiler placed the unpadded content inside the padded + # buffer: n_padded_ch_inputs = [[N(lo,hi), H(lo,hi), W(lo,hi), + # C(lo,hi)]] (one entry per input) + npad_data = manifest["n_padded_ch_inputs"] + if isinstance(npad_data, list) and npad_data and isinstance(npad_data[0], list): + npad_data = npad_data[0] + npad = [int(x) for x in npad_data] + if len(npad) == 8: + self._content_top, self._content_left = npad[2], npad[4] + else: + self._content_top = (self._padded_h - self._model_h) // 2 + self._content_left = (self._padded_w - self._model_w) // 2 + + out_padded = [tuple(int(x) for x in o) for o in manifest["output_shapes"]] + out_real = [ + tuple(int(x) for x in o) for o in manifest["output_shapes_original"] + ] + dequant = manifest["dequantize_params"] + if not (len(out_padded) == len(out_real) == len(dequant)): + raise ValueError( + "compiled manifest output counts mismatch: padded " + f"{len(out_padded)}, original {len(out_real)}, " + f"dequant {len(dequant)}" + ) + # per head: real channel count (crop target) + dequant (scale, zp) + self._head_real = [o[1] for o in out_real] + self._head_dequant = [(np.float32(d[0]), np.float32(d[1])) for d in dequant] + self._out_shapes = out_padded + + pp_path = os.path.join(model_dir, "postprocess_graph.onnx") + if not os.path.isfile(pp_path): + raise FileNotFoundError( + f"postprocess_graph.onnx not found next to {model_json}" + ) + # one frame is in flight through the decoder, so a single intra-op + # thread is optimal; extra ORT threads only add context switches + sess_opts = onnx_factory.SessionOptions() + sess_opts.intra_op_num_threads = 1 + self._pp = onnx_factory.InferenceSession( + pp_path, + sess_options=sess_opts, + providers=["CPUExecutionProvider"], + ) + pp_inputs = self._pp.get_inputs() + self._pp_input_names = [i.name for i in pp_inputs] + if len(pp_inputs) != len(self._head_real): + raise ValueError( + "postprocess graph input count mismatch: " + f"{len(pp_inputs)} graph inputs vs " + f"{len(self._head_real)} model outputs" + ) + # Map each postprocess input to the model output whose dequantized NCHW + # crop has exactly that (1, C, H, W) shape: the compile order of the + # raw heads is not guaranteed to match the postprocess graph's input + # order (they differ for YOLOX vs YOLO-NAS), shapes are unique per head. + self._feed_index: list[int] = [] + for pin in pp_inputs: + want = tuple(int(x) for x in pin.shape) + matches = [ + i + for i in range(len(self._head_real)) + if ( + 1, + self._head_real[i], + out_padded[i][1], + out_padded[i][2], + ) + == want + ] + if len(matches) != 1: + raise ValueError( + f"cannot map postprocess input {pin.name} {want} to a " + "unique model output" + ) + self._feed_index.append(matches[0]) + + # pooled model-input + raw-int8-output sets, ping-ponged between + # the producer (runs the model) and the decoder (postprocesses the + # previous frame's outputs while the NPU is busy on the next one). + self._pool: queue.Queue = queue.Queue(maxsize=BUFFER_SETS) + for _ in range(BUFFER_SETS): + in_buf = np.full( + (1, self._padded_h, self._padded_w, 4), + INPUT_Q_ZERO, + np.int8, + ) + out_bufs = [np.zeros(shape, np.int8) for shape in out_padded] + self._pool.put((in_buf, out_bufs)) + self._decode_q: queue.Queue = queue.Queue(maxsize=BUFFER_SETS) + # fp32 postprocess feeds, sized to each graph input and reused: the + # dequantizing copy writes straight into them (copyto fuses the + # NHWC->NCHW transpose with the int8->fp32 cast, and the dequant math + # then runs in place) so the hot path never allocates + self._pp_scratch = { + pin.name: np.zeros( + tuple(int(x) for x in pin.shape), + np.float32, + ) + for pin in pp_inputs + } + if (self._width, self._height) != (self._model_w, self._model_h): + raise ValueError( + f"axelera: model width/height must match the compiled input " + f"({self._model_w}x{self._model_h}); configured " + f"{self._width}x{self._height}" + ) + # per-frame scratch (producer-thread-owned): the resized frame, the + # channel-oriented copy, and the quantized int8 + self._resize_scratch = np.zeros((self._model_h, self._model_w, 3), np.uint8) + self._orient_scratch = np.zeros((self._model_h, self._model_w, 3), np.uint8) + self._quant_scratch = np.zeros((self._model_h, self._model_w, 3), np.int8) + + # -- public contract used by AxeleraDetector -------------------------- + + def submit(self, connection_id: str, frame: np.ndarray) -> None: + """Queue one frame for inference, owned by `connection_id`. + + On overload (input queue full) the frame's INFERENCE is dropped but a + zero-row result is still emitted for `connection_id`: the async runner + contract is strictly 1 submission : 1 result (SHM cleanup + timing + pairing), so raising here would leave the runner waiting for a result + that never arrives. + """ + try: + self._frame_q.put_nowait((connection_id, frame)) + except queue.Full: + logger.warning( + "axelera: inference queue full; dropping frame for %s", + connection_id, + ) + self._push_result(connection_id, np.zeros((MAX_ROWS, 6), np.float32)) + return + self._ensure_threads() + + def _ensure_threads(self) -> None: + if self._consumer_thread is None or not self._consumer_thread.is_alive(): + self._consumer_thread = threading.Thread( + target=self._consumer, name="axelera-decode", daemon=True + ) + self._consumer_thread.start() + if self._producer_thread is None or not self._producer_thread.is_alive(): + self._producer_thread = threading.Thread( + target=self._producer, name="axelera-runtime", daemon=True + ) + self._producer_thread.start() + + def collect(self): + """Return (connection_id, rows) for the next completed inference. + + Returns (None, None) on stop so the async runner can exit cleanly. + """ + while not self._stop.is_set(): + try: + return self._result_q.get(timeout=0.25) + except queue.Empty: + continue + return None, None + + def stop(self) -> None: + """Stop the workers, release the Metis device. Never raises.""" + self._stop.set() + for t in (self._producer_thread, self._consumer_thread): + if t is not None and t.is_alive(): + t.join(timeout=2) + try: + # Context.release() releases the context and every object it + # created (connection, model, instance); there is no per-object + # free() in this runtime API + self._ctx.release() + except Exception: # noqa: BLE001, S110 + pass + logger.info("axelera: runtime inference stopped") + + # -- internals --------------------------------------------------------- + + def _push_result(self, connection_id: str, rows: np.ndarray) -> None: + """Deliver one completion, applying backpressure instead of dropping. + + The async runner contract is strictly 1 submission : 1 delivered + result (SHM cleanup + timing pairing), so an accepted submission must + never lose its completion: if the result worker is behind, the + producing thread blocks until the queue drains. The only consumer is + the runner's result worker (which drains until stop), so the block + self-throttles rather than deadlocking; it is stop-aware so shutdown + stays responsive while the runner is unwinding. + """ + warned = False + while not self._stop.is_set(): + try: + self._result_q.put((connection_id, rows), timeout=0.25) + return + except queue.Full: + if not warned: + logger.warning( + "axelera: result queue full, applying backpressure for %s", + connection_id, + ) + warned = True + logger.warning( + "axelera: stopping, dropping undelivered result for %s", connection_id + ) + + def _producer(self) -> None: + """Resize + quantize + run(); hand raw outputs to the decoder. + + All ``ModelInstance.run`` calls happen on THIS single thread, so the + three in-flight buffer sets never execute the model in parallel (a + known card-wedge condition) while still overlapping with the decoder's + CPU postprocessing. + """ + while not self._stop.is_set(): + try: + connection_id, frame = self._frame_q.get(timeout=0.25) + except queue.Empty: + continue + try: + in_buf, out_bufs = self._pool.get(timeout=1.0) + except queue.Empty: + logger.error("axelera: no free buffers; dropping %s", connection_id) + self._push_result(connection_id, np.zeros((MAX_ROWS, 6), np.float32)) + continue + try: + self._preprocess(frame, in_buf) + self._instance.run([in_buf], out_bufs) + self._decode_q.put_nowait((connection_id, in_buf, out_bufs)) + except Exception: + logger.exception("axelera: inference failed for %s", connection_id) + self._pool.put((in_buf, out_bufs)) + self._push_result(connection_id, np.zeros((MAX_ROWS, 6), np.float32)) + + def _consumer(self) -> None: + """Decode the raw int8 heads (dequant + onnx + NMS) to (20,6) rows.""" + while not self._stop.is_set(): + try: + connection_id, in_buf, out_bufs = self._decode_q.get(timeout=0.25) + except queue.Empty: + continue + try: + rows = self._decode(out_bufs) + except Exception: + logger.exception("axelera: decode failed for %s", connection_id) + rows = np.zeros((MAX_ROWS, 6), np.float32) + self._pool.put((in_buf, out_bufs)) + self._push_result(connection_id, rows) + + def _preprocess(self, frame: np.ndarray, buf: np.ndarray) -> None: + """Resize the source (H, W, 3) into the model input and quantize. + + All three passes write into preallocated scratch: resize straight into + the scratch buffer (cv2 ``dst=``), an in-place channel swap (no strided + ``[..., ::-1]`` copy at write time), and an out= subtract for the + int8 quantize. + """ + cv2.resize( + frame, + (self._model_w, self._model_h), + interpolation=cv2.INTER_LINEAR, + dst=self._resize_scratch, + ) + if self._flip_channels: + # BGR -> RGB: Frigate delivers BGR under the documented + # input_pixel_format: bgr, while the compiled input expects RGB + # semantics; with input_pixel_format: rgb the frame already + # carries RGB and is only swapped during quantisation + cv2.cvtColor( + self._resize_scratch, + cv2.COLOR_BGR2RGB, + dst=self._orient_scratch, + ) + src_u8 = self._orient_scratch + else: + src_u8 = self._resize_scratch + # uint8 - 128 wraps mod 256, then cast to int8 == px - 128 (exactly); + # dtype= on subtract keeps it one pass with no intermediate array + np.subtract( + src_u8, + np.uint8(128), + out=self._quant_scratch, + dtype=np.int8, + casting="unsafe", + ) + # the frame fills the whole content region; anything the compiler + # padded around it stays at the zero-point filler + buf[0, :, :, :].fill(INPUT_Q_ZERO) + dst = buf[ + 0, + self._content_top : self._content_top + self._model_h, + self._content_left : self._content_left + self._model_w, + ] + dst[..., 0:3] = self._quant_scratch + dst[..., 3] = 127 + + def _decode(self, out_bufs) -> np.ndarray: + """Dequantize heads + postprocess graph + conf threshold + NMS -> rows.""" + # dequantize + crop to the real channel count + transpose NHWC -> NCHW + # without temporaries: copyto casts int8 -> fp32 (lossless) while it + # does the strided transpose copy into the reused scratch feed, then + # the zp/scale dequant math runs in place on the contiguous NCHW result + feeds = {} + for feed_i, name in enumerate(self._pp_input_names): + idx = self._feed_index[feed_i] + obuf = out_bufs[idx] + scale, zp = self._head_dequant[idx] + real = self._head_real[idx] + dst = self._pp_scratch[name] + np.copyto(dst, obuf[..., :real].transpose(0, 3, 1, 2), casting="unsafe") + np.subtract(dst, zp, out=dst) + np.multiply(dst, scale, out=dst) + feeds[name] = dst + + pp_out = self._pp.run(None, feeds) + if len(pp_out) == 2: + # two outputs: class scores (sigmoid baked into the graph) + xyxy + class_scores, boxes = pp_out + scores = class_scores[0] # (8400, 80) + boxes = boxes[0] + # one argmax pass over the class axis; conf gathers the argmaxed + # cells instead of a second full max(axis=1) reduction + cls_idx = scores.argmax(axis=1).astype(np.int32) + conf = scores[np.arange(cls_idx.shape[0]), cls_idx] + else: + # one fused output: [cx, cy, w, h, (objectness), C class scores]. + # The SDK DecodeYolo emits CENTER-format boxes (confirmed by the + # vendor's axruntime_yolo11_onnxruntime.py example), absolute in + # the model input space. + raw = pp_out[0][0] + cx, cy, w, h = raw[:, 0], raw[:, 1], raw[:, 2], raw[:, 3] + boxes = np.stack([cx - w / 2, cy - h / 2, cx + w / 2, cy + h / 2], axis=1) + cls = raw[:, 4:] + if self._model_type == ModelTypeEnum.yolox: + cls = raw[:, 5:] * raw[:, 4:5] # YOLOX: score = obj * cls + if cls.shape[1] == 0: + return np.zeros((MAX_ROWS, 6), np.float32) + cls_idx = cls.argmax(axis=1).astype(np.int32) + conf = cls[np.arange(cls_idx.shape[0]), cls_idx] + keep = conf > CONF_THRESHOLD + boxes, conf, cls_idx = boxes[keep], conf[keep], cls_idx[keep] + if len(conf) == 0: + return np.zeros((MAX_ROWS, 6), np.float32) + # boxes are already in the frame space: the input is not letterboxed + # (width/height must equal the compiled input size) + x0i = np.clip(boxes[:, 0], 0, self._width) + x1i = np.clip(boxes[:, 2], 0, self._width) + y0i = np.clip(boxes[:, 1], 0, self._height) + y1i = np.clip(boxes[:, 3], 0, self._height) + # NMS on the model-space boxes (scale-invariant IoU), per class + keep_nms = _nms_per_class(boxes, conf, cls_idx, NMS_IOU_THRESHOLD) + rows = [] + for i in keep_nms: + score = float(conf[i]) + rows.append( + ( + int(cls_idx[i]), + score, + float(y0i[i]) / self._height, + float(x0i[i]) / self._width, + float(y1i[i]) / self._height, + float(x1i[i]) / self._width, + ) + ) + rows.sort(key=lambda r: r[1], reverse=True) + out = np.zeros((MAX_ROWS, 6), np.float32) + n = min(len(rows), MAX_ROWS) + for i in range(n): + out[i] = rows[i] + return out + + +# ----------------- AxeleraDetector Class ----------------- # + + +class AxeleraDetector(DetectionApi): + type_key = DETECTOR_KEY + supported_models = [ + ModelTypeEnum.yolox, + ModelTypeEnum.yolonas, + ModelTypeEnum.yologeneric, + ] + runtime_manifest = AXELERA_MANIFEST + + def __init__(self, detector_config: "AxeleraDetectorConfig"): + model = detector_config.model + if model is None: + raise ValueError( + "axelera: the model block is required (a .zip of the compiled " + "model output, local or http(s) URL)." + ) + # the runtime wheels install into the user site at first start + self.activate_dependencies() + _ensure_runtime_enumeration_tools() + _point_runtime_at_installed_firmware() + super().__init__(detector_config) + + self.width = model.width + self.height = model.height + + self.working_model = _resolve_model_json(model.path) + + self._inference = _AxeleraRuntimeInference( + model_json=self.working_model, + model_type=model.model_type, + width=self.width, + height=self.height, + pixel_format=str( + getattr(model.input_pixel_format, "value", model.input_pixel_format) + ), + ) + logger.info( + "axelera: runtime model=%s (%dx%d input, frame %dx%d) ready", + self.working_model, + self._inference._model_w, + self._inference._model_h, + self.width, + self.height, + ) + + # --- async API (AsyncDetectorRunner, same contract as the memryx plugin) --- + + def send_input(self, connection_id, tensor_input: np.ndarray) -> None: + """Preprocess (if needed) and send a frame to the Metis runtime.""" + if tensor_input is None: + raise ValueError("[send_input] No image data provided for inference") + + frame = np.asarray(tensor_input) + if frame.ndim == 4: + # Frigate feeds a batched NHWC frame (1, H, W, C); drop the batch dim + if frame.shape[0] != 1: + raise ValueError(f"Unexpected batch dimension in frame: {frame.shape}") + frame = frame[0] + # BGR uint8 for the model (resize/quantization happen in the + # worker). This must be a REAL copy: the tensor backs the camera's + # shared memory, which the runner unmaps once this frame's result is + # collected, potentially before the producer thread reads a queued + # frame (np.ascontiguousarray would be a no-op on a contiguous view). + frame = np.array(frame, dtype=np.uint8, copy=True) + self._inference.submit(connection_id, frame) + + def receive_output(self): + """Return (connection_id, rows) for the next completed inference.""" + return self._inference.collect() + + def shutdown(self) -> None: + """Gracefully shut down the runtime and release the Metis device.""" + self._inference.stop() + + def detect_raw(self, tensor_input: np.ndarray): + """Synchronous path unused: axelera uses the async detector contract.""" + return 0 # type: ignore[override] + + def __del__(self): + # destructor must not raise + try: + if getattr(self, "_inference", None) is not None: + self.shutdown() + except Exception: # noqa: S110, BLE001 + pass + + +def _ensure_runtime_enumeration_tools() -> None: + """Provide minimal lspci/lsmod stand-ins when the image lacks them. + + The runtime's list_devices() shells out to ``lspci`` (PCI count, matched + against the driver count) and ``lsmod`` (AIPU driver presence) through + popen and greps their output. Both only read /sys and /proc, which are + readable in the container, so when the real tools are absent (slim image) + the plugin supplies shell stand-ins in a directory it owns and prepends + that to PATH. The real tools always win when installed. + """ + if shutil.which("lspci") is not None and shutil.which("lsmod") is not None: + return + bin_dir = os.path.join(MODEL_CACHE_DIR, DETECTOR_KEY, "bin") + try: + os.makedirs(bin_dir, exist_ok=True) + scripts = { + # lsmod(8)-shaped output built from /proc/modules + "lsmod": ( + "#!/bin/sh\n" + 'echo "Module Size Used by"\n' + "while read -r name size used _; do\n" + ' printf \'%-19s %8s %s\\n\' "$name" "$size" "$used"\n' + "done < /proc/modules\n" + ), + # lspci(8)-shaped Device lines from sysfs; the runtime matches the + # vendor:device hex pair (1f9d for Metis) + "lspci": ( + "#!/bin/sh\n" + "for dev in /sys/bus/pci/devices/*; do\n" + ' [ -r "$dev/vendor" ] || continue\n' + ' read -r vendor < "$dev/vendor"\n' + ' read -r device < "$dev/device"\n' + " printf '%s Device %s:%s\\n' \"${dev##*/}\" " + '"${vendor#0x}" "${device#0x}"\n' + "done\n" + ), + } + for name, body in scripts.items(): + if shutil.which(name) is not None: + continue + path = os.path.join(bin_dir, name) + with open(path, "w") as f: + f.write(body) + os.chmod(path, 0o755) + os.environ["PATH"] = bin_dir + os.pathsep + os.environ.get("PATH", "") + logger.debug("axelera: prepended runtime tools dir %s to PATH", bin_dir) + except OSError as exc: + logger.warning( + "axelera: could not install runtime enumeration helpers in %s: %s", + bin_dir, + exc, + ) + + +def _point_runtime_at_installed_firmware() -> None: + """Export the firmware paths the runtime's native library looks for. + + The pip-installed layout keeps the AIPU device firmware under the user + site's ``axelera/omega`` directory (from the axelera-firmware wheel). The + native device library resolves firmware through ``AXELERA_DEVICE_DIR`` (and + the explicit stage0/elf variables), and has no default inside the Frigate + image, so point it at the installed copy before the runtime is imported. + """ + if os.environ.get("AXELERA_DEVICE_DIR"): + return + omega = os.path.join(str(user_site()), "axelera", "omega") + if not os.path.isdir(omega): + return + os.environ["AXELERA_DEVICE_DIR"] = omega + os.environ.setdefault( + "AIPU_FIRMWARE_OMEGA", os.path.join(omega, "bin", "start_axelera_runtime.elf") + ) + os.environ.setdefault( + "AIPU_RUNTIME_STAGE0_OMEGA", + os.path.join(omega, "bin", "start_axelera_runtime_stage0.bin"), + ) + + +# ----------------- AxeleraDetectorConfig Class ----------------- # + + +class AxeleraDetectorConfig(BaseDetectorConfig): + model_config = ConfigDict(title="Axelera Metis") + + # the Metis card can only be opened by one process + shareable: ClassVar[bool] = False + + type: Literal[DETECTOR_KEY] + device: str = Field( + default="PCIe", + title="Device Path", + description="The Metis device to run inference on (e.g. 'PCIe').", + ) diff --git a/frigate/object_detection/base.py b/frigate/object_detection/base.py index 14805086c9..0ab975df6c 100644 --- a/frigate/object_detection/base.py +++ b/frigate/object_detection/base.py @@ -277,7 +277,8 @@ class AsyncDetectorRunner(FrigateProcess): self._frame_manager = SharedMemoryFrameManager() self._publisher = ObjectDetectorPublisher() self._detector = AsyncLocalObjectDetector( - detector_config=self.detector_config, stop_event=self.stop_event + detector_config=self.detector_config, + stop_event=self.stop_event, ) for name in self.cameras: @@ -366,8 +367,8 @@ class ObjectDetectProcess: self.detection_start.value = 0.0 - # Async path for MemryX - if self.detector_config.type == "memryx": + # Async path for MemryX and Axelera (send/receive contract) + if self.detector_config.type in ("memryx", "axelera"): self.detect_process = AsyncDetectorRunner( f"frigate.detector:{self.name}", self.detection_queue, diff --git a/frigate/stats/hardware.py b/frigate/stats/hardware.py index dbe18cc548..1738a8409e 100644 --- a/frigate/stats/hardware.py +++ b/frigate/stats/hardware.py @@ -21,6 +21,7 @@ from frigate.detectors.hardware import DEV_ROOT, hardware_prober from frigate.util.services import ( get_amd_gpu_stats, get_axcl_npu_stats, + get_axelera_board_temp, get_cpu_stats, get_hailo_temps, get_intel_gpu_stats, @@ -481,5 +482,7 @@ def get_hardware_temperatures(detector_type: str) -> list[float | None]: elif detector_type == "hailo": hailo_temps = get_hailo_temps() return [hailo_temps[name] for name in sorted(hailo_temps.keys())] + elif detector_type == "axelera": + return [get_axelera_board_temp()] return [] diff --git a/frigate/test/test_axelera_detector.py b/frigate/test/test_axelera_detector.py new file mode 100755 index 0000000000..319e288108 --- /dev/null +++ b/frigate/test/test_axelera_detector.py @@ -0,0 +1,558 @@ +"""Offline unit tests for the Axelera Metis detector plugin. + +The runtime and onnxruntime objects are constructor-injected, so these cover +the decode paths (fused cxcywh and two-output scores+boxes), the raw class +index contract, the zip-only model resolution (download, fresh extraction, +rejection of anything else), and the zip extraction escape guard, without +the vendor SDK or a Metis card. +""" + +import json +import os +import shutil +import sys +import tempfile +import types +import unittest +import zipfile +from typing import ClassVar +from unittest.mock import patch + +import numpy as np + +from frigate.detectors.detector_config import ModelTypeEnum +from frigate.detectors.plugins import axelera as ax + +MANIFEST = { + "input_shapes": [[1, 642, 656, 4]], + "input_shapes_original": [[1, 3, 640, 640]], + "n_padded_ch_inputs": [[0, 1, 1, 641, 0, 656, 0, 4]], + "output_shapes": [[1, 80, 80, 128]], + "output_shapes_original": [[1, 85, 80, 80]], + "dequantize_params": [[0.01, -128]], +} + + +def _make_compiled_model(root: str) -> str: + """Write a fake compiled-model tree, return the model_json path.""" + kernel = os.path.join(root, "fakenet", "1") + os.makedirs(kernel) + with open(os.path.join(kernel, "manifest.json"), "w") as f: + json.dump(MANIFEST, f) + with open(os.path.join(kernel, "model.json"), "w") as f: + json.dump({"inputs": []}, f) + with open(os.path.join(kernel, "postprocess_graph.onnx"), "wb") as f: + f.write(b"fake") + return os.path.join(kernel, "model.json") + + +def _make_model_zip(zip_path: str, kernel_payload: bytes = b"fake") -> None: + """Write a compiled-model zip: fakenet/1/{model.json,manifest.json,onnx}.""" + with zipfile.ZipFile(zip_path, "w") as archive: + archive.writestr("fakenet/1/model.json", json.dumps({"inputs": []})) + archive.writestr("fakenet/1/manifest.json", json.dumps(MANIFEST)) + archive.writestr("fakenet/1/postprocess_graph.onnx", kernel_payload) + + +class _FakeInput: + def __init__(self, name, shape): + self.name = name + self.shape = shape + + +class _FakeInstance: + def run(self, ins, outs): + pass + + +class _FakeConn: + def load_model_instance(self, model, num_sub_devices=1, aipu_cores=1): + return _FakeInstance() + + +class _FakeContext: + def load_model(self, path): + return object() + + def list_devices(self): + return ["metis-0:1:0"] + + def device_connect(self, *_args, **_kwargs): + return _FakeConn() + + def release(self): + pass + + +class _FakeSessionOptions: + def __init__(self): + self.intra_op_num_threads = 0 + + +class _FakeOnnx: + """Postprocess-graph fake with a programmable output. + + ``outputs`` is what ``run`` returns; the feed dict keys/shapes are + recorded so tests can assert the plugin dequantized and laid out the + heads correctly before feeding the graph. Mirrors the onnxruntime + module surface the plugin uses: SessionOptions() and InferenceSession(). + """ + + session = None + outputs: list = [] + + def __init__(self, path, sess_options=None, providers=None): + self._inputs = [_FakeInput("head0", [1, 85, 80, 80])] + _FakeOnnx.session = self + + def get_inputs(self): + return self._inputs + + def run(self, _out, feeds): + self.feeds = feeds + return _FakeOnnx.outputs + + @staticmethod + def SessionOptions(): + return _FakeSessionOptions() + + @staticmethod + def InferenceSession(path, sess_options=None, providers=None): + return _FakeOnnx(path, sess_options, providers) + + +class _FakeModelDownloader: + """The ModelDownloader stand-in bound into the stubbed downloader module.""" + + calls: ClassVar[list] = [] + + @staticmethod + def download_from_url(download_url, save_path, silent=False): + _FakeModelDownloader.calls.append((download_url, save_path)) + _make_model_zip(save_path) + + +def _make_engine(root: str, outputs, model_type=ModelTypeEnum.yolox): + model_json = _make_compiled_model(root) + _FakeOnnx.outputs = outputs + return ax._AxeleraRuntimeInference( + model_json=model_json, + model_type=model_type, + width=640, + height=640, + runtime_factory=_FakeContext, + onnx_factory=_FakeOnnx, + ) + + +def _fused_row(cx, cy, w, h, obj, cls0): + """Build a (1, 8400, 85) fused output with one detection in row 0.""" + out = np.zeros((1, 8400, 85), np.float32) + out[0, 0, 0:5] = [cx, cy, w, h, obj] + out[0, 0, 5] = cls0 + return [out] + + +class TestDecodeFused(unittest.TestCase): + """One fused graph output: [cx, cy, w, h, obj, 80 class scores].""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + # a box centered in the 640 model space: model xyxy [288, 288, 352, 352] + self.engine = _make_engine( + self.tmp.name, _fused_row(320, 320, 64, 64, 1.0, 0.9) + ) + + def tearDown(self): + self.tmp.cleanup() + + def test_dequantizes_heads_into_graph_feed(self): + out_bufs = [np.full((1, 80, 80, 128), -128, np.int8)] + self.engine._decode(out_bufs) + ((name, feed),) = _FakeOnnx.session.feeds.items() + self.assertEqual(name, "head0") + self.assertEqual(feed.shape, (1, 85, 80, 80)) + # (q - zero_point) * scale with q=-128, zp=-128, scale=0.01 + self.assertAlmostEqual(float(feed[0, 0, 0, 0]), 0.0, places=6) + + def test_row_shape_and_dtype(self): + out_bufs = [np.zeros((1, 80, 80, 128), np.int8)] + rows = self.engine._decode(out_bufs) + self.assertEqual(rows.shape, (ax.MAX_ROWS, 6)) + self.assertEqual(rows.dtype, np.float32) + + def test_boxes_map_straight_to_the_frame_space(self): + # no letterbox: the frame is the model input (640x640), so model + # xyxy maps to normalized frame coords 1:1 + out_bufs = [np.zeros((1, 80, 80, 128), np.int8)] + rows = self.engine._decode(out_bufs) + class_id, score, ymin, xmin, ymax, xmax = rows[0] + self.assertEqual(int(class_id), 0) # raw class index, not a name + self.assertAlmostEqual(float(score), 0.9, places=5) # obj * cls + self.assertAlmostEqual(float(xmin), 288 / 640, places=4) + self.assertAlmostEqual(float(xmax), 352 / 640, places=4) + self.assertAlmostEqual(float(ymin), 288 / 640, places=4) + self.assertAlmostEqual(float(ymax), 352 / 640, places=4) + self.assertLess(float(ymin), float(ymax)) + self.assertLess(float(xmin), float(xmax)) + + def test_class_index_passes_through_unmapped(self): + # detections carry the raw class index; Frigate's labelmap names it + out = np.zeros((1, 8400, 85), np.float32) + out[0, 0, 0:5] = [320, 320, 64, 64, 1.0] + out[0, 0, 5 + 5] = 0.9 # class 5 + tmp = tempfile.TemporaryDirectory() + try: + engine = _make_engine(tmp.name, [out]) + rows = engine._decode([np.zeros((1, 80, 80, 128), np.int8)]) + self.assertEqual(int(rows[0][0]), 5) + finally: + tmp.cleanup() + + def test_yolox_score_multiplies_objectness(self): + out = _fused_row(320, 320, 64, 64, 0.5, 0.9) # obj 0.5 * cls 0.9 + tmp = tempfile.TemporaryDirectory() + try: + engine = _make_engine(tmp.name, out, ModelTypeEnum.yolox) + rows = engine._decode([np.zeros((1, 80, 80, 128), np.int8)]) + self.assertAlmostEqual(float(rows[0][1]), 0.45, places=5) + finally: + tmp.cleanup() + + def test_yolonas_score_skips_objectness(self): + out = np.zeros((1, 8400, 84), np.float32) + out[0, 0, :4] = [320, 320, 64, 64] + out[0, 0, 4] = 0.9 # class 0, with no objectness column + tmp = tempfile.TemporaryDirectory() + try: + engine = _make_engine(tmp.name, [out], ModelTypeEnum.yolonas) + rows = engine._decode([np.zeros((1, 80, 80, 128), np.int8)]) + self.assertEqual(int(rows[0][0]), 0) + self.assertAlmostEqual(float(rows[0][1]), 0.9, places=5) + finally: + tmp.cleanup() + + def test_below_conf_threshold_yields_zero_rows(self): + tmp = tempfile.TemporaryDirectory() + try: + engine = _make_engine(tmp.name, _fused_row(320, 320, 64, 64, 1.0, 0.1)) + rows = engine._decode([np.zeros((1, 80, 80, 128), np.int8)]) + self.assertEqual(float(np.abs(rows).sum()), 0.0) + finally: + tmp.cleanup() + + +class TestDecodeTwoOutputs(unittest.TestCase): + """Two graph outputs: (1, 8400, 80) class scores + (1, 8400, 4) xyxy.""" + + def test_scores_and_boxes_branch(self): + tmp = tempfile.TemporaryDirectory() + try: + scores = np.zeros((1, 8400, 80), np.float32) + scores[0, 7, 0] = 0.8 # person at row 7 + boxes = np.zeros((1, 8400, 4), np.float32) + boxes[0, 7] = [288, 288, 352, 352] # xyxy, absolute model space + engine = _make_engine(tmp.name, [scores, boxes]) + rows = engine._decode([np.zeros((1, 80, 80, 128), np.int8)]) + self.assertEqual(int(rows[0][0]), 0) + self.assertAlmostEqual(float(rows[0][1]), 0.8, places=5) + self.assertAlmostEqual(float(rows[0][2]), 288 / 640, places=4) + self.assertAlmostEqual(float(rows[0][4]), 352 / 640, places=4) + finally: + tmp.cleanup() + + +class TestModelDimensions(unittest.TestCase): + """width/height must equal the compiled input; no letterboxing.""" + + def test_mismatched_dimensions_raise(self): + tmp = tempfile.TemporaryDirectory() + try: + model_json = _make_compiled_model(tmp.name) + with self.assertRaises(ValueError): + ax._AxeleraRuntimeInference( + model_json=model_json, + model_type=ModelTypeEnum.yolox, + width=704, + height=576, + runtime_factory=_FakeContext, + onnx_factory=_FakeOnnx, + ) + finally: + tmp.cleanup() + + def test_matching_dimensions_initialize(self): + tmp = tempfile.TemporaryDirectory() + try: + model_json = _make_compiled_model(tmp.name) + engine = ax._AxeleraRuntimeInference( + model_json=model_json, + model_type=ModelTypeEnum.yolox, + width=640, + height=640, + runtime_factory=_FakeContext, + onnx_factory=_FakeOnnx, + ) + self.assertEqual((engine._width, engine._height), (640, 640)) + finally: + tmp.cleanup() + + +class TestZipGuard(unittest.TestCase): + def test_member_escaping_target_raises(self): + tmp = tempfile.TemporaryDirectory() + try: + zip_path = os.path.join(tmp.name, "evil.zip") + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("../escaped.txt", "boom") + with self.assertRaises(ValueError): + ax._extract_model_zip(zip_path, os.path.join(tmp.name, "out")) + self.assertFalse(os.path.exists(os.path.join(tmp.name, "escaped.txt"))) + finally: + tmp.cleanup() + + def test_normal_members_extract(self): + tmp = tempfile.TemporaryDirectory() + try: + zip_path = os.path.join(tmp.name, "ok.zip") + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("net/1/model.json", "{}") + zf.writestr("net/1/manifest.json", "{}") + dest = os.path.join(tmp.name, "out") + ax._extract_model_zip(zip_path, dest) + self.assertTrue( + os.path.isfile(os.path.join(dest, "net", "1", "model.json")) + ) + finally: + tmp.cleanup() + + +class TestFindModelJson(unittest.TestCase): + def test_direct_and_wrapper_dir(self): + tmp = tempfile.TemporaryDirectory() + try: + root = tmp.name + # direct: root/1/model.json + os.makedirs(os.path.join(root, "direct", "1")) + direct = os.path.join(root, "direct", "1", "model.json") + open(direct, "w").write("{}") + self.assertEqual(ax._find_model_json(os.path.join(root, "direct")), direct) + # single wrapper dir: root/wrap/net/1/model.json + wrapped = os.path.join(root, "wrap", "net", "1") + os.makedirs(wrapped) + hit = os.path.join(wrapped, "model.json") + open(hit, "w").write("{}") + self.assertEqual(ax._find_model_json(os.path.join(root, "wrap")), hit) + finally: + tmp.cleanup() + + def test_ambiguous_subdirs_raise(self): + tmp = tempfile.TemporaryDirectory() + try: + for net in ("aaa", "bbb"): + sub = os.path.join(tmp.name, net, "1") + os.makedirs(sub) + open(os.path.join(sub, "model.json"), "w").write("{}") + with self.assertRaises(ValueError): + ax._find_model_json(tmp.name) + finally: + tmp.cleanup() + + def test_missing_raises_filenotfound(self): + tmp = tempfile.TemporaryDirectory() + try: + with self.assertRaises(FileNotFoundError): + ax._find_model_json(tmp.name) + finally: + tmp.cleanup() + + +class TestRuntimeToolShims(unittest.TestCase): + def test_ensure_runtime_enumeration_tools_only_installs_missing(self): + tmp = tempfile.TemporaryDirectory() + saved_path = os.environ["PATH"] + # resolve sh before PATH is emptied: the shim check below must still + # find a shell once PATH points only at the empty temp dir + sh_bin = shutil.which("sh") or "/bin/sh" + try: + os.environ["PATH"] = tmp.name # nothing on PATH + with tempfile.TemporaryDirectory() as cache: + orig_cache = ax.MODEL_CACHE_DIR + ax.MODEL_CACHE_DIR = cache + try: + ax._ensure_runtime_enumeration_tools() + finally: + ax.MODEL_CACHE_DIR = orig_cache + bindir = os.path.join(cache, "axelera", "bin") + for tool in ("lspci", "lsmod"): + self.assertTrue(os.access(os.path.join(bindir, tool), os.X_OK)) + self.assertTrue(os.environ["PATH"].startswith(bindir)) + # the shims are valid sh and produce their /proc, /sys reads + import subprocess + + out = subprocess.run( + [sh_bin, os.path.join(bindir, "lsmod")], + capture_output=True, + text=True, + ) + self.assertEqual(out.returncode, 0) + self.assertIn("Module", out.stdout.splitlines()[0]) + finally: + os.environ["PATH"] = saved_path + tmp.cleanup() + + +class TestModelResolve(unittest.TestCase): + """`path` is a .zip of the compiled output, local or http(s) URL.""" + + def _resolve(self, path): + _FakeModelDownloader.calls = [] + # the plugin imports the downloader lazily at the call site (its module + # chain pulls the IPC stack), so the test binds a stub module instead + stub = types.ModuleType("frigate.util.downloader") + setattr(stub, "ModelDownloader", _FakeModelDownloader) + tmp = tempfile.TemporaryDirectory() + saved_cache = ax.MODEL_CACHE_DIR + ax.MODEL_CACHE_DIR = os.path.join(tmp.name, "cache") + try: + with patch.dict(sys.modules, {"frigate.util.downloader": stub}): + got = ax._resolve_model_json(path) + finally: + ax.MODEL_CACHE_DIR = saved_cache + tmp.cleanup() + return _FakeModelDownloader.calls, got + + def test_https_url_downloads_and_resolves(self): + calls, got = self._resolve("https://example.com/models/fakenet.zip") + + self.assertEqual(calls[0][0], "https://example.com/models/fakenet.zip") + self.assertTrue(calls[0][1].endswith(os.path.join("axelera", "fakenet.zip"))) + self.assertTrue(got.endswith(os.path.join("fakenet", "1", "model.json"))) + + def test_url_without_filename_falls_back_to_default_name(self): + calls, _ = self._resolve("https://example.com/") + + self.assertTrue(calls[0][1].endswith("yolox-s-coco-onnx.zip")) + + def test_scheme_less_host_is_rejected(self): + # no scheme means a local path; a host-looking string is not a file + with self.assertRaises(FileNotFoundError): + self._resolve("www.example.com/models/fakenet.zip") + + def test_local_zip_resolves(self): + tmp = tempfile.TemporaryDirectory() + try: + zip_path = os.path.join(tmp.name, "fakenet.zip") + _make_model_zip(zip_path) + _FakeModelDownloader.calls = [] + stub = types.ModuleType("frigate.util.downloader") + setattr(stub, "ModelDownloader", _FakeModelDownloader) + saved_cache = ax.MODEL_CACHE_DIR + ax.MODEL_CACHE_DIR = os.path.join(tmp.name, "cache") + try: + with patch.dict(sys.modules, {"frigate.util.downloader": stub}): + got = ax._resolve_model_json(zip_path) + finally: + ax.MODEL_CACHE_DIR = saved_cache + self.assertEqual(_FakeModelDownloader.calls, []) + self.assertTrue(got.endswith(os.path.join("fakenet", "1", "model.json"))) + finally: + tmp.cleanup() + + def test_non_zip_path_is_rejected(self): + tmp = tempfile.TemporaryDirectory() + try: + model_json = _make_compiled_model(tmp.name) + with self.assertRaises(ValueError): + self._resolve(model_json) + with self.assertRaises(ValueError): + self._resolve(os.path.join(tmp.name, "fakenet")) + finally: + tmp.cleanup() + + def test_missing_local_zip_raises_filenotfound(self): + with self.assertRaises(FileNotFoundError): + self._resolve("/nope/fakenet.zip") + + +class TestFreshExtract(unittest.TestCase): + """The selected archive extracts fresh on every start.""" + + def _resolve(self, path, tmp_name): + stub = types.ModuleType("frigate.util.downloader") + setattr(stub, "ModelDownloader", _FakeModelDownloader) + saved_cache = ax.MODEL_CACHE_DIR + ax.MODEL_CACHE_DIR = os.path.join(tmp_name, "cache") + try: + with patch.dict(sys.modules, {"frigate.util.downloader": stub}): + return ax._resolve_model_json(path) + finally: + ax.MODEL_CACHE_DIR = saved_cache + + def test_replaced_archive_becomes_the_model_that_runs(self): + tmp = tempfile.TemporaryDirectory() + try: + _FakeModelDownloader.calls = [] + self._resolve("https://example.com/models/fakenet.zip", tmp.name) + zip_path = _FakeModelDownloader.calls[0][1] + # the archive at the cache path is replaced (re-fetched upstream, + # a pre-seeded image, a mirror refresh) + _make_model_zip(zip_path, kernel_payload=b"replaced") + self._resolve("https://example.com/models/fakenet.zip", tmp.name) + + extract_dir = os.path.join( + tmp.name, "cache", "axelera", "fakenet-extracted" + ) + with open( + os.path.join(extract_dir, "fakenet", "1", "postprocess_graph.onnx"), + "rb", + ) as f: + self.assertEqual(f.read(), b"replaced") + finally: + tmp.cleanup() + + def test_same_archive_does_not_download_twice(self): + tmp = tempfile.TemporaryDirectory() + try: + _FakeModelDownloader.calls = [] + self._resolve("https://example.com/models/fakenet.zip", tmp.name) + self.assertEqual(len(_FakeModelDownloader.calls), 1) + self._resolve("https://example.com/models/fakenet.zip", tmp.name) + + # the zip is cached (no second download), the extract is fresh + self.assertEqual(len(_FakeModelDownloader.calls), 1) + finally: + tmp.cleanup() + + +class TestBackpressure(unittest.TestCase): + """Overload keeps the 1 submission : 1 result runner contract.""" + + def _engine(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + return _make_engine(tmp.name, []) + + def test_full_input_queue_still_delivers_zero_rows(self): + engine = self._engine() + frame = np.zeros((1, 640, 640, 3), np.uint8) + for i in range(ax.QUEUE_MAXSIZE): + engine._frame_q.put_nowait((f"cam{i}", frame)) + + overflow = "overflow-cam" + engine.submit(overflow, frame) + + cid, rows = engine._result_q.get_nowait() + self.assertEqual(cid, overflow) + self.assertEqual(rows.shape, (ax.MAX_ROWS, 6)) + self.assertTrue(np.all(rows == 0)) + + def test_stop_ends_collect_and_releases_context(self): + engine = self._engine() + engine.stop() + + self.assertEqual(engine.collect(), (None, None)) + + +if __name__ == "__main__": + unittest.main() diff --git a/frigate/test/test_detector_hardware.py b/frigate/test/test_detector_hardware.py index c438cb8f4f..fbc1817597 100644 --- a/frigate/test/test_detector_hardware.py +++ b/frigate/test/test_detector_hardware.py @@ -227,6 +227,14 @@ class TestAccelerators(HardwareProbeTestCase): def test_no_deepx_is_reported_without_a_node(self): self.assertNotIn("deepx", self.probe()) + def test_an_axelera_metis_is_a_single_unit(self): + write(os.path.join(self.dev_root, "metis0")) + + axelera = self.probe()["axelera"] + + self.assertEqual(axelera.units[0].device, "axelera:PCIe") + self.assertEqual(axelera.count, 1) + def test_a_supported_rockchip_soc_is_reported(self): write( os.path.join(self.proc_root, "device-tree", "compatible"), diff --git a/frigate/test/test_detector_stats.py b/frigate/test/test_detector_stats.py index db7f265e0d..bf09eb3884 100644 --- a/frigate/test/test_detector_stats.py +++ b/frigate/test/test_detector_stats.py @@ -34,6 +34,41 @@ class TestDetectorTemperatures(unittest.TestCase): self.assertEqual(stats["hailo:PCIe#2"]["temperature"], 50.0) self.assertEqual(stats["hailo:PCIe:1"]["temperature"], 60.0) + def test_axelera_temperature_reads_the_board_controller(self): + from frigate.stats.hardware import get_hardware_temperatures + + with patch("frigate.stats.hardware.get_axelera_board_temp", return_value=45.4): + self.assertEqual(get_hardware_temperatures("axelera"), [45.4]) + + with patch("frigate.stats.hardware.get_axelera_board_temp", return_value=None): + self.assertEqual(get_hardware_temperatures("axelera"), [None]) + + +class TestAxeleraBoardTempCooldown(unittest.TestCase): + """A failed board read backs off instead of delaying every stats update.""" + + def test_failed_read_starts_a_cooldown(self): + import subprocess as sp + + import frigate.util.services as services + + with ( + patch.object(services, "_axelera_board_temp_cache", None), + patch.object(services, "_axelera_board_temp_error_until", 0.0), + patch("frigate.util.runtime_deps.find_tool", return_value="/bin/true"), + patch.object( + services.sp, + "run", + side_effect=sp.TimeoutExpired(cmd="axcmd", timeout=5), + ) as run_mock, + ): + self.assertIsNone(services.get_axelera_board_temp()) + # the cooldown prevents the second update from paying the + # blocking read cost again while the board is unresponsive + self.assertIsNone(services.get_axelera_board_temp()) + + self.assertEqual(run_mock.call_count, 1) + if __name__ == "__main__": unittest.main() diff --git a/frigate/util/services.py b/frigate/util/services.py index eda10628f2..a88e44b5c7 100644 --- a/frigate/util/services.py +++ b/frigate/util/services.py @@ -842,6 +842,83 @@ def get_axcl_npu_stats() -> dict[str, str | float] | None: return None +_axelera_board_temp_cache: tuple[float, float] | None = None +_axelera_board_temp_error_until = 0.0 +_AXELERA_BOARD_TEMP_TTL = 5.0 +_AXELERA_BOARD_TEMP_ERROR_COOLDOWN = 60.0 + + +def get_axelera_board_temp() -> float | None: + """Get the Axelera Metis board temperature through the SDK's axcmd tool. + + The read goes through the board controller and does not open an inference + stream, so it is safe to call while a detector is running. ``axcmd`` ships + with the runtime wheel; ``find_tool`` prefers the runtime-installed copy + under the user base. Results are cached for a few seconds so one stats tick + spawns axcmd at most once, and a failed read starts a cooldown so an + unresponsive board does not delay every stats update. + + NOTE: this function is blocking (up to ~5s on a hung board). + """ + global _axelera_board_temp_cache + global _axelera_board_temp_error_until + + now = time.monotonic() + if ( + _axelera_board_temp_cache is not None + and now - _axelera_board_temp_cache[0] < _AXELERA_BOARD_TEMP_TTL + ): + return _axelera_board_temp_cache[1] + + if now < _axelera_board_temp_error_until: + # the last read already cost up to 5s; back off instead of paying + # that cost again on every stats update while the board is down + return None + + try: + from frigate.util.runtime_deps import find_tool + + axcmd_path = find_tool("axcmd") + except Exception: # noqa: BLE001 + axcmd_path = shutil.which("axcmd") + + if axcmd_path is None or not os.path.exists(axcmd_path): + return None + + try: + result = sp.run( + [axcmd_path, "--board-temp"], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + except (OSError, sp.TimeoutExpired): + _axelera_board_temp_error_until = now + _AXELERA_BOARD_TEMP_ERROR_COOLDOWN + return None + + if result.returncode != 0: + _axelera_board_temp_error_until = now + _AXELERA_BOARD_TEMP_ERROR_COOLDOWN + return None + + # anchor on the board-controller line ("N: temp_sensorX@YY = ") so + # unrelated "= " output (firmware/version preambles) is not picked up + match = re.search(r"temp_sensor[^=\n]*=\s*([\d.]+)", result.stdout) + + if match is None: + _axelera_board_temp_error_until = now + _AXELERA_BOARD_TEMP_ERROR_COOLDOWN + return None + + try: + temp = round(float(match.group(1)), 1) + except ValueError: + _axelera_board_temp_error_until = now + _AXELERA_BOARD_TEMP_ERROR_COOLDOWN + return None + + _axelera_board_temp_cache = (now, temp) + return temp + + def try_get_info(f, h, default="N/A", sensor=None): try: if h: