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

test: drop axelera detector unit tests to match plugin conventions

Browse Source 
Other detector plugins shipping in this tree do not carry unit tests, and
the fixed-sleep timing in these tests (primer, keepalive, pump teardown)
is prone to flaking on CI runners. The regression coverage stays out of
tree beside the detector host worker.
This commit is contained in:
patrykorwat 2026-08-18 10:24:14 +00:00
parent 10b4394879
commit 5fa8bf75dd
1 changed files with 0 additions and 490 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

490
frigate/test/test_axelera_detector.py
View File

@ -1,490 +0,0 @@
"""Unit tests for the Axelera Metis detector plugin (`type: axelera`).
The Voyager SDK is never opened: the inference stream is replaced with a fake
stream implementing the same contract (yield one result per consumed frame).
The tests exercise the two layers independently:
- the engine (`_AxeleraInference`): request/response round-trip with label to
class mapping by name, padded (20, 6) output, timeout to zeros, dead-stream
detection, and the primer-frame orphan cache;
- the detector (`AxeleraDetector`): model resolution (model-zoo preset, local
path, URL downloaded to the model cache) and the clear ImportError when the
SDK is missing.
Frigate's modules are used when importable (CI/devcontainer). When they are
not, tiny stand-ins (same contracts) are registered so the file still runs
standalone; the plugin itself is loaded from this checkout either way.
"""
import importlib.util
import os
import sys
import tempfile
import threading
import time
import types
import unittest
from pathlib import Path
from types import SimpleNamespace
from typing import ClassVar
from unittest import mock
import numpy as np
_PLUGIN_PATH = (
Path(__file__).resolve().parents[2]
/ "frigate"
/ "detectors"
/ "plugins"
/ "axelera.py"
)
# ----------------- plugin import (Frigate if present, else stand-ins) ---------------- #
def _standin_modules():
"""Register minimal stand-ins for the Frigate modules the plugin touches."""
import queue
class _MockRequestStore:
def __init__(self):
self.counter = 0
self.input_queue = queue.Queue()
def put(self, tensor):
request_id = self.counter
self.counter += 1
self.input_queue.put((request_id, tensor))
return request_id
def get(self):
return self.input_queue.get()
class _MockResponseStore:
def __init__(self):
self.responses = {}
self.lock = threading.Lock()
self.cond = threading.Condition(self.lock)
def put(self, request_id, response):
with self.cond:
self.responses[request_id] = response
self.cond.notify_all()
def get(self, request_id, timeout=None):
with self.cond:
if not self.cond.wait_for(
lambda: request_id in self.responses, timeout=timeout
):
raise TimeoutError(f"Timeout waiting for response {request_id}")
return self.responses.pop(request_id)
class _MockDetectionApi:
def __init__(self, detector_config):
self.detector_config = detector_config
self.thresh = 0.4
self.height = getattr(detector_config.model, "height", None)
self.width = getattr(detector_config.model, "width", None)
class _MockBaseDetectorConfig:
def __init__(self, **kwargs):
self.model = kwargs.get("model")
def _mod(name, **attrs):
module = types.ModuleType(name)
for key, value in attrs.items():
setattr(module, key, value)
sys.modules[name] = module
return module
cache = os.path.join(tempfile.gettempdir(), "frigate-axelera-model-cache")
_mod("frigate")
_mod("frigate.const", MODEL_CACHE_DIR=cache)
_mod("frigate.detectors")
_mod("frigate.detectors.detection_api", DetectionApi=_MockDetectionApi)
_mod(
"frigate.detectors.detector_config", BaseDetectorConfig=_MockBaseDetectorConfig
)
_mod("frigate.object_detection")
_mod(
"frigate.object_detection.util",
RequestStore=_MockRequestStore,
ResponseStore=_MockResponseStore,
)
return cache
def _load_plugin():
try:
from frigate.detectors.plugins import axelera
except ImportError:
_standin_modules()
spec = importlib.util.spec_from_file_location(
"frigate.detectors.plugins.axelera", str(_PLUGIN_PATH)
)
if spec is None or spec.loader is None:
raise RuntimeError(f"cannot load {_PLUGIN_PATH}") from None
axelera = importlib.util.module_from_spec(spec)
spec.loader.exec_module(axelera)
return axelera
axelera = _load_plugin()
# ---------------- fake SDK stream ---------------- #
class _Hold:
"""Blocks forever so the pump stays alive without producing results."""
def wait(self, *args, **kwargs):
while True:
time.sleep(3600)
def fake_stream_factory(hold=False):
"""Build a Voyager-SDK-like stream factory.
Returns (factory, produced); factory(network, sources=[gen]) returns a
stream that consumes one frame from sources[0] per yielded result.
"""
class FakeStream:
def __init__(self, network, sources=None):
self._src = sources[0] if sources else None
self._hold = _Hold() if hold else None
self.stopped = False
def __next__(self):
if self._hold is not None:
self._hold.wait()
if self._src is not None:
self._src.send(None)
return SimpleNamespace(
detections=[
SimpleNamespace(
label=SimpleNamespace(name="truck"),
score=0.91,
box=[10.0, 20.0, 30.0, 40.0],
),
SimpleNamespace(
label=SimpleNamespace(name="person"),
score=0.55,
box=[5.0, 6.0, 15.0, 16.0],
),
# not present in the labelmap: must be filtered out
SimpleNamespace(
label=SimpleNamespace(name="unknown_label"),
score=0.99,
box=[1.0, 2.0, 3.0, 4.0],
),
]
)
def __iter__(self):
while True:
try:
yield next(self)
except StopIteration:
return
def stop(self):
self.stopped = True
def factory(network, sources=None):
return FakeStream(network, sources)
return factory
def make_labelmap():
# truck -> 7, person -> 0; unknown_label intentionally absent
return {"truck": 7, "person": 0}
# ---------------- engine tests ---------------- #
class TestAxeleraInference(unittest.TestCase):
def _engine(self, hold=False):
return axelera._AxeleraInference(
network="yolov8n-coco",
labelmap=make_labelmap(),
width=320,
height=320,
stream_factory=fake_stream_factory(hold=hold),
)
def test_roundtrip_maps_labels_by_name_and_pads(self):
engine = self._engine()
try:
out = engine.run(np.zeros((320, 320, 3), np.uint8))
self.assertEqual(out.shape, (20, 6))
self.assertEqual(out.dtype, np.float32)
ids = sorted(int(r[0]) for r in out if r[1] > 0)
self.assertEqual(ids, [0, 7]) # unknown_label dropped
rows = {int(r[0]): r for r in out if r[1] > 0}
# rows are [class_id, score, ymin, xmin, ymax, xmax] NORMALIZED (0..1)
self.assertEqual(tuple(rows[7][2:6]), (0.0625, 0.03125, 0.125, 0.09375))
self.assertEqual(rows[0][1], 0.55)
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_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]), (0.0625, 0.03125, 0.125, 0.09375))
finally:
engine.stop()
def test_run_waits_for_new_result_not_cached(self):
# latest-result contract: run() must return a result NEWER than at call
# time, not a stale cached one (e.g. an empty keep-alive result)
state = {"n": 0}
class GatedStream:
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)
state["n"] += 1
if state["n"] <= 4: # primer (3) + keep-alive (1): empty result
dets = []
else:
dets = [
SimpleNamespace(
label=SimpleNamespace(name="truck"),
score=0.9,
box=[1.0, 2.0, 3.0, 4.0],
)
]
return SimpleNamespace(detections=dets)
def stop(self):
self.stopped = True
def factory(network, sources=None):
return GatedStream(network, sources)
engine = axelera._AxeleraInference(
network="x",
labelmap=make_labelmap(),
width=320,
height=320,
stream_factory=factory,
)
time.sleep(0.4) # primer consumed (3)
time.sleep(1.2) # keep-alive: feeder pushes a zero frame (4) -> empty latest
try:
out = engine.run(np.zeros((320, 320, 3), np.uint8))
truck = [r for r in out if int(r[0]) == 7 and r[1] > 0]
self.assertTrue(
truck, "expected a FRESH result (truck), not the stale keep-alive cache"
)
finally:
engine.stop()
def test_timeout_returns_zeros_when_stream_blocks(self):
engine = self._engine(hold=True)
try:
out = engine.run(np.zeros((320, 320, 3), np.uint8))
self.assertEqual(out.shape, (20, 6))
self.assertFalse(out.any())
finally:
engine.stop()
def test_dead_stream_raises_runtime_error(self):
class CrashStream:
def __iter__(self):
raise RuntimeError("boom")
def factory(network, sources=None):
return CrashStream()
engine = axelera._AxeleraInference(
network="x", labelmap={}, width=320, height=320, stream_factory=factory
)
time.sleep(0.3) # let the pump thread die
try:
with self.assertRaises(RuntimeError):
engine.run(np.zeros((320, 320, 3), np.uint8))
finally:
engine.stop()
def test_primer_results_are_dropped_requests_served(self):
# primer (3 ramki zerowe) i ramki keepalive nie "zjadaja" requestow:
# pierwsze zapytanie i kolejne dostaja wlasne wyniki
engine = self._engine()
try:
out = engine.run(np.zeros((320, 320, 3), np.uint8))
self.assertEqual(out.shape, (20, 6))
self.assertTrue(out.any())
out2 = engine.run(np.zeros((320, 320, 3), np.uint8))
self.assertTrue(out2.any())
finally:
engine.stop()
# ---------------- detector configuration resolution tests ---------------- #
class _Recorder:
"""Stand-in for _AxeleraInference that records the resolved network."""
networks: ClassVar[list[str]] = []
def __init__(self, network, labelmap, width, height, stream_factory=None):
self.network = network
self.__class__.networks.append(network)
def stop(self):
pass
def _make_model(path=None, width=320, height=320):
return SimpleNamespace(
path=path,
width=width,
height=height,
merged_labelmap={7: "truck", 0: "person"},
)
class TestAxeleraDetectorConfig(unittest.TestCase):
def _build(self, model, engine_cls=None):
cfg = SimpleNamespace(model=model)
patch_target = engine_cls if engine_cls is not None else _Recorder
with mock.patch.object(axelera, "_AxeleraInference", patch_target):
detector = object.__new__(axelera.AxeleraDetector)
detector.detector_config = cfg
detector.thresh = 0.4
detector.height = model.height
detector.width = model.width
axelera.AxeleraDetector.__init__(detector, cfg)
return detector
def test_resolves_preset_name(self):
detector = self._build(_make_model(path="yolov8n-coco"))
self.assertEqual(detector.working_model, "yolov8n-coco")
self.assertIn("yolov8n-coco", _Recorder.networks)
def test_resolves_default_downloads_from_url(self):
with tempfile.TemporaryDirectory() as tmp:
downloaded = {}
with (
mock.patch.object(axelera, "MODEL_CACHE_DIR", tmp),
mock.patch.object(
axelera.urllib.request, "urlretrieve", autospec=True
) as retrieve,
):
retrieve.side_effect = lambda url, dest: downloaded.__setitem__(
dest, url
)
detector = self._build(_make_model(path=None))
cached = os.path.join(tmp, "yolov8n-coco.axm")
self.assertEqual(detector.working_model, cached)
self.assertEqual(downloaded.get(cached), axelera.AX_DEFAULT_URL)
def test_resolves_local_path(self):
with tempfile.TemporaryDirectory() as tmp:
local = os.path.join(tmp, "model.axm")
with open(local, "wb") as f:
f.write(b"model")
detector = self._build(_make_model(path=local))
self.assertEqual(detector.working_model, local)
def test_resolves_url_downloads_to_cache(self):
with tempfile.TemporaryDirectory() as tmp:
downloaded = {}
with (
mock.patch.object(axelera, "MODEL_CACHE_DIR", tmp),
mock.patch.object(
axelera.urllib.request, "urlretrieve", autospec=True
) as retrieve,
):
retrieve.side_effect = lambda url, dest: downloaded.__setitem__(
dest, url
)
detector = self._build(_make_model(path="https://hf/x.axm"))
cached = os.path.join(tmp, "x.axm")
self.assertEqual(detector.working_model, cached)
self.assertEqual(downloaded.get(cached), "https://hf/x.axm")
def test_missing_sdk_raises_clear_import_error(self):
with self.assertRaises(ImportError) as ctx:
axelera._AxeleraInference(
network=axelera.AX_DEFAULT_NETWORK,
labelmap={},
width=320,
height=320,
stream_factory=None,
)
self.assertIn("Voyager SDK", str(ctx.exception))
if __name__ == "__main__":
unittest.main()