mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-18 07:29:45 +00:00
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.
This commit is contained in:
patrykorwat
parent
65a429ab33
commit
92112cf909
@ -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 ----------------- #
|
||||
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user