fix(axelera): content-based pixel-token pairing for results (P1)

The SDK does not tag results with a per-frame id (image_id is always
empty, src_timestamp is a pipeline-clock value), so pairing relied on
FIFO ordering, which a buffered primer/keep-alive result could break:
the stale result would pop the pending id and be delivered to the
wrong camera/crop (synthetic results consuming real IDs).

Every real frame now carries a 32-bit request token written into a
small top-left pixel band; the SDK returns the source frame back 1:1
(verified on hardware: BGR pass-through exact, BGRA), so the pump
decodes the token from the result image and delivers the outcome only
to the frame that carries it (content-based, no ordering assumption).
Primer/keep-alive frames carry no token; their results are dropped
even when they surface while a request is pending.
This commit is contained in:
patrykorwat 2026-08-21 08:49:26 +00:00
parent 780ae97c26
commit fc28ee1892

View File

@ -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 "