nik/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-10-18 07:29:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(axelera): accept batched NHWC frames (1, H, W, C) from Frigate

Browse Source 
Frigate feeds detect_raw() a batched NHWC frame (1, H, W, C); the plugin
raised ValueError for ndim != 3, so every real inference failed. Squeeze
the leading batch dim when present (same contract as other IPU plugins,
e.g. memryx). Regression test added.
This commit is contained in:
patrykorwat 2026-08-17 19:58:08 +00:00
parent d0537684c1
commit 9b731a14e9
2 changed files with 22 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
frigate/detectors/plugins/axelera.py
View File

@ -216,12 +216,20 @@ class _AxeleraInference:
)
def run(self, tensor_input, timeout: float = INFERENCE_TIMEOUT) -> np.ndarray:
if tensor_input is None or getattr(tensor_input, "ndim", 0) != 3:
if tensor_input is None or getattr(tensor_input, "ndim", 0) < 3:
raise ValueError(
f"Unexpected frame shape: {getattr(tensor_input, 'shape', tensor_input)}"
)
frame = np.ascontiguousarray(tensor_input)
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]
frame = np.ascontiguousarray(frame)
request_id = self._input.put(frame)
with self._lock:
self._pending.append(request_id)

12
frigate/test/test_axelera_detector.py
View File

@ -233,6 +233,18 @@ class TestAxeleraInference(unittest.TestCase):
finally:
engine.stop()
def test_batched_nhwc_frame_is_squeezed(self):
# Frigate feeds a batched NHWC frame (1, H, W, C); covers the live
# `type: zmq` worker crash (ValueError: unexpected shape) on every frame
engine = self._engine()
try:
flat = engine.run(np.zeros((320, 320, 3), np.uint8))
batched = engine.run(np.zeros((1, 320, 320, 3), np.uint8))
self.assertEqual(batched.shape, (20, 6))
np.testing.assert_array_equal(batched, flat)
finally:
engine.stop()
def test_timeout_returns_zeros_when_stream_blocks(self):
engine = self._engine(hold=True)
try: