diff --git a/docs/data/object_detectors_models.yaml b/docs/data/object_detectors_models.yaml
index 1c5a189e5e..a21ae2ea7d 100644
--- a/docs/data/object_detectors_models.yaml
+++ b/docs/data/object_detectors_models.yaml
@@ -1269,3 +1269,37 @@ axengine:
input_dtype: int
input_pixel_format: bgr
labelmap_path: /labelmap/coco-80.txt
+axelera:
+ title: Axelera Metis
+ models:
+ - key: yoloxs
+ label: YOLOX-S
+ recommended: true
+ download: |-
+ The **yolox-s-coco-onnx** model is a Voyager SDK model-zoo preset (Megvii YOLOX, Apache-2.0 licensed), so no download or compilation is needed. To use a custom model, compile your ONNX model with the SDK compiler (`axcompile`) or build from the model zoo (`axzoo build yolox_s_coco --backend axrelay --cal-source repr`), then point `model.path` at the compiled model or its URL (e.g. hosted on Hugging Face).
+ ui: |-
+ Navigate to **Settings > System > Detectors and model** and select **Axelera Metis NPU** from the detector type dropdown and click **Add**. Then on the same page, in the **Custom Model** tab, configure:
+
+ | Field | Value |
+ | ---------------------------------------- | ----------------------- |
+ | **Custom object detector model path** | `yolox-s-coco-onnx` |
+ | **Label map for custom object detector** | `/labelmap/coco-80.txt` |
+ | **Object detection model input width** | `640` |
+ | **Object detection model input height** | `640` |
+ | **Model Input Pixel Color Format** | `bgr` |
+ | **Model Input Tensor Shape** | `nhwc` (Frigate's default value) |
+ | **Model Input D Type** | `int` |
+ | **Object Detection Model Type** | `yolo-generic` |
+ yaml: |-
+ detectors:
+ axelera:
+ type: axelera
+
+ model:
+ path: yolox-s-coco-onnx
+ model_type: yolo-generic
+ width: 640
+ height: 640
+ input_dtype: int
+ 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 602ebfcb85..eb7244d9ae 100644
--- a/docs/docs/configuration/object_detectors.md
+++ b/docs/docs/configuration/object_detectors.md
@@ -60,6 +60,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): axm models can run on Axelera AI Metis NPUs.
+
**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.
@@ -784,3 +788,26 @@ The AXEngine detector downloads its default model from HuggingFace on first star
When configuring the AXEngine detector, you have to specify the model name.
+
+## Axelera Metis
+
+ Hardware accelerated object detection is supported on the [Axelera Metis](https://www.axelera.ai/) PCIe / M.2 NPU accelerator from Axelera AI, driven by the vendor's Voyager SDK. The Voyager SDK (Python 3.10 through 3.13) must be installed on the host running Frigate.
+
+See the [installation docs](../frigate/installation.md#axelera-metis) for information on configuring the Axelera hardware.
+
+### Configuration {#configuration-axelera}
+
+The integration runs inference through a single stream created with `create_inference_stream(network=...)`, which accepts a compiled model as a URL (downloaded into the model cache), a local file path, or an SDK model-zoo preset name. When no model is specified, the SDK model-zoo preset `yolox-s-coco-onnx` is used (no download); it can also be selected explicitly with `model.path: yolox-s-coco-onnx`.
+
+Models must be compiled for the Metis architecture — see [Building a model](/frigate/installation#building-a-model) for the full walkthrough.
+
+
+
+### Downloading YOLO Models
+
+#### Axelera Metis
+
+The Metis board runs models compiled for its architecture (`.axm` files, or a compiled model directory). Two options are available:
+
+- **SDK model-zoo preset (no compilation or download needed):** the `yolox-s-coco-onnx` preset ships with the Voyager SDK (YOLOX, Megvii, Apache-2.0 licensed). Select it with `model.path: yolox-s-coco-onnx`.
+- **Compiled model:** see [Building a model](/frigate/installation#building-a-model) for how to compile one, then point `model.path` at it via a local path or an HTTP URL (e.g. on Hugging Face). Set `model.width` and `model.height` to the compiled model's input size.
diff --git a/docs/docs/frigate/installation.md b/docs/docs/frigate/installation.md
index 14c85ec4f3..6aa4191b35 100644
--- a/docs/docs/frigate/installation.md
+++ b/docs/docs/frigate/installation.md
@@ -497,6 +497,108 @@ 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](https://www.axelera.ai/) is a PCIe / M.2 NPU accelerator from Axelera AI. Detections run through the vendor's Voyager SDK (`axelera.app`), so the SDK must be importable in the process that runs the detector.
+
+#### Installation
+
+Install the Voyager SDK on the host as described in the official [SDK installation guide](https://docs.axelera.ai/sdk/user-guides/sdk-install):
+
+```bash
+git clone https://github.com/axelera-ai-hub/voyager-sdk.git
+cd voyager-sdk
+./install-dependencies.sh
+python3 -m venv axelera-env
+source axelera-env/bin/activate
+pip install --extra-index-url https://software.axelera.ai/artifactory/api/pypi/axelera-pypi/simple axelera-rt==1.7.0 'axelera-devkit[all]==1.7.0'
+make operators
+```
+
+Then install the Metis kernel driver, for example with `axdevice driver --install` (or via the Axelera apt repository). See the SDK guide for the current driver version and options.
+
+#### Building a model
+
+Detector models are compiled for the Metis architecture. The simplest path needs no compilation: the detector uses the `yolox-s-coco-onnx` model-zoo preset by default, which ships with the SDK.
+
+To compile your own ONNX model, use the SDK compiler (`axcompile`, see the [Compiler CLI reference](https://docs.axelera.ai/sdk/reference/compiler/compiler-cli)):
+
+```bash
+axcompile -i /path/to/model.onnx -o /path/to/compiled/
+```
+
+The compiler produces the artifacts under `output/compiled_model/` (see the SDK "Model Formats" documentation for what each file is). Give the compiled model to the detector through `model.path` (URL or local path) and set `model.width`/`model.height` to the model input size. Models from the SDK model zoo can also be built with the SDK CLI, e.g. `axzoo build yolox_s_coco --backend axrelay --cal-source repr`.
+
+To use a custom model in Frigate, configure [hardware object detection](/configuration/object_detectors#axelera-metis) and point `model.path` at your compiled model or its URL.
+
+#### Setup
+
+Frigate must be able to reach the Metis device. After installing the driver, verify the device is present with `ls /dev/metis*`. Grant the user running Frigate read/write access on the Metis device nodes; the installer creates an `axelera` group for this purpose.
+
+The Voyager SDK is a Python package (`axelera.app`), so the plain Frigate Docker image cannot load it: the detector imports the SDK inside Frigate's own process. Build a custom image with the SDK installed, then pass the Metis device into the container.
+
+#### Building a custom image
+
+The following reference `Dockerfile` follows the SDK's own installation flow (`install-dependencies.sh`, `pip install axelera-rt`/`axelera-devkit`, `make operators`) and installs the SDK into the image's Python environment so the `axelera` plugin can import it in-process:
+
+```dockerfile
+FROM ghcr.io/blakeblackshear/frigate:stable
+
+# Build dependencies for the SDK operators (same set as the SDK install-dependencies.sh).
+RUN apt-get update && apt-get install --no-install-recommends -y \
+ clinfo g++ gcc glslang-tools graphviz gstreamer1.0-libav \
+ gstreamer1.0-plugins-base gstreamer1.0-plugins-good gstreamer1.0-rtsp \
+ gstreamer1.0-tools libavcodec-dev libavformat-dev libavutil-dev \
+ libcairo2-dev libeigen3-dev libgirepository1.0-dev \
+ libgstreamer-plugins-base1.0-dev libgstreamer1.0-dev libavdevice-dev \
+ libopencv-dev libqt6svg6-dev libsimde-dev libswscale-dev libva-dev \
+ libvulkan1 ninja-build nlohmann-json3-dev ocl-icd-opencl-dev \
+ opencl-headers pkg-config python3-dev unzip wget \
+ && rm -rf /var/lib/apt/lists/*
+
+# Voyager SDK installed into the image Python (no venv: Frigate runs as `python3`,
+# so the SDK must be importable from the system site-packages).
+RUN python3 -m pip install --no-cache-dir \
+ --extra-index-url https://software.axelera.ai/artifactory/api/pypi/axelera-pypi/simple \
+ 'axelera-rt==1.7.0' 'axelera-devkit[all]==1.7.0'
+
+# Compile the SDK operators for this host architecture (run with the SDK runtime
+# activated, as the SDK Makefile requires).
+RUN git clone --depth 1 https://github.com/axelera-ai-hub/voyager-sdk.git /opt/voyager-sdk \
+ && cd /opt/voyager-sdk && make operators
+```
+
+:::note
+
+The operator build is architecture-specific. Build and validate the image on the same host architecture (and with the same SDK version) as the Metis card, and match the `axelera-rt`/`axelera-devkit` versions to the installed kernel driver.
+
+:::
+
+Then pass the Metis device and the GPU/dma-buf nodes into the container. Add the following lines to your `docker-compose.yml`:
+
+```yaml
+devices:
+ - /dev/metis-0-1-0
+ - /dev/dma_heap/system:/dev/dma_heap/system
+ - /dev/dri/renderD128:/dev/dri/renderD128 # Intel/AMD GPU render node (OpenCL)
+```
+
+The SDK needs the dma-buf exporter for its shared-memory pool (`No dmabuf driver found` at startup means it is missing). The GPU render node is required for the SDK to detect OpenCL and use the GPU hardware path (`hardware_caps.opencl`); without it the SDK falls back to software decoding. Which nodes exist depends on the host (Intel iGPU render nodes are numbered, e.g. `renderD128`; AMD uses `renderD129`).
+
+Metis device node names depend on the driver version (for example `/dev/metis-0-1-0` or `/dev/metis-0:1:0`); list the nodes under `/dev/metis*` after installing the driver and pass the one that appears for your card.
+
+If you are using `docker run`, add these options to your command:
+
+```bash
+--device /dev/metis-0-1-0 \
+--device /dev/dma_heap/system:/dev/dma_heap/system \
+--device /dev/dri/renderD128:/dev/dri/renderD128
+```
+
+#### 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/frigate/detectors/plugins/axelera.py b/frigate/detectors/plugins/axelera.py
new file mode 100644
index 0000000000..e1e6a0198d
--- /dev/null
+++ b/frigate/detectors/plugins/axelera.py
@@ -0,0 +1,754 @@
+"""Frigate object detector plugin for the Axelera Metis NPU (`type: 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 the hardware vendors use 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 our ~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:
+ - ``model.path`` = path to a COMPILED model directory (containing
+ ``1/model.json`` + ``1/postprocess_graph.onnx``) or to the ``model.json``
+ itself; a preset name (e.g. ``yolonas-m-coco-onnx``) is looked up under
+ ``AXELERA_BUILD_ROOT`` (default ``/opt/voyager-sdk/build``), and
+ http(s)/scheme-less host URLs download into the model cache.
+ - ``model.labelmap_path`` / ``model.labelmap``: labels are mapped to class ids
+ BY NAME via the merged labelmap, independent of label ordering.
+ - ``model.width`` / ``model.height``: input frame dimensions (the coordinate
+ space of the region Frigate sends, e.g. 704x576); the plugin letterboxes to
+ the model's square input (640x640 for the YOLO-NAS M preset above) and maps
+ boxes back.
+
+The SDK's own baked postprocess graph (``1/postprocess_graph.onnx``) converts
+the six raw AIPU tensors into (8400, 4) boxes + (8400, 80) class scores, so
+the plugin only dequantises the int8 heads and runs a thin class-max + NMS.
+"""
+
+import json
+import logging
+import os
+import queue
+import re
+import threading
+from typing import Literal
+
+import cv2
+import numpy as np
+from pydantic import ConfigDict
+
+from frigate.const import MODEL_CACHE_DIR
+from frigate.detectors.detection_api import DetectionApi
+from frigate.detectors.detector_config import BaseDetectorConfig
+from frigate.util.downloader import ModelDownloader
+
+logger = logging.getLogger(__name__)
+
+DETECTOR_KEY = "axelera"
+
+AX_DEFAULT_NETWORK = "yolox-s-coco-onnx"
+AX_DEFAULT_BUILD_ROOT = "/opt/voyager-sdk/build"
+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 ----------------- #
+
+
+# Matches a scheme-less host URL: host([:port])/path, e.g. www.example.com/model.axm
+# or localhost:8000/model.axm (bare filenames like foo.axm and relative paths like
+# models/foo.axm have no leading dotted host and stay local paths / presets).
+_HOST_URL_RE = re.compile(
+ r"^(localhost|[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)*\.[a-zA-Z]{2,})(:[0-9]+)?/"
+)
+
+
+def is_url(value: str) -> bool:
+ """Return True for http(s) URLs and scheme-less host URLs.
+
+ A bare host such as ``www.example.com/model.axm`` (no scheme) is a usable
+ URL too: rejecting it would raise FileNotFoundError on a valid remote
+ model. ``_url_with_scheme`` prepends ``http://`` before any fetch.
+ """
+ return value.startswith(("http://", "https://")) or bool(_HOST_URL_RE.match(value))
+
+
+def _url_with_scheme(value: str) -> str:
+ """Prepend http:// to a scheme-less host URL (vetted by is_url)."""
+ if value.startswith(("http://", "https://")):
+ return value
+ return f"http://{value}"
+
+
+def _looks_like_path(value: str) -> bool:
+ """Heuristic: a value that should refer to a file on disk, not a preset."""
+ return "/" in value or "." in value
+
+
+def _resolve_model_json(model_path, model_url, build_root) -> str:
+ """Resolve ``model.path`` to the compiled ``model.json`` for axelera.runtime.
+
+ Returns the path to ``1/model.json`` (or ``model.json``) for a directory,
+ the model.json itself when given, or a compiled-model dir found under
+ ``build_root`` for a bare preset name. Raises FileNotFoundError otherwise.
+ """
+ if model_url and model_path:
+ raise ValueError("`model.path` cannot be both a URL and a local path.")
+
+ if model_url:
+ # a URL points at a compiled archive; download to the cache and treat
+ # it as a directory the caller must have extracted into (documented as
+ # out of scope for the common flow: most users point at an on-disk
+ # compiled model or an SDK preset under the build root).
+ name = os.path.basename(model_url) or AX_DEFAULT_NETWORK
+ os.makedirs(MODEL_CACHE_DIR, exist_ok=True)
+ destination = os.path.join(MODEL_CACHE_DIR, name)
+ if not os.path.isfile(destination):
+ logger.info("axelera: downloading model from %s", model_url)
+ try:
+ ModelDownloader.download_from_url(
+ _url_with_scheme(model_url), destination
+ )
+ except Exception as exc:
+ raise RuntimeError(
+ f"Failed to download model from {model_url}: {exc}"
+ ) from exc
+ return _find_model_json(destination)
+
+ if model_path and _looks_like_path(model_path):
+ return _find_model_json(model_path)
+
+ preset = model_path or AX_DEFAULT_NETWORK
+ root = build_root or AX_DEFAULT_BUILD_ROOT
+ for candidate in (
+ os.path.join(root, preset, preset, "1", "model.json"),
+ os.path.join(root, preset, "1", "model.json"),
+ os.path.join(root, preset, "model.json"),
+ os.path.join(root, preset, preset, "model.json"),
+ ):
+ if os.path.isfile(candidate):
+ return candidate
+ raise FileNotFoundError(
+ f"Compiled model not found for {model_path!r} under {root}. "
+ "Point model.path at the compiled model directory (e.g. a dir "
+ "containing 1/model.json) or a preset that was built into the "
+ "AXELERA_BUILD_ROOT."
+ )
+
+
+def _find_model_json(where: str) -> str:
+ """Locate the runtime's model.json under a dir or return the path itself."""
+ if os.path.isfile(where) and os.path.basename(where) == "model.json":
+ return where
+ for candidate in (
+ os.path.join(where, "1", "model.json"),
+ os.path.join(where, "model.json"),
+ ):
+ if os.path.isfile(candidate):
+ return candidate
+ raise FileNotFoundError(
+ f"model.json not found for {where}. Point model.path at the compiled "
+ "model directory (containing 1/model.json) or the model.json itself."
+ )
+
+
+def _nms_xyxy(boxes, scores, iou_threshold: float) -> np.ndarray:
+ """Greedy NMS on xyxy boxes via cv2.dnn.NMSBoxes (C++, GIL-free).
+
+ Returns kept indices in descending score order, same contract as the
+ previous pure-Python loop but without the O(n^2) worst case on dense
+ frames (daytime, multiple motion regions at 48 fps offered).
+ """
+ if len(boxes) == 0:
+ return np.array([], dtype=np.int64)
+ x1 = boxes[:, 0]
+ y1 = boxes[:, 1]
+ xywh = np.stack(
+ [
+ x1,
+ y1,
+ np.maximum(0, boxes[:, 2] - x1),
+ np.maximum(0, boxes[:, 3] - y1),
+ ],
+ axis=1,
+ )
+ indices = cv2.dnn.NMSBoxes(
+ xywh.tolist(),
+ [float(s) for s in scores],
+ # inputs are already filtered above CONF_THRESHOLD by the caller
+ score_threshold=0.0,
+ nms_threshold=float(iou_threshold),
+ )
+ if len(indices) == 0:
+ return np.array([], dtype=np.int64)
+ if isinstance(indices, np.ndarray):
+ return indices.reshape(-1).astype(np.int64)
+ # OpenCV versions that return a (N, 1) list of nested lists
+ return np.asarray(indices, dtype=np.int64).reshape(-1)
+
+
+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). ``1/manifest.json`` is
+ the cross-model source of truth (verified for YOLO-NAS M and YOLOX-S);
+ fall back to ``/quantized/manifest.json`` for older compilations.
+ """
+ candidates = [
+ os.path.join(model_dir, "manifest.json"),
+ os.path.join(os.path.dirname(model_dir), "quantized", "manifest.json"),
+ ]
+ for candidate in candidates:
+ if not os.path.isfile(candidate):
+ continue
+ 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
+ raise FileNotFoundError(
+ "compiled manifest not found next to the model kernel. Point "
+ "model.path at a directory compiled by the Voyager SDK that contains "
+ "1/model.json + 1/manifest.json + 1/postprocess_graph.onnx."
+ )
+
+
+# ----------------- 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 letterboxes +
+ quantizes + runs the model (``ModelInstance.run``, synchronous and flat:
+ ~21 ms measured), and a consumer feeds the six 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 (contrast:
+ the create_inference_stream variant).
+ """
+
+ def __init__(
+ self,
+ model_json,
+ labels: list[str],
+ labelmap: dict,
+ width: int,
+ height: int,
+ build_root: str | None = None,
+ runtime_factory=None,
+ onnx_factory=None,
+ ) -> None:
+ self._labels = labels
+ self._labelmap = labelmap
+ self._width, self._height = int(width), int(height)
+ 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. Install the Axelera runtime package and grant "
+ "the Frigate process access to the Metis device."
+ ) 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)
+ conn = self._ctx.device_connect(None, num_sub_devices=1)
+ self._instance = 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 letterbox 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}"
+ )
+ self._pp = onnx_factory.InferenceSession(
+ pp_path, 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 dequantised 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 letterbox-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)
+ # letterbox geometry from the frame coords (width x height) into the
+ # model's own workspace (640x640 for the YOLO presets)
+ self._scale = min(self._model_w / self._width, self._model_h / self._height)
+ new_w = round(self._width * self._scale)
+ new_h = round(self._height * self._scale)
+ self._new_w, self._new_h = max(1, new_w), max(1, new_h)
+ self._pad_t = (self._model_h - self._new_h) // 2
+ self._pad_l = (self._model_w - self._new_w) // 2
+
+ # -- 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:
+ self._instance.free()
+ except Exception: # noqa: BLE001, S110
+ pass
+ logger.info("axelera: runtime inference stopped")
+
+ # -- internals ---------------------------------------------------------
+
+ def _push_result(self, connection_id: str, rows: np.ndarray) -> None:
+ try:
+ self._result_q.put_nowait((connection_id, rows))
+ except queue.Full:
+ logger.error("axelera: result queue full; dropping %s", connection_id)
+
+ def _producer(self) -> None:
+ """Letterbox + 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 overlap 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:
+ """Letterbox the source (H, W, 3) into the padded input buffer."""
+ resized = cv2.resize(
+ frame,
+ (self._new_w, self._new_h),
+ interpolation=cv2.INTER_LINEAR,
+ )
+ # uint8 - 128 wraps mod 256, then cast to int8 == px - 128 (exactly),
+ # in one pass over the resized frame (no int16/clip intermediates)
+ q = (resized - np.uint8(128)).astype(np.int8)
+ buf.fill(INPUT_Q_ZERO)
+ y0, y1 = self._pad_t, self._pad_t + self._new_h
+ x0, x1 = self._pad_l, self._pad_l + self._new_w
+ ct, cl = self._content_top, self._content_left
+ # channels 0..2 = quantized RGB (BGR -> RGB flip as in the SDK examples;
+ # the compiled input expects RGB semantics), channel 3 = alpha-255
+ buf[0, ct + y0 : ct + y1, cl + x0 : cl + x1, 0:3] = q[..., ::-1]
+ buf[0, ct + y0 : ct + y1, cl + x0 : cl + x1, 3] = 127
+
+ def _decode(self, out_bufs) -> np.ndarray:
+ """Dequantise heads + postprocess graph + conf threshold + NMS -> rows."""
+ # dequantize + crop to the real channel count + transpose NHWC -> NCHW
+ 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]
+ arr = obuf[..., :real].astype(np.float32)
+ arr = (arr - zp) * scale
+ feeds[name] = np.ascontiguousarray(arr.transpose(0, 3, 1, 2))
+
+ 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]
+ conf = scores.max(axis=1)
+ cls_idx = scores.argmax(axis=1).astype(np.int32)
+ 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)
+ total = raw.shape[-1]
+ cls = raw[:, 4:]
+ if total == 4 + 1 + len(self._labels):
+ cls = raw[:, 5:] * raw[:, 4:5] # YOLOX: score = obj * cls
+ if cls.shape[1] == 0:
+ return np.zeros((MAX_ROWS, 6), np.float32)
+ conf = cls.max(axis=1)
+ cls_idx = cls.argmax(axis=1).astype(np.int32)
+ 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)
+ # inverse letterbox into the source (width x height) frame space
+ bx = (boxes[:, [0, 2]] - self._pad_l) / self._scale
+ by = (boxes[:, [1, 3]] - self._pad_t) / self._scale
+ x0i = np.clip(bx[:, 0], 0, self._width)
+ x1i = np.clip(bx[:, 1], 0, self._width)
+ y0i = np.clip(by[:, 0], 0, self._height)
+ y1i = np.clip(by[:, 1], 0, self._height)
+ # NMS on the 640-space boxes (scale-invariant IoU)
+ keep_nms = _nms_xyxy(boxes, conf, NMS_IOU_THRESHOLD)
+ rows = []
+ for i in keep_nms:
+ label = (
+ self._labels[cls_idx[i]]
+ if 0 <= cls_idx[i] < len(self._labels)
+ else None
+ )
+ class_id = self._labelmap.get(label)
+ if class_id is None:
+ continue
+ score = float(conf[i])
+ rows.append(
+ (
+ class_id,
+ 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
+
+ def __init__(self, detector_config: "AxeleraDetectorConfig"):
+ model = detector_config.model
+ if model is None:
+ raise ValueError(
+ f"type={DETECTOR_KEY}: the `model:` block is required "
+ "(path to a compiled model directory/JSON, URL, or SDK "
+ "model-zoo preset name)."
+ )
+ super().__init__(detector_config)
+
+ self.width = model.width
+ self.height = model.height
+ model_path = model.path
+ self.model_url = model_path if (model_path and is_url(model_path)) else None
+ self.model_path = None if self.model_url else model_path
+
+ build_root = os.environ.get("AXELERA_BUILD_ROOT", AX_DEFAULT_BUILD_ROOT)
+ self.working_model = _resolve_model_json(
+ self.model_path, self.model_url, build_root
+ )
+
+ # label name -> Frigate class id by NAME (independent of ordering)
+ self._labelmap = {}
+ for class_id, label in (model.merged_labelmap or {}).items():
+ if label is not None:
+ self._labelmap[str(label).strip()] = class_id
+ if not self._labelmap:
+ logger.warning("axelera: empty labelmap: detections will not be mapped")
+
+ # COCO-ordered label names from the compiled model metadata (index -> name)
+ labels = _read_model_labels(os.path.dirname(self.working_model))
+
+ self._inference = _AxeleraRuntimeInference(
+ model_json=self.working_model,
+ labels=labels,
+ labelmap=self._labelmap,
+ width=self.width,
+ height=self.height,
+ build_root=build_root,
+ )
+ 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 stream."""
+ 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 (letterboxing/quantization happen in the worker)
+ frame = np.ascontiguousarray(frame, dtype=np.uint8)
+ 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 removed: axelera uses the async detector contract."""
+ return 0 # type: ignore[override]
+
+ def close(self):
+ """Stop the inference runtime and release the Metis device."""
+ self._inference.stop()
+
+ def __del__(self):
+ # destructor must not raise
+ try:
+ self.close()
+ except Exception: # noqa: S110, BLE001
+ pass
+
+
+def _read_model_labels(model_dir: str) -> list[str]:
+ """Read the COCO-ordered label list from the compiled model metadata."""
+ # model_info.json lives next to the model's parent (1/ holds the kernel)
+ candidates = [model_dir, os.path.dirname(model_dir)]
+ for base in candidates:
+ for name in ("model_info.json", "model.json"):
+ path = os.path.join(base, name)
+ if not os.path.isfile(path):
+ continue
+ try:
+ with open(path) as f:
+ data = json.load(f)
+ for key in ("labels", "class_names"):
+ if isinstance(data.get(key), list) and data[key]:
+ return [str(x).strip() for x in data[key]]
+ except Exception as exc: # noqa: BLE001
+ logger.debug("axelera: could not read labels from %s: %s", path, exc)
+ # fall back to the merged labelmap's labels in id order (unlikely path)
+ return []
+
+
+# ----------------- AxeleraDetectorConfig Class ----------------- #
+
+
+class AxeleraDetectorConfig(BaseDetectorConfig):
+ model_config = ConfigDict(title="Axelera Metis NPU")
+
+ type: Literal[DETECTOR_KEY]
diff --git a/frigate/object_detection/base.py b/frigate/object_detection/base.py
index bc7910e4d0..eca9b72d73 100644
--- a/frigate/object_detection/base.py
+++ b/frigate/object_detection/base.py
@@ -357,8 +357,8 @@ class ObjectDetectProcess:
if (self.detect_process is not None) and self.detect_process.is_alive():
self.stop()
- # Async path for MemryX
- if self.detector_config.type == "memryx":
+ # Async path for MemryX and Axelera (stream-based send/receive threads)
+ 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/util.py b/frigate/stats/util.py
index 6e20197391..f89118557a 100644
--- a/frigate/stats/util.py
+++ b/frigate/stats/util.py
@@ -21,6 +21,7 @@ from frigate.util.services import (
calculate_shm_requirements,
get_amd_gpu_stats,
get_axcl_npu_stats,
+ get_axelera_board_temp,
get_bandwidth_stats,
get_cpu_stats,
get_fs_type,
@@ -132,6 +133,10 @@ def get_detector_temperature(
# Rockchip temperatures are handled by the GPU / NPU stats
# as there are not detector specific temperatures
pass
+ elif detector_type == "axelera":
+ # Axelera Metis temperature is read from the board controller
+ # through the SDK's axcmd tool
+ return get_axelera_board_temp()
return None
@@ -335,6 +340,16 @@ async def set_npu_usages(config: FrigateConfig, all_stats: dict[str, Any]) -> No
# AXERA NPU usage
axcl_usage = get_axcl_npu_stats()
stats["axengine"] = axcl_usage
+ elif detector.type == "axelera":
+ # Axelera Metis: temperature from the board controller.
+ # There is no utilization readout without the axsystemserver
+ # broker, which cannot run alongside a live detector.
+ # axcmd spawns can take seconds on a busy board; this stats task
+ # runs in the async event loop, so run the blocking read in a thread.
+ metis_temp = await asyncio.to_thread(get_axelera_board_temp)
+
+ if metis_temp is not None:
+ stats["axelera"] = {"mem": "-%", "temp": metis_temp}
if stats:
all_stats["npu_usages"] = stats
diff --git a/frigate/util/services.py b/frigate/util/services.py
index 38737907ef..20c610dcf9 100644
--- a/frigate/util/services.py
+++ b/frigate/util/services.py
@@ -800,6 +800,71 @@ def get_axcl_npu_stats() -> dict[str, str | float] | None:
return None
+_board_temp_cache: tuple[float, float] | None = None
+_BOARD_TEMP_TTL = 5.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. Results are
+ cached for a few seconds so the stats cycle (detector temperature + NPU
+ usage run in the same tick) spawns axcmd at most once.
+
+ NOTE: this function is blocking (up to ~5s on a hung board). Callers in
+ async contexts must wrap it in ``asyncio.to_thread``.
+ """
+ global _board_temp_cache
+
+ now = time.monotonic()
+ if _board_temp_cache is not None and now - _board_temp_cache[0] < _BOARD_TEMP_TTL:
+ return _board_temp_cache[1]
+
+ axcmd_path = None
+
+ for candidate in (
+ shutil.which("axcmd"),
+ "/usr/local/bin/axcmd",
+ "/opt/axelera/runtime-1.8.0-1/bin/axcmd",
+ ):
+ if candidate and os.path.exists(candidate):
+ axcmd_path = candidate
+ break
+
+ if axcmd_path is None:
+ return None
+
+ try:
+ result = sp.run(
+ [axcmd_path, "--board-temp"],
+ capture_output=True,
+ text=True,
+ timeout=5,
+ check=False,
+ )
+ except (OSError, sp.TimeoutExpired):
+ return None
+
+ if result.returncode != 0:
+ 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:
+ return None
+
+ try:
+ temp = round(float(match.group(1)), 1)
+ except ValueError:
+ return None
+
+ _board_temp_cache = (now, temp)
+ return temp
+
+
def try_get_info(f, h, default="N/A", sensor=None):
try:
if h: