diff --git a/frigate/detectors/plugins/axelera.py b/frigate/detectors/plugins/axelera.py index e3ea5436e7..d83a8629e5 100644 --- a/frigate/detectors/plugins/axelera.py +++ b/frigate/detectors/plugins/axelera.py @@ -24,7 +24,6 @@ Model configuration follows the hardware-backend convention: - `model.width` / `model.height`: input frame dimensions for the network. """ -import collections import logging import os import queue @@ -53,6 +52,21 @@ INFERENCE_TIMEOUT = 5.0 MAX_ROWS = 20 PRIMER_FRAMES = 3 +# Per-frame tag: the SDK returns each source frame back 1:1 in the result +# (verified on hardware: BGR pass-through is exact, image dimension preserved), +# but it provides NO per-frame id (image_id is always empty, src_timestamp is a +# pipeline-clock value, not an id). We therefore write a 32-bit request token +# into a small pixel band in the top-left corner of every REAL frame and decode +# it from the result image, so results are paired to requests by CONTENT. This +# is immune to the SDK buffering, priming or reordering results: a buffered +# primer/keep-alive result carries token 0 and is dropped, and a real result +# can only be delivered to the frame that carries its token. +TOKEN_BITS = 32 +TOKEN_ROWS = (TOKEN_BITS + 2) // 3 # 11 rows x 3 channel-cells = 33 (one spare) +TOKEN_X0 = 4 +TOKEN_CELL_W = 4 +TOKEN_NO_ID = 0 + # ----------------- Utility Functions ----------------- # @@ -114,6 +128,44 @@ def _pad_rows(rows) -> np.ndarray: return out +def _stamp_token(frame: np.ndarray, token: int) -> None: + """Write a 32-bit request token into the top-left pixel band of `frame`. + + Band geometry must match `_read_token`. The values 0/255 survive the SDK's + BGR->BGRA pass-through exactly (verified on hardware). + """ + band = frame[0:TOKEN_ROWS, TOKEN_X0 : TOKEN_X0 + TOKEN_CELL_W] + for i in range(TOKEN_BITS): + row, ch = divmod(i, 3) + val = 255 if (token >> (TOKEN_BITS - 1 - i)) & 1 else 0 + band[row, :, ch] = val + + +def _read_token(frame_result) -> int: + """Decode the request token stamped on a stream result's source frame. + + The SDK returns the source frame 1:1 (BGRA); we read the matching band from + the first 3 channels. Returns TOKEN_NO_ID for tagless results (primer / + keep-alive, or a frame_result whose image is unavailable). + """ + img = getattr(frame_result, "image", None) + if img is None: + return TOKEN_NO_ID + try: + arr = img.asarray() + except Exception: # noqa: BLE001 + return TOKEN_NO_ID + if getattr(arr, "ndim", 0) < 3: + return TOKEN_NO_ID + band = arr[0:TOKEN_ROWS, TOKEN_X0 : TOKEN_X0 + TOKEN_CELL_W, 0:3] + token = 0 + for i in range(TOKEN_BITS): + row, ch = divmod(i, 3) + bit = 1 if band[row, :, ch].mean() > 127 else 0 + token = (token << 1) | bit + return token + + # ----------------- AxeleraInference (engine) ----------------- # @@ -129,12 +181,13 @@ class _AxeleraInference: never starves and is not torn down by its internal inference timeout during quiet periods (e.g. a camera with no motion). - Results are STRICTLY paired to requests: every frame the feeder pushes into - the SDK registers its request id on a deque (in submission order), and each - stream result pops the next pending id and stores the outcome under it. - ``run()`` waits for its own id only, so a result can never be attributed to - a different camera or crop, and primer/keep-alive results (unregistered ids) - are dropped instead of being returned to a caller. + Results are STRICTLY paired to requests by CONTENT: every real frame is + stamped with a 32-bit request token in its top-left pixel band, the pump + decodes the token from the result image and delivers the outcome only to + that frame's caller. Primer/keep-alive frames carry no token, so their + results are provably dropped even if the SDK buffers or reorders them — + no FIFO ordering assumption remains, and a buffered result can never be + attributed to a different camera or crop. """ def __init__( @@ -144,14 +197,15 @@ class _AxeleraInference: self._width, self._height = int(width), int(height) self._frame_q: queue.Queue = queue.Queue(maxsize=64) self._cond = threading.Condition() - self._pending_ids: collections.deque = collections.deque() + # request ids a caller is still waiting on (guarded by _cond); the pump + # only delivers results whose decoded token is in this set + self._waiting: set = set() self._results: dict[int, np.ndarray] = {} self._next_id = 0 - # Serializes run() calls: the SDK stream does not tag results with a - # per-frame id and can reorder/buffer results under sustained load - # (observed after many thousands of frames), so with several frames in - # flight the FIFO pop can misattribute a result to the wrong camera. - # With at most one frame in flight the result can only belong to it. + # Serializes run() calls: pairing is content-based (see _read_token), + # so concurrent frames would be perfectly matched too, but keeping at + # most ONE frame in flight gives fully deterministic behaviour and + # bounds the SDK's internal buffering. self._run_lock = threading.Lock() if stream_factory is None: @@ -196,9 +250,9 @@ class _AxeleraInference: if frame is None: # shutdown sentinel from stop() return if frame_id is not None: - # submission order == result order: the pump pops this id when - # the SDK yields the result for this frame. - self._pending_ids.append(frame_id) + # Tag the frame with its request id; the pump decodes it from + # the result image (content-based pairing, no ordering assumed). + _stamp_token(frame, frame_id) yield frame def _detection_rows(self, frame_result) -> list: @@ -256,13 +310,20 @@ class _AxeleraInference: try: for frame_result in self._stream: out = _pad_rows(self._detection_rows(frame_result)) - if self._pending_ids: - # FIFO: the stream returns results in submission order, so - # the oldest pending id belongs to this result. - frame_id = self._pending_ids.popleft() - with self._cond: - self._results[frame_id] = out - self._cond.notify_all() + token = _read_token(frame_result) + if token == TOKEN_NO_ID: + # primer / keep-alive result: it carries no request token, + # so it can NEVER be delivered to a caller, even if it + # surfaces while a request is pending. This is what makes a + # buffered/synthetic result provably unable to steal a + # request id (cf. prior FIFO pairing). + continue + with self._cond: + if token not in self._waiting: + # stale result for a frame that already timed out + continue + self._results[token] = out + self._cond.notify_all() except Exception: logger.exception( "axelera: inference stream failed; detector requires restart" @@ -301,13 +362,18 @@ class _AxeleraInference: # permanently wedging this detector. raise RuntimeError("Axelera inference thread stopped; restart required.") - # Every submitted frame gets a unique id; the pump stores the result - # under this id, so we can wait for OUR frame's outcome exclusively. + # Every submitted frame gets a unique id; its token is decoded from the + # result image by the pump, so we can wait for OUR frame's outcome + # exclusively (content-based pairing, no ordering assumptions). self._next_id += 1 frame_id = self._next_id + with self._cond: + self._waiting.add(frame_id) try: self._frame_q.put((frame_id, frame), timeout=2.0) except queue.Full: + with self._cond: + self._waiting.discard(frame_id) raise RuntimeError( "Axelera inference queue is full; restart required." ) from None @@ -321,7 +387,11 @@ class _AxeleraInference: self._cond.wait(remaining) if frame_id in self._results: result = self._results.pop(frame_id) + self._waiting.discard(frame_id) return result + # timeout: drop the id so a late result is not delivered/stored + with self._cond: + self._waiting.discard(frame_id) if self._thread.is_alive(): raise RuntimeError( "Axelera inference stream returned no result for the submitted "