From 92112cf909c044860e9f9555a670e3bb64c4eaa5 Mon Sep 17 00:00:00 2001 From: patrykorwat Date: Sun, 23 Aug 2026 20:09:21 +0000 Subject: [PATCH] feat(axelera): run inference via the axelera.runtime API instead of the SDK stream The GStreamer-based create_inference_stream pipeline is depth-4 and clocked by input arrival, so a frame's result round-trips in ~4x the input gap (~150 ms at ~35 fps) even though the accelerator computes a frame in ~21 ms back-to-back. Switch to the vendor's direct ModelInstance.run() runtime API (the same integration pattern as hailo8l/memryx): a synchronous run() with no video pipeline, so per-frame latency is the flat model time (~21 ms measured for YOLO-NAS M 704x576 at 1 instance), independent of scene activity. - _AxeleraRuntimeInference: single ModelInstance plus a worker thread that pops frames strictly FIFO, letterboxes + quantizes (int8, BGR->RGB), runs the model, feeds the six int8 heads through the compiled model's own baked postprocess_graph.onnx (dequant, crop per TensorInfo.padding[3], transpose) and emits (class_id, score, box) rows after a conf threshold + NMS. - Keeps the async contract (send_input/receive_output/shutdown); requests are paired to results by construction through the FIFO queue, so no pixel tokens and no keep-alives. - Model resolution unchanged (preset under AXELERA_BUILD_ROOT, local path, or URL); labels mapped by name via the merged labelmap. --- frigate/detectors/plugins/axelera.py | 638 +++++++++++++++------------ 1 file changed, 360 insertions(+), 278 deletions(-) diff --git a/frigate/detectors/plugins/axelera.py b/frigate/detectors/plugins/axelera.py index 4c564f799b..8f9bb57d05 100644 --- a/frigate/detectors/plugins/axelera.py +++ b/frigate/detectors/plugins/axelera.py @@ -1,54 +1,46 @@ """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 Voyager SDK (`axelera.app`), which is installed on -the host running Frigate. This plugin follows the same conventions as the -other vendor NPU plugins (`hailo8l.py`, `memryx.py`): a compiled model can be -given as a URL (downloaded into the model cache), a local path, or omitted to -use the SDK's bundled model-zoo preset. +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 a stream instead of per-frame `session.run()` / `device_connect()`: the -Voyager SDK exposes `create_inference_stream(network=..., sources=[...])`, -which accepts generators of numpy frames (BGR) and yields per-frame results. -This is the only code path that is reliable across Metis hosts (raw -device-level connections can wedge the board on some hosts), and it amortises -model compilation across frames. +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. -The plugin uses Frigate's ASYNC detector contract (`AsyncDetectorRunner`, -same as the memryx plugin): `send_input(connection_id, tensor)` is called -from the detect worker thread as frames arrive, and `receive_output()` from -the result worker thread returns `(connection_id, rows)` per completed frame. -Because the runner keeps submitting real frames as cameras feed them, the SDK -pipeline is continuously clocked by real frames and no keep-alives are -needed (measured: back-to-back real frames round-trip in ~14 ms, while -interleaving 1 s keep-alives stretched a real frame's result to ~4 s). - -Results are paired to requests by RECORDED ORDER: the feeder registers every -frame it yields on a pending deque (``None`` for primer/keep-alive synthetic -frames, the connection id for real frames) and ``collect`` pops exactly one -entry per completed result. The SDK stream returns results strictly in input -order, one per yielded frame (verified on hardware: 200/200 frames in order, -no drops or duplicates, including with synthetics interleaved), so a real -frame always gets its own result, synthetics pop a ``None`` and are dropped, -and no frame id is ever assumed from an unrecorded ordering. - -Model configuration follows the hardware-backend convention: - - `model.path` = URL | scheme-less host URL (``www.example.com/...``) | local - file path | SDK model-zoo preset name (e.g. ``yolox-s-coco-onnx``). When - omitted, the SDK model-zoo preset ``AX_DEFAULT_NETWORK`` (YOLOX, Apache-2.0 - licensed) is used. - - `model.labelmap_path` / `model.labelmap`: labels are mapped to class ids +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 for the network. + - ``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 -from collections import deque +import threading from typing import Literal +import cv2 import numpy as np from pydantic import ConfigDict @@ -61,31 +53,33 @@ logger = logging.getLogger(__name__) DETECTOR_KEY = "axelera" -# Default SDK model-zoo preset used when no `model.path` is given. -# YOLOX (Megvii) is Apache-2.0 licensed; Ultralytics YOLO models are AGPL-3.0 -# and cannot be referenced by Frigate (MIT), see blakeblackshear/frigate#10717. AX_DEFAULT_NETWORK = "yolox-s-coco-onnx" +AX_DEFAULT_BUILD_ROOT = "/opt/voyager-sdk/build" MAX_ROWS = 20 -PRIMER_FRAMES = 3 -# Safety-net keep-alive spacing while the incoming queue is momentarily empty -# (stream init, brief scheduling gaps, true global idle). AsyncDetectorRunner -# keeps the queue full with real frames in steady state, so these zeros rarely -# fire and real frames still clock the pipeline back-to-back (measured ~14 ms). -# A short spacing matters when they DO fire: the SDK pipeline is clocked by the -# input arrival interval (depth 4 x gap), so a 1 s gap would stretch the next -# real frame's result to ~4 s; 0.1 s keeps that penalty in the tens/hundreds -# of ms. These zeros carry no request id and their results are dropped by collect(). -KEEPALIVE_INTERVAL = 0.1 - -# Request pairing is done by RECORDED ORDER, not by content: the SDK's -# InferenceStream returns each source frame back in a result and yields results -# strictly in input order, one per yielded frame (verified on hardware: 200/200 -# frames in order, no drops or duplicates, synthetics interleaved included). -# The feeder therefore records connection ids (or None for primer/keep-alive -# synthetic frames) on a pending deque at each yield, and collect() pops exactly -# one entry per result: a synthetic entry is dropped, a real one delivers the -# result to its connection id. This mirrors how Frigate's own AsyncDetectorRunner -# pairs its send/recv timestamps, and how the memryx plugin pairs its queues. +# YOLO-NAS M compiled model input (padded) from the manifest: +# input_shapes=[1, 642, 656, 4], input_shapes_original=[1, 3, 640, 640] +# Padding: H (1,1), W (1,15), channels (0,1); quantize scale ~= 1/255, zp -128. +MODEL_INPUT_H, MODEL_INPUT_W = 640, 640 +PADDED_H, PADDED_W = 642, 656 # includes top/left 1px border introduced by compile +INPUT_Q_SCALE = 0.003921568859368563 +INPUT_Q_ZERO = -128 +# dequantize_params from the compiled manifest, one per head output, in the +# AIPU output order: (cls, reg, cls, reg, cls, reg) per scale (80, 40, 20). +# AIPU tensors are padded NHWC channels (128); the real channel counts come from +# output_shapes_original: [80, 68, 80, 68, 80, 68]. +HEAD_REAL_CHANNELS = (80, 68, 80, 68, 80, 68) +HEAD_DEQUANT = ( + (0.07594172656536102, 84), + (0.08785015344619751, -63), + (0.06776463985443115, 66), + (0.07863091677427292, -67), + (0.06013955920934677, 44), + (0.06746677309274673, -61), +) +# confidence / NMS defaults mirror the SDK DecodeYolo operator defaults +CONF_THRESHOLD = 0.25 +NMS_IOU_THRESHOLD = 0.45 +QUEUE_MAXSIZE = 32 # ----------------- Utility Functions ----------------- # @@ -123,254 +117,327 @@ def _looks_like_path(value: str) -> bool: return "/" in value or "." in value -def _resolve_network(model_path, model_url) -> str: - """Resolve the value passed to ``create_inference_stream(network=...)``. +def _resolve_model_json(model_path, model_url, build_root) -> str: + """Resolve ``model.path`` to the compiled ``model.json`` for axelera.runtime. - Returns a download URL cached as a local file / an existing local file / - an SDK model-zoo preset name, mirroring the resolution hailo8l performs - for ``.hef``. + 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: - download_url = _url_with_scheme(model_url) - ModelDownloader.download_from_url(download_url, destination) + 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 destination + return _find_model_json(destination) - if model_path: - if _looks_like_path(model_path): - if not os.path.isfile(model_path): - raise FileNotFoundError( - f"Model file not found: {model_path}. If you meant to use " - "an SDK model-zoo preset name, drop the file extension / path." - ) - return model_path - # Bare name: SDK model-zoo preset. - return model_path + if model_path and _looks_like_path(model_path): + return _find_model_json(model_path) - return AX_DEFAULT_NETWORK + 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 _pad_rows(rows) -> np.ndarray: - """Pad detection rows to a fixed (20, 6) float32 array.""" - 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 +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." + ) -# ----------------- AxeleraInference (engine) ----------------- # +def _nms_xyxy(boxes, scores, iou_threshold: float) -> list: + """Classic greedy NMS on xyxy boxes. Returns kept indices (desc score).""" + if len(boxes) == 0: + return [] + x1, y1, x2, y2 = boxes[:, 0], boxes[:, 1], boxes[:, 2], boxes[:, 3] + areas = np.maximum(0, x2 - x1) * np.maximum(0, y2 - y1) + order = scores.argsort()[::-1] + keep = [] + while order.size > 0: + i = order[0] + keep.append(int(i)) + if order.size == 1: + break + xx1 = np.maximum(x1[i], x1[order[1:]]) + yy1 = np.maximum(y1[i], y1[order[1:]]) + xx2 = np.minimum(x2[i], x2[order[1:]]) + yy2 = np.minimum(y2[i], y2[order[1:]]) + inter = np.maximum(0, xx2 - xx1) * np.maximum(0, yy2 - yy1) + union = areas[i] + areas[order[1:]] - inter + ovr = np.where(union > 0, inter / np.maximum(union, 1e-9), 0) + order = order[1:][ovr <= iou_threshold] + return keep -class _AxeleraInference: - """Feeds frames through a single Voyager SDK inference stream. +# ----------------- AxeleraRuntimeInference (engine) ----------------- # - The stream is built from a frame generator (``sources``), so frames can be - pushed by ``submit`` while ``collect`` (called from the result thread) - pulls one completed frame at a time. A short primer of zero frames - kick-starts model pipeline init; they are registered as synthetic (None) - and their results are dropped by ``collect``. - In steady state real frames clock the SDK pipeline back-to-back: the - caller (Frigate's AsyncDetectorRunner) submits frames continuously as - cameras feed them, so the pipeline is never idle (verified: back-to-back - frames round-trip in ~14 ms, while 1 s input gaps stretch a real frame's - latency to ~4 s on this depth-4 pipeline). A sparse 0.1 s keep-alive zero - is emitted only while the incoming queue is momentarily empty (stream - init, scheduling gaps, true idle) so the SDK source never starves; these - zeros are registered as synthetic (None) and their results are dropped. +class _AxeleraRuntimeInference: + """Runs a compiled model via ``axelera.runtime`` in a dedicated worker thread. - Results are STRICTLY paired by RECORDED ORDER: ``_feed`` registers each - yielded frame on ``self._pending`` (``None`` for primer/keep-alive zeros, - the connection id for real frames) and ``collect`` pops exactly one entry - per result. The SDK yields results in input order, one per frame (verified - on hardware: 200/200 in order, no drops or duplicates, synthetics - interleaved included), so a synthetic result pops a ``None`` and is - dropped, and a real result always reaches the connection id that owns it; - no content read of the returned image is involved. + Frigate's AsyncDetectorRunner calls ``submit(connection_id, frame)`` from + the detect thread and ``collect()`` from the result thread. A single worker + thread pops frames strictly in order, letterboxes + quantizes, runs the + model (``ModelInstance.run``, synchronous and flat: ~21 ms measured), feeds + the six int8 heads through the SDK's baked ``postprocess_graph.onnx`` and + pushes ``(connection_id, rows)``. Because one thread processes frames 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, network, labelmap: dict, width: int, height: int, stream_factory=None + 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=64) - # request ids (or None) for frames in flight, in the exact order they - # are yielded to the SDK (appended by _feed on the feeder thread, - # popped once per result by collect on the result thread) - self._pending: deque[str | None] = deque() + 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._thread: threading.Thread | None = None - if stream_factory is None: - # lazy import: never import the SDK unless the detector is used + if runtime_factory is None: try: - from axelera.app import create_inference_stream + from axelera.runtime import Context except (ImportError, ModuleNotFoundError) as exc: raise ImportError( - "Axelera Voyager SDK (axelera.app) is not available in this " - "environment. Install the Voyager SDK (Python 3.10 through " - "3.13) and grant the Frigate process access to the Metis device." + "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 - stream_factory = create_inference_stream + runtime_factory = Context + if onnx_factory is None: + import onnxruntime as _ort - self._stream = stream_factory(network=network, sources=[self._feed()]) - # the SDK stream object is iterable but not an iterator itself - # (`next(stream)` is a TypeError); grab one iterator for collect(). - self._results_iter = iter(self._stream) + onnx_factory = _ort - def _feed(self): - # primer frames: the SDK init pre-buffers several frames before the - # source yields (validated on hardware); they are synthetic, so each - # one is registered as None and its result is dropped by collect(). - for _ in range(PRIMER_FRAMES): - self._pending.append(None) - yield np.zeros((self._height, self._width, 3), np.uint8) - while True: - frame = None - connection_id = None - try: - connection_id, frame = self._frame_q.get( - timeout=KEEPALIVE_INTERVAL - ) - except queue.Empty: - # safety net: the SDK source must never starve (stream init, - # scheduling gaps, true idle). spawn a synthetic zero frame; - # registered as None, its result is dropped by collect(). In - # steady state real frames keep the queue populated, so this - # rarely fires. - frame = np.zeros((self._height, self._width, 3), np.uint8) - if frame is None: # shutdown sentinel from stop() - return - # record this yield's request id (None for synthetics) so collect() - # can pop results back to the right connection in order. - self._pending.append(connection_id) - yield frame + 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 + ) - def _detection_rows(self, frame_result) -> list: - """Convert a stream result into (class_id, score, ymin, xmin, ymax, xmax) rows. - - Box coordinates are NORMALIZED [0..1]: Frigate's detect() multiplies - them by the region size, pixel boxes are discarded as outside the frame - (zero events despite working detections). - """ - dets = getattr(frame_result, "detections", None) - if dets is None: - return [] - try: - # some SDK builds return a numpy array; `or []` on an array raises - # ValueError (ambiguous truth value) with 2+ detections - dets = list(dets) - except TypeError: - dets = [dets] - rows = [] - for det in dets: - label = getattr(getattr(det, "label", None), "name", None) - if label is None: - continue - class_id = self._labelmap.get(str(label).strip()) - if class_id is None: - continue - raw_box = getattr(det, "box", None) - if raw_box is None: - continue - try: - # the SDK returns box as a numpy array in some builds; `or []` - # on an array raises ValueError (ambiguous truth value) - box = [float(v) for v in raw_box] - score = float(getattr(det, "score", 0.0) or 0.0) - except (TypeError, ValueError): - continue - if len(box) != 4: - continue - x0, y0, x1, y1 = box - rows.append( - ( - class_id, - score, - max(0.0, y0) / self._height, - max(0.0, x0) / self._width, - min(y1, self._height) / self._height, - min(x1, self._width) / self._width, - ) + 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}" ) - rows.sort(key=lambda r: r[1], reverse=True) - return rows[:MAX_ROWS] + self._pp = onnx_factory.InferenceSession( + pp_path, providers=["CPUExecutionProvider"] + ) + self._pp_input_names = [i.name for i in self._pp.get_inputs()] + if len(self._pp_input_names) != len(HEAD_REAL_CHANNELS): + raise ValueError( + "unexpected postprocess graph: expected " + f"{len(HEAD_REAL_CHANNELS)} inputs, got {len(self._pp_input_names)}" + ) + + # pre-allocated input/output buffers (reused; do not allocate per frame) + self._in_buf = np.full( + (1, PADDED_H, PADDED_W, 4), INPUT_Q_ZERO, np.int8 + ) + self._out_bufs = [ + np.zeros((1, MODEL_INPUT_H // s, MODEL_INPUT_W // s, 128), np.int8) + for s in (8, 8, 16, 16, 32, 32) + ] + # letterbox geometry from the frame coords (width x height) into 640x640 + self._scale = min( + MODEL_INPUT_W / self._width, MODEL_INPUT_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 = (MODEL_INPUT_H - self._new_h) // 2 + self._pad_l = (MODEL_INPUT_W - self._new_w) // 2 + + # dequantization per head: (int8 - zp) * scale (tuple stores scale, zp) + self._head_dequant = [ + (np.float32(scale), np.float32(zp)) for scale, zp in HEAD_DEQUANT + ] + + # -- public contract used by AxeleraDetector -------------------------- def submit(self, connection_id: str, frame: np.ndarray) -> None: - """Send one frame for inference, owned by `connection_id`. - - The frame is queued as-is (no pixel writes); `_feed` records the - connection id in yield order and `collect` pops it with the result. - """ + """Queue one frame for inference, owned by `connection_id`.""" try: self._frame_q.put_nowait((connection_id, frame)) except queue.Full: raise RuntimeError( "Axelera inference queue is full; restart required." ) from None + if self._thread is None or not self._thread.is_alive(): + self._thread = threading.Thread( + target=self._worker, name="axelera-runtime", daemon=True + ) + self._thread.start() def collect(self): - """Wait for the next completed frame; return (connection_id, rows). + """Return (connection_id, rows) for the next completed inference. - ``rows`` is a fixed (20, 6) float32 numpy array of normalized - detections. Primer/synthetic results (recorded as None) are dropped - internally. Returns ``(None, None)`` when the stream is stopped. + Returns (None, None) on stop so the async runner can exit cleanly. """ - while True: + while not self._stop.is_set(): try: - frame_result = next(self._results_iter) - except StopIteration: - return None, None - except Exception: - logger.exception( - "axelera: inference stream failed; detector requires restart" - ) - raise - out = _pad_rows(self._detection_rows(frame_result)) - try: - entry = self._pending.popleft() - except IndexError: - # a result with no registered yield: the SDK diverged from the - # recorded input order. Fail fast so the detector process is - # restarted, rather than silently misattribute every following - # result to the wrong connection (the P1 failure mode). - logger.error( - "axelera: inference result with no matching frame; " - "pairing desynced, restarting detector" - ) - raise RuntimeError( - "Axelera result/input pairing desynced" - ) from None - if entry is None: - # primer / keep-alive result: no request id recorded for it, so - # it is dropped and can never be delivered to a caller. + return self._result_q.get(timeout=0.25) + except queue.Empty: continue - return entry, out + return None, None def stop(self) -> None: - """Stop the stream and release the Metis device. Never raises.""" + """Stop the worker, release the Metis device. Never raises.""" + self._stop.set() try: - self._frame_q.put_nowait((None, None)) - except Exception: # noqa: S110, BLE001 + self._instance.free() + except Exception: # noqa: BLE001, S110 pass - try: - if hasattr(self, "_stream"): - self._stream.stop() - except Exception: # noqa: S110, BLE001 - pass - logger.info("axelera: inference stream stopped") + logger.info("axelera: runtime inference stopped") + + # -- internals --------------------------------------------------------- + + def _worker(self) -> None: + while not self._stop.is_set(): + try: + connection_id, frame = self._frame_q.get(timeout=0.25) + except queue.Empty: + continue + try: + rows = self._process(frame) + except Exception: + logger.exception( + "axelera: inference failed for %s", connection_id + ) + rows = np.zeros((MAX_ROWS, 6), np.float32) + try: + self._result_q.put_nowait((connection_id, rows)) + except queue.Full: + logger.error("axelera: result queue full; dropping frame") + + def _process(self, frame: np.ndarray) -> np.ndarray: + """Letterbox + quantize, run, decode, threshold + NMS -> (20,6) rows.""" + # letterbox the source (H, W, 3) into 640x640 and quantize to int8 + resized = cv2.resize( + frame, + (self._new_w, self._new_h), + interpolation=cv2.INTER_LINEAR, + ) + q = resized.astype(np.int16) - 128 # uint8 [0,255] -> int8 [-128,127] + q = np.clip(q, -128, 127).astype(np.int8) + buf = self._in_buf + 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 + # 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, y0 + 1 : y1 + 1, x0 + 1 : x1 + 1, 0:3] = q[..., ::-1] + buf[0, y0 + 1 : y1 + 1, x0 + 1 : x1 + 1, 3] = 127 + + self._instance.run([buf], self._out_bufs) + + # dequantize + crop to the real channel count + transpose NHWC -> NCHW + feeds = {} + for i, (name, obuf) in enumerate( + zip(self._pp_input_names, self._out_bufs) + ): + scale, zp = self._head_dequant[i] + real = HEAD_REAL_CHANNELS[i] + arr = obuf[..., :real].astype(np.float32) + arr = (arr - zp) * scale + feeds[name] = np.ascontiguousarray(arr.transpose(0, 3, 1, 2)) + + class_scores, boxes = self._pp.run(None, feeds) + scores = class_scores[0] # (8400, 80), sigmoid baked into the graph + boxes = boxes[0] # (8400, 4), xyxy in 640x640 input space + conf = scores.max(axis=1) + cls_idx = scores.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 ----------------- # @@ -384,7 +451,8 @@ class AxeleraDetector(DetectionApi): if model is None: raise ValueError( f"type={DETECTOR_KEY}: the `model:` block is required " - "(path to a compiled .axm, URL, or SDK model-zoo preset name)." + "(path to a compiled model directory/JSON, URL, or SDK " + "model-zoo preset name)." ) super().__init__(detector_config) @@ -393,10 +461,13 @@ class AxeleraDetector(DetectionApi): 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 - self.working_model = _resolve_network(self.model_path, self.model_url) - # map labels to class ids by NAME so the labelmap does not need to - # match the ordering the model was trained with + 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: @@ -404,15 +475,22 @@ class AxeleraDetector(DetectionApi): if not self._labelmap: logger.warning("axelera: empty labelmap: detections will not be mapped") - self._inference = _AxeleraInference( - network=self.working_model, + # 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: model=%s (%dx%d) ready", + "axelera: runtime model=%s (%dx%d input, frame %dx%d) ready", self.working_model, + MODEL_INPUT_W, + MODEL_INPUT_H, self.width, self.height, ) @@ -430,33 +508,16 @@ class AxeleraDetector(DetectionApi): if frame.shape[0] != 1: raise ValueError(f"Unexpected batch dimension in frame: {frame.shape}") frame = frame[0] - # Contiguous uint8 BGR for the SDK. No pixel writes happen here: the - # frame is queued and fed as-is (pairing is by recorded order, not by - # content). If `frame` is a view of the detect process's shared-memory - # buffer and already contiguous, `ascontiguousarray` returns it - # unchanged, so the SDK reads the shared buffer directly; the runner - # keeps it valid until our receive_output() returns. + # 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. - - On a fatal stream error the detector process exits so Frigate restarts - it (a silent (None, None) would stall detection forever). - """ - try: - return self._inference.collect() - except Exception: - logger.exception("axelera: fatal stream error; exiting detector process") - try: - self._inference.stop() - except Exception: # noqa: S110, BLE001 - pass - os._exit(1) + """Return (connection_id, rows) for the next completed inference.""" + return self._inference.collect() def shutdown(self) -> None: - """Gracefully shut down the Metis stream (called at detector stop).""" + """Gracefully shut down the runtime and release the Metis device.""" self._inference.stop() def detect_raw(self, tensor_input: np.ndarray): @@ -464,7 +525,7 @@ class AxeleraDetector(DetectionApi): return 0 # type: ignore[override] def close(self): - """Stop the inference stream and release the Metis device.""" + """Stop the inference runtime and release the Metis device.""" self._inference.stop() def __del__(self): @@ -475,6 +536,27 @@ class AxeleraDetector(DetectionApi): 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 ----------------- #