693 lines
22 KiB
Python
693 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import asdict
|
|
import json
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from cmvr_edge_ai.contracts.events import (
|
|
BoundingBox,
|
|
Detection,
|
|
DetectionAlert,
|
|
DetectionResult,
|
|
)
|
|
from cmvr_edge_ai.contracts.media import EncodedImage, ImageFrame
|
|
from cmvr_edge_ai.core import ComponentContext, Emission, Envelope
|
|
import cmvr_edge_ai.detection.rules as detection_rules
|
|
from cmvr_edge_ai.detection.annotation import ImageAnnotationDependencyError
|
|
from cmvr_edge_ai.detection.rules import DetectionTriggerOperator
|
|
|
|
|
|
NS_PER_MS = 1_000_000
|
|
|
|
|
|
def _frame(value: int = 0) -> ImageFrame:
|
|
return ImageFrame(bytes([value]) * 4 * 4 * 3, 4, 4, "BGR8")
|
|
|
|
|
|
async def _keep_restricted_event_loop_responsive() -> None:
|
|
while True:
|
|
await asyncio.sleep(0.01)
|
|
|
|
|
|
def _detection(
|
|
label: str = "No-Helmet",
|
|
confidence: float = 0.9,
|
|
*,
|
|
track_id: str | None = None,
|
|
) -> Detection:
|
|
return Detection(
|
|
label=label,
|
|
confidence=confidence,
|
|
box=BoundingBox(1.0, 2.0, 30.0, 40.0),
|
|
track_id=track_id,
|
|
)
|
|
|
|
|
|
def _envelope(
|
|
sequence: int,
|
|
at_ms: float,
|
|
*detections: Detection,
|
|
source_id: str = "camera-a",
|
|
captured: bool = False,
|
|
trace_id: str | None = None,
|
|
frame: ImageFrame | None = None,
|
|
) -> Envelope[DetectionResult]:
|
|
timestamp_ns = int(at_ms * NS_PER_MS)
|
|
kwargs: dict[str, Any] = {
|
|
"payload": DetectionResult(
|
|
detections=tuple(detections),
|
|
model_id="construction-ppe-yolov8@1",
|
|
inference_ms=12.5,
|
|
model_name="Construction PPE YOLOv8",
|
|
source_frame=frame,
|
|
),
|
|
"schema_name": "DetectionResult",
|
|
"schema_version": 1,
|
|
"source_id": source_id,
|
|
"sequence": sequence,
|
|
"received_at_ns": timestamp_ns,
|
|
"captured_at_ns": timestamp_ns if captured else None,
|
|
}
|
|
if trace_id is not None:
|
|
kwargs["trace_id"] = trace_id
|
|
return Envelope(**kwargs)
|
|
|
|
|
|
def _operator(
|
|
*,
|
|
labels: list[str] | None = None,
|
|
min_hits: int = 3,
|
|
window_ms: float = 1_000,
|
|
cooldown_ms: float = 0,
|
|
min_confidence: float = 0.5,
|
|
scope: str = "source",
|
|
time_source: str = "received",
|
|
alert_image: dict[str, Any] | None = None,
|
|
) -> DetectionTriggerOperator:
|
|
params: dict[str, Any] = {
|
|
"time_source": time_source,
|
|
"rules": [
|
|
{
|
|
"id": "missing-ppe",
|
|
"labels": labels or ["No-Helmet"],
|
|
"min_hits": min_hits,
|
|
"window_ms": window_ms,
|
|
"cooldown_ms": cooldown_ms,
|
|
"min_confidence": min_confidence,
|
|
"scope": scope,
|
|
}
|
|
],
|
|
}
|
|
if alert_image is not None:
|
|
params["alert_image"] = alert_image
|
|
return DetectionTriggerOperator("trigger", params)
|
|
|
|
|
|
def _process(
|
|
operator: DetectionTriggerOperator, envelope: Envelope[Any]
|
|
) -> tuple[Emission, ...] | None:
|
|
return asyncio.run(operator.process(envelope))
|
|
|
|
|
|
def _single_alert(
|
|
result: tuple[Emission, ...] | None
|
|
) -> tuple[Emission, DetectionAlert]:
|
|
assert result is not None
|
|
assert len(result) == 1
|
|
emission = result[0]
|
|
assert isinstance(emission.envelope.payload, DetectionAlert)
|
|
return emission, emission.envelope.payload
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("params", "message"),
|
|
[
|
|
({"rules": [], "unexpected": True}, "unknown detection trigger"),
|
|
({"rules": []}, "non-empty list"),
|
|
(
|
|
{"rules": [], "alert_image": "enabled"},
|
|
"alert_image must be a mapping",
|
|
),
|
|
(
|
|
{"rules": [], "alert_image": {"unexpected": True}},
|
|
"alert_image has unknown field",
|
|
),
|
|
(
|
|
{"rules": [], "alert_image": {"enabled": "true"}},
|
|
"alert_image.enabled must be a boolean",
|
|
),
|
|
(
|
|
{"rules": [], "alert_image": {"jpeg_quality": True}},
|
|
"alert_image.jpeg_quality must be an integer",
|
|
),
|
|
(
|
|
{"rules": [], "alert_image": {"jpeg_quality": 0}},
|
|
"alert_image.jpeg_quality must be between 1 and 95",
|
|
),
|
|
(
|
|
{"rules": [], "alert_image": {"jpeg_quality": 96}},
|
|
"alert_image.jpeg_quality must be between 1 and 95",
|
|
),
|
|
(
|
|
{
|
|
"rules": [
|
|
{
|
|
"id": "r",
|
|
"labels": ["No-Helmet"],
|
|
"min_hits": 2,
|
|
"window_ms": 100,
|
|
"typo": True,
|
|
}
|
|
]
|
|
},
|
|
"unknown field",
|
|
),
|
|
(
|
|
{
|
|
"rules": [
|
|
{
|
|
"id": "r",
|
|
"labels": ["No-Helmet", "No-Helmet"],
|
|
"min_hits": 2,
|
|
"window_ms": 100,
|
|
}
|
|
]
|
|
},
|
|
"duplicates",
|
|
),
|
|
(
|
|
{
|
|
"rules": [
|
|
{
|
|
"id": "r",
|
|
"labels": ["No-Helmet"],
|
|
"min_hits": True,
|
|
"window_ms": 100,
|
|
}
|
|
]
|
|
},
|
|
"must be an integer",
|
|
),
|
|
(
|
|
{
|
|
"time_source": "wall",
|
|
"rules": [
|
|
{
|
|
"id": "r",
|
|
"labels": ["No-Helmet"],
|
|
"min_hits": 2,
|
|
"window_ms": 100,
|
|
}
|
|
],
|
|
},
|
|
"time_source",
|
|
),
|
|
],
|
|
)
|
|
def test_trigger_configuration_is_strict(params: dict[str, Any], message: str) -> None:
|
|
with pytest.raises(ValueError, match=message):
|
|
DetectionTriggerOperator("trigger", params)
|
|
|
|
|
|
def test_source_rule_counts_distinct_frames_not_boxes_and_preserves_metadata() -> None:
|
|
operator = _operator(min_hits=3, window_ms=1_000, min_confidence=0.7)
|
|
|
|
first = _envelope(
|
|
1,
|
|
0,
|
|
_detection(confidence=0.8),
|
|
_detection(confidence=0.91),
|
|
_detection("No-Vest", confidence=0.99),
|
|
)
|
|
assert _process(operator, first) is None
|
|
# Replaying a frame with the same source and sequence must not create a hit.
|
|
assert _process(operator, _envelope(1, 100, _detection(confidence=0.99))) is None
|
|
# Below-threshold detections do not count.
|
|
assert _process(operator, _envelope(2, 500, _detection(confidence=0.69))) is None
|
|
assert _process(operator, _envelope(2, 500, _detection(confidence=0.85))) is None
|
|
|
|
threshold = _envelope(
|
|
3,
|
|
1_000,
|
|
_detection(confidence=0.95),
|
|
trace_id="threshold-frame",
|
|
)
|
|
emission, alert = _single_alert(_process(operator, threshold))
|
|
|
|
assert emission.port == "alerts"
|
|
assert emission.envelope.schema == "DetectionAlert/v1"
|
|
assert emission.envelope.trace_id == "threshold-frame"
|
|
assert emission.envelope.source_id == "camera-a"
|
|
assert alert.rule_id == "missing-ppe"
|
|
assert alert.model_id == "construction-ppe-yolov8@1"
|
|
assert alert.model_name == "Construction PPE YOLOv8"
|
|
assert alert.labels == ("No-Helmet",)
|
|
assert alert.scope == "source"
|
|
assert alert.scope_id == "camera-a"
|
|
assert alert.hit_count == 3
|
|
assert alert.window_ms == 1_000
|
|
assert alert.first_seen_ns == 0
|
|
assert alert.last_seen_ns == 1_000 * NS_PER_MS
|
|
assert alert.triggered_at_ns == 1_000 * NS_PER_MS
|
|
assert alert.max_confidence == pytest.approx(0.95)
|
|
assert alert.detections == threshold.payload.detections
|
|
assert alert.event_id
|
|
|
|
|
|
def test_invalid_detections_are_counted_and_never_reach_an_alert_payload() -> None:
|
|
operator = _operator(min_hits=1)
|
|
invalid_nan = Detection(
|
|
"No-Helmet",
|
|
0.99,
|
|
BoundingBox(float("nan"), 2.0, 30.0, 40.0),
|
|
)
|
|
invalid_reversed = Detection(
|
|
"No-Helmet",
|
|
0.98,
|
|
BoundingBox(30.0, 40.0, 1.0, 2.0),
|
|
)
|
|
valid = _detection(confidence=0.95)
|
|
|
|
assert _process(operator, _envelope(1, 0, invalid_nan, invalid_reversed)) is None
|
|
_, alert = _single_alert(
|
|
_process(
|
|
operator,
|
|
_envelope(2, 10, invalid_nan, invalid_reversed, valid),
|
|
)
|
|
)
|
|
|
|
assert alert.detections == (valid,)
|
|
# The remaining structured alert can be emitted by a strict JSON client.
|
|
json.dumps(asdict(alert), allow_nan=False)
|
|
health = asyncio.run(operator.health())
|
|
assert health.healthy is False
|
|
assert "invalid_detections=4" in health.detail
|
|
|
|
|
|
def test_window_lower_boundary_is_inclusive_and_older_hits_expire() -> None:
|
|
inclusive = _operator(min_hits=2, window_ms=1_000)
|
|
assert _process(inclusive, _envelope(1, 0, _detection())) is None
|
|
_, alert = _single_alert(_process(inclusive, _envelope(2, 1_000, _detection())))
|
|
assert alert.first_seen_ns == 0
|
|
|
|
expired = _operator(min_hits=2, window_ms=1_000)
|
|
assert _process(expired, _envelope(1, 0, _detection())) is None
|
|
assert _process(expired, _envelope(2, 1_001, _detection())) is None
|
|
|
|
|
|
def test_cooldown_does_not_accumulate_hits_and_requires_a_fresh_window() -> None:
|
|
operator = _operator(min_hits=2, window_ms=1_000, cooldown_ms=500)
|
|
assert _process(operator, _envelope(1, 0, _detection())) is None
|
|
_single_alert(_process(operator, _envelope(2, 100, _detection())))
|
|
|
|
# The threshold frame itself remains deduplicated even after the hit deque
|
|
# was reset, and new frames during cooldown are ignored rather than queued.
|
|
assert _process(operator, _envelope(2, 100, _detection())) is None
|
|
assert _process(operator, _envelope(3, 200, _detection())) is None
|
|
assert _process(operator, _envelope(4, 599, _detection())) is None
|
|
assert _process(operator, _envelope(5, 600, _detection())) is None
|
|
_, alert = _single_alert(_process(operator, _envelope(6, 700, _detection())))
|
|
assert alert.first_seen_ns == 600 * NS_PER_MS
|
|
assert alert.last_seen_ns == 700 * NS_PER_MS
|
|
|
|
health = asyncio.run(operator.health())
|
|
assert health.healthy is True
|
|
assert "alerts_emitted=2" in health.detail
|
|
assert "cooldown_suppressed=2" in health.detail
|
|
assert "duplicate_frames=1" in health.detail
|
|
|
|
|
|
def test_source_state_is_isolated_per_camera() -> None:
|
|
operator = _operator(min_hits=2)
|
|
assert (
|
|
_process(operator, _envelope(1, 0, _detection(), source_id="camera-a")) is None
|
|
)
|
|
assert (
|
|
_process(operator, _envelope(1, 10, _detection(), source_id="camera-b")) is None
|
|
)
|
|
|
|
_, alert = _single_alert(
|
|
_process(operator, _envelope(2, 20, _detection(), source_id="camera-a"))
|
|
)
|
|
assert alert.scope_id == "camera-a"
|
|
|
|
|
|
def test_track_rules_are_isolated_and_can_emit_multiple_alerts_per_frame() -> None:
|
|
operator = _operator(min_hits=2, scope="track")
|
|
first = _envelope(
|
|
1,
|
|
0,
|
|
_detection(track_id="worker-1"),
|
|
_detection(confidence=0.95, track_id="worker-1"),
|
|
_detection(track_id="worker-2"),
|
|
_detection(track_id=None),
|
|
)
|
|
assert _process(operator, first) is None
|
|
|
|
result = _process(
|
|
operator,
|
|
_envelope(
|
|
2,
|
|
100,
|
|
_detection(track_id="worker-1"),
|
|
_detection(track_id="worker-2"),
|
|
),
|
|
)
|
|
assert result is not None
|
|
assert {item.envelope.payload.scope_id for item in result} == {
|
|
"worker-1",
|
|
"worker-2",
|
|
}
|
|
assert all(item.envelope.payload.hit_count == 2 for item in result)
|
|
assert all(item.envelope.payload.scope == "track" for item in result)
|
|
assert "untracked_detections=1" in asyncio.run(operator.health()).detail
|
|
|
|
|
|
def test_multiple_rules_can_trigger_from_one_result() -> None:
|
|
operator = DetectionTriggerOperator(
|
|
"trigger",
|
|
{
|
|
"time_source": "received",
|
|
"rules": [
|
|
{
|
|
"id": "helmet",
|
|
"labels": ["No-Helmet"],
|
|
"min_hits": 1,
|
|
"window_ms": 100,
|
|
"cooldown_ms": 0,
|
|
"min_confidence": 0.5,
|
|
"scope": "source",
|
|
},
|
|
{
|
|
"id": "vest",
|
|
"labels": ["No-Vest"],
|
|
"min_hits": 1,
|
|
"window_ms": 100,
|
|
"cooldown_ms": 0,
|
|
"min_confidence": 0.5,
|
|
"scope": "source",
|
|
},
|
|
],
|
|
},
|
|
)
|
|
result = _process(
|
|
operator,
|
|
_envelope(1, 0, _detection("No-Helmet"), _detection("No-Vest")),
|
|
)
|
|
|
|
assert result is not None
|
|
assert [item.envelope.payload.rule_id for item in result] == ["helmet", "vest"]
|
|
|
|
|
|
def test_alert_image_is_encoded_once_only_when_the_threshold_frame_triggers(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
render_calls: list[
|
|
tuple[int, ImageFrame, tuple[Detection, ...], int]
|
|
] = []
|
|
encoded = EncodedImage(b"jpeg", "image/jpeg", 4, 4)
|
|
|
|
def render(
|
|
frame: ImageFrame,
|
|
detections: tuple[Detection, ...],
|
|
*,
|
|
jpeg_quality: int,
|
|
) -> EncodedImage:
|
|
render_calls.append(
|
|
(threading.get_ident(), frame, tuple(detections), jpeg_quality)
|
|
)
|
|
return encoded
|
|
|
|
monkeypatch.setattr(detection_rules, "ensure_annotation_support", lambda: None)
|
|
monkeypatch.setattr(detection_rules, "render_annotated_jpeg", render)
|
|
operator = _operator(
|
|
min_hits=2,
|
|
alert_image={"enabled": True, "jpeg_quality": 73},
|
|
)
|
|
empty_frame = _frame(1)
|
|
first_hit_frame = _frame(2)
|
|
threshold_frame = _frame(3)
|
|
threshold_detection = _detection(confidence=0.96)
|
|
|
|
async def scenario() -> tuple[Emission, ...] | None:
|
|
event_loop_thread = threading.get_ident()
|
|
executor = ThreadPoolExecutor(max_workers=1)
|
|
ticker = asyncio.create_task(_keep_restricted_event_loop_responsive())
|
|
try:
|
|
await operator.setup(
|
|
ComponentContext(
|
|
pipeline_id="detection",
|
|
node_id="trigger",
|
|
shutdown_event=asyncio.Event(),
|
|
metadata={"thread_executor": executor},
|
|
)
|
|
)
|
|
assert (
|
|
await operator.process(_envelope(1, 0, frame=empty_frame)) is None
|
|
)
|
|
assert (
|
|
await operator.process(
|
|
_envelope(2, 100, _detection(), frame=first_hit_frame)
|
|
)
|
|
is None
|
|
)
|
|
assert render_calls == []
|
|
result = await operator.process(
|
|
_envelope(
|
|
3,
|
|
200,
|
|
threshold_detection,
|
|
frame=threshold_frame,
|
|
)
|
|
)
|
|
assert render_calls[0][0] != event_loop_thread
|
|
return result
|
|
finally:
|
|
ticker.cancel()
|
|
await asyncio.gather(ticker, return_exceptions=True)
|
|
await operator.stop()
|
|
executor.shutdown(wait=True, cancel_futures=True)
|
|
|
|
result = asyncio.run(scenario())
|
|
_, alert = _single_alert(result)
|
|
|
|
assert len(render_calls) == 1
|
|
_, rendered_frame, rendered_detections, quality = render_calls[0]
|
|
assert rendered_frame is threshold_frame
|
|
assert rendered_detections == (threshold_detection,)
|
|
assert quality == 73
|
|
assert alert.image is encoded
|
|
health = asyncio.run(operator.health())
|
|
assert health.healthy is True
|
|
assert "alert_images_encoded=1" in health.detail
|
|
assert "alert_images_attached=1" in health.detail
|
|
|
|
|
|
def test_multiple_alerts_from_one_frame_share_one_union_image(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
helmet = _detection("No-Helmet", confidence=0.91)
|
|
vest = _detection("No-Vest", confidence=0.88)
|
|
threshold_frame = _frame(4)
|
|
encoded = EncodedImage(b"shared-jpeg", "image/jpeg", 4, 4)
|
|
render_calls: list[tuple[ImageFrame, tuple[Detection, ...]]] = []
|
|
|
|
def render(
|
|
frame: ImageFrame,
|
|
detections: tuple[Detection, ...],
|
|
*,
|
|
jpeg_quality: int,
|
|
) -> EncodedImage:
|
|
assert jpeg_quality == 85
|
|
render_calls.append((frame, tuple(detections)))
|
|
return encoded
|
|
|
|
async def run_inline(
|
|
function: Any,
|
|
*args: Any,
|
|
executor: Any = None,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
del executor
|
|
return function(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(detection_rules, "render_annotated_jpeg", render)
|
|
monkeypatch.setattr(detection_rules, "run_blocking", run_inline)
|
|
operator = DetectionTriggerOperator(
|
|
"trigger",
|
|
{
|
|
"time_source": "received",
|
|
"alert_image": {"enabled": True},
|
|
"rules": [
|
|
{
|
|
"id": "helmet",
|
|
"labels": ["No-Helmet"],
|
|
"min_hits": 1,
|
|
"window_ms": 100,
|
|
},
|
|
{
|
|
"id": "vest",
|
|
"labels": ["No-Vest"],
|
|
"min_hits": 1,
|
|
"window_ms": 100,
|
|
},
|
|
],
|
|
},
|
|
)
|
|
|
|
result = _process(
|
|
operator,
|
|
_envelope(1, 0, helmet, vest, frame=threshold_frame),
|
|
)
|
|
|
|
assert result is not None
|
|
assert render_calls == [(threshold_frame, (helmet, vest))]
|
|
assert [item.envelope.payload.rule_id for item in result] == ["helmet", "vest"]
|
|
assert all(item.envelope.payload.image is encoded for item in result)
|
|
health = asyncio.run(operator.health())
|
|
assert "alert_images_encoded=1" in health.detail
|
|
assert "alert_images_attached=2" in health.detail
|
|
|
|
|
|
def test_missing_threshold_frame_is_fail_soft_and_commits_cooldown() -> None:
|
|
operator = _operator(
|
|
min_hits=1,
|
|
cooldown_ms=500,
|
|
alert_image={"enabled": True},
|
|
)
|
|
|
|
_, alert = _single_alert(_process(operator, _envelope(1, 0, _detection())))
|
|
assert alert.image is None
|
|
# The alert state is committed even though its evidence image is missing.
|
|
assert _process(operator, _envelope(1, 100, _detection())) is None
|
|
assert _process(operator, _envelope(2, 100, _detection())) is None
|
|
|
|
health = asyncio.run(operator.health())
|
|
assert health.healthy is False
|
|
assert "alerts_emitted=1" in health.detail
|
|
assert "duplicate_frames=1" in health.detail
|
|
assert "cooldown_suppressed=1" in health.detail
|
|
assert "alert_image_missing_frames=1" in health.detail
|
|
assert "alert_images_encoded=0" in health.detail
|
|
|
|
|
|
def test_renderer_failure_is_fail_soft_and_commits_cooldown(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
render_calls = 0
|
|
|
|
def fail_render(*args: Any, **kwargs: Any) -> EncodedImage:
|
|
nonlocal render_calls
|
|
render_calls += 1
|
|
raise ValueError("bad packed frame")
|
|
|
|
async def run_inline(
|
|
function: Any,
|
|
*args: Any,
|
|
executor: Any = None,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
del executor
|
|
return function(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(detection_rules, "render_annotated_jpeg", fail_render)
|
|
monkeypatch.setattr(detection_rules, "run_blocking", run_inline)
|
|
operator = _operator(
|
|
min_hits=1,
|
|
cooldown_ms=500,
|
|
alert_image={"enabled": True},
|
|
)
|
|
frame = _frame(5)
|
|
|
|
_, alert = _single_alert(
|
|
_process(operator, _envelope(1, 0, _detection(), frame=frame))
|
|
)
|
|
assert alert.image is None
|
|
assert render_calls == 1
|
|
assert _process(operator, _envelope(1, 100, _detection(), frame=frame)) is None
|
|
assert _process(operator, _envelope(2, 100, _detection(), frame=frame)) is None
|
|
|
|
health = asyncio.run(operator.health())
|
|
assert health.healthy is False
|
|
assert "alerts_emitted=1" in health.detail
|
|
assert "duplicate_frames=1" in health.detail
|
|
assert "cooldown_suppressed=1" in health.detail
|
|
assert "alert_image_failures=1" in health.detail
|
|
assert "alert_images_encoded=0" in health.detail
|
|
|
|
|
|
def test_pillow_dependency_is_checked_only_when_alert_images_are_enabled(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
checks = 0
|
|
|
|
def missing_pillow() -> None:
|
|
nonlocal checks
|
|
checks += 1
|
|
raise ImageAnnotationDependencyError("install the 'image' extra")
|
|
|
|
monkeypatch.setattr(detection_rules, "ensure_annotation_support", missing_pillow)
|
|
context = ComponentContext(
|
|
pipeline_id="detection",
|
|
node_id="trigger",
|
|
shutdown_event=asyncio.Event(),
|
|
)
|
|
|
|
asyncio.run(_operator(min_hits=1).setup(context))
|
|
assert checks == 0
|
|
|
|
enabled = _operator(
|
|
min_hits=1,
|
|
alert_image={"enabled": True},
|
|
)
|
|
with pytest.raises(ImageAnnotationDependencyError, match="image.*extra"):
|
|
asyncio.run(enabled.setup(context))
|
|
assert checks == 1
|
|
|
|
|
|
def test_time_source_is_explicit_and_captured_requires_a_timestamp() -> None:
|
|
captured = _operator(min_hits=2, window_ms=10, time_source="captured")
|
|
with pytest.raises(ValueError, match="requires captured_at_ns"):
|
|
_process(captured, _envelope(1, 0, _detection()))
|
|
|
|
auto = _operator(min_hits=2, window_ms=10, time_source="auto")
|
|
first = _envelope(1, 0, _detection(), captured=True)
|
|
second = Envelope(
|
|
payload=first.payload,
|
|
schema_name="DetectionResult",
|
|
source_id="camera-a",
|
|
sequence=2,
|
|
captured_at_ns=20 * NS_PER_MS,
|
|
received_at_ns=5 * NS_PER_MS,
|
|
)
|
|
assert _process(auto, first) is None
|
|
# auto prefers capture time, so the first hit is outside this 10 ms window.
|
|
assert _process(auto, second) is None
|
|
|
|
received = _operator(min_hits=2, window_ms=10, time_source="received")
|
|
assert _process(received, first) is None
|
|
_single_alert(_process(received, second))
|
|
|
|
|
|
def test_wrong_payload_and_out_of_order_frames_fail_closed() -> None:
|
|
operator = _operator(min_hits=2)
|
|
with pytest.raises(TypeError, match="expected DetectionResult"):
|
|
_process(
|
|
operator,
|
|
Envelope(
|
|
"not-a-detection",
|
|
source_id="camera-a",
|
|
sequence=1,
|
|
received_at_ns=0,
|
|
),
|
|
)
|
|
|
|
assert _process(operator, _envelope(1, 100, _detection())) is None
|
|
assert _process(operator, _envelope(2, 99, _detection())) is None
|
|
assert "stale_frames=1" in asyncio.run(operator.health()).detail
|