fix(axelera): tolerate numpy-array boxes from SDK

The SDK returns detection boxes as numpy arrays on some frames; the
'list(box or [])' idiom evaluated array truthiness and raised ValueError,
killing the pump thread mid-detection. Extract boxes defensively. Live
repro: worker crashed the moment a real detection fired, so no event was
ever recorded.
This commit is contained in:
patrykorwat 2026-08-17 20:20:00 +00:00
parent 9b731a14e9
commit f3b97c27e9
2 changed files with 61 additions and 2 deletions

View File

@ -185,8 +185,17 @@ class _AxeleraInference:
class_id = self._labelmap.get(str(label).strip())
if class_id is None:
continue
score = float(getattr(det, "score", 0.0) or 0.0)
box = list(getattr(det, "box", None) or [])
score = 0.0
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

View File

@ -245,6 +245,56 @@ class TestAxeleraInference(unittest.TestCase):
finally:
engine.stop()
def test_numpy_box_is_accepted(self):
# regression: the SDK returns box as np.ndarray; `or []` on an array
# raised ValueError (ambiguous truth value), killing the pump thread
import numpy as _np
class NpBoxStream:
def __init__(self, network, sources=None):
self._src = sources[0] if sources else None
self.stopped = False
def __iter__(self):
return self
def __next__(self):
if self._src is not None:
self._src.send(None)
return SimpleNamespace(
detections=[
SimpleNamespace(
label=SimpleNamespace(name="truck"),
score=_np.float32(0.8),
box=_np.array([10.0, 20.0, 30.0, 40.0]),
)
]
)
def stop(self):
self.stopped = True
def factory(network, sources=None):
return NpBoxStream(network, sources)
engine = axelera._AxeleraInference(
network="x",
labelmap=make_labelmap(),
width=320,
height=320,
stream_factory=factory,
)
time.sleep(0.3)
try:
out = engine.run(np.zeros((320, 320, 3), np.uint8))
self.assertEqual(out.shape, (20, 6))
truck = [r for r in out if int(r[0]) == 7 and r[1] > 0]
self.assertTrue(truck, "expected truck row")
self.assertAlmostEqual(float(truck[0][1]), 0.8, places=2)
self.assertEqual(tuple(truck[0][2:6]), (20.0, 10.0, 40.0, 30.0))
finally:
engine.stop()
def test_timeout_returns_zeros_when_stream_blocks(self):
engine = self._engine(hold=True)
try: