cmvr_edge_ai/tests/unit/test_detection_viewer_demo.py

from __future__ import annotations

import asyncio
import importlib.util
from pathlib import Path
import sys
from typing import Any

import pytest

from cmvr_edge_ai.application import validate_application
from cmvr_edge_ai.config import load_config
from cmvr_edge_ai.contracts import (
    BoundingBox,
    Detection,
    DetectionResult,
    ImageFrame,
)
from cmvr_edge_ai.core import ComponentContext, Envelope

PROJECT_ROOT = Path(__file__).resolve().parents[2]
DEMO_CONFIG = PROJECT_ROOT / "configs" / "detection_viewer.yaml"
DEMO_SCRIPT = PROJECT_ROOT / "server" / "detect" / "show_detections.py"


def _load_demo_module() -> Any:
    spec = importlib.util.spec_from_file_location(
        "cmvr_edge_ai_detection_viewer_demo",
        DEMO_SCRIPT,
    )
    if spec is None or spec.loader is None:
        raise RuntimeError(f"could not load demo module: {DEMO_SCRIPT}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


demo = _load_demo_module()


@pytest.fixture(autouse=True)
def _render_inline(monkeypatch: pytest.MonkeyPatch) -> None:
    """Keep these pure viewer tests independent of native worker threads."""

    async def run_inline(
        function: Any,
        *args: Any,
        executor: Any = None,
        **kwargs: Any,
    ) -> Any:
        del executor
        return function(*args, **kwargs)

    monkeypatch.setattr(demo, "run_blocking", run_inline)


class _FakeCv2:
    WINDOW_NORMAL = 0
    WND_PROP_VISIBLE = 1
    FONT_HERSHEY_SIMPLEX = 2
    LINE_AA = 3
    COLOR_RGB2BGR = 4

    def __init__(self, *, key: int = -1, visible: float = 1.0, gui: str = "QT5") -> None:
        self.key = key
        self.visible = visible
        self.gui = gui
        self.named: list[tuple[str, int]] = []
        self.resized: list[tuple[str, int, int]] = []
        self.shown: list[tuple[str, Any]] = []
        self.destroyed: list[str] = []

    def getBuildInformation(self) -> str:
        return f"OpenCV test build\n  GUI: {self.gui}\n"

    def namedWindow(self, name: str, mode: int) -> None:
        self.named.append((name, mode))

    def resizeWindow(self, name: str, width: int, height: int) -> None:
        self.resized.append((name, width, height))

    def imshow(self, name: str, image: Any) -> None:
        self.shown.append((name, image))

    def waitKey(self, delay: int) -> int:
        del delay
        return self.key

    def getWindowProperty(self, name: str, prop: int) -> float:
        del name, prop
        return self.visible

    def destroyWindow(self, name: str) -> None:
        self.destroyed.append(name)


def _frame(
    *,
    pixel_format: str = "BGR8",
    codec: str = "none",
    data: bytes | None = None,
) -> ImageFrame:
    return ImageFrame(
        data=bytes((10, 20, 30)) * 4 if data is None else data,
        width=2,
        height=2,
        pixel_format=pixel_format,
        codec=codec,
    )


def _result(
    *,
    source_frame: ImageFrame | None = None,
    label: str = "No-Helmet",
    model_id: str = "construction-ppe-yolov8@1",
    model_name: str = "Construction PPE YOLOv8s",
    inference_ms: float = 12.5,
) -> DetectionResult:
    return DetectionResult(
        detections=(
            Detection(
                label,
                0.91,
                BoundingBox(0.0, 0.0, 1.0, 1.0),
            ),
        ),
        model_id=model_id,
        model_name=model_name,
        inference_ms=inference_ms,
        source_frame=_frame() if source_frame is None else source_frame,
    )


def _envelope(
    payload: Any,
    *,
    sequence: int = 7,
    trace_id: str = "viewer-test-frame",
) -> Envelope[Any]:
    return Envelope(
        payload,
        schema_name="DetectionResult",
        schema_version=1,
        source_id="wrist_cam",
        sequence=sequence,
        trace_id=trace_id,
        session_id="viewer-test-session",
    )


def _context() -> ComponentContext:
    return ComponentContext(
        pipeline_id="detection_show",
        node_id="viewer",
        shutdown_event=asyncio.Event(),
        metadata={},
    )


def test_demo_config_compiles_with_shared_fanout_and_two_detectors() -> None:
    config = load_config(DEMO_CONFIG)
    compiled = validate_application(
        config,
        demo._build_registry(lambda: None),
        ("detection_show",),
    )

    assert len(compiled) == 1
    assert set(compiled[0].plugin_specs) == {
        "camera",
        "decoder",
        "frame_fanout",
        "detector",
        "phone_detector",
        "viewer",
    }
    assert config.runtime.thread_workers == 4
    detector = config.pipelines["detection_show"].nodes["detector"]
    assert detector.params["attach_frame"] is True
    assert "max_fps" not in detector.params
    phone_detector = config.pipelines["detection_show"].nodes["phone_detector"]
    assert phone_detector.params["model"] == "yolov8n-mobile-phone@2"
    assert phone_detector.params["detect_labels"] == ["mobile_phone"]
    assert phone_detector.params["attach_frame"] is True
    assert "max_fps" not in phone_detector.params
    viewer = config.pipelines["detection_show"].nodes["viewer"]
    assert viewer.params["expected_inputs"] == ["ppe", "phone"]

    edges = {
        (edge.source, edge.target): edge
        for edge in config.pipelines["detection_show"].edges
    }
    latest = edges[("decoder.frames", "frame_fanout.input")]
    assert latest.qos.profile == "realtime_latest"
    assert latest.qos.capacity == 1
    assert latest.qos.overflow == "drop_oldest"
    for target in ("detector.frames", "phone_detector.frames"):
        fanout = edges[("frame_fanout.output", target)]
        assert fanout.qos.profile == "telemetry"
        assert fanout.qos.capacity == 1
        assert fanout.qos.overflow == "block"
    assert edges[("detector.detections", "viewer.ppe")].qos.overflow == "block"
    assert (
        edges[("phone_detector.detections", "viewer.phone")].qos.overflow
        == "block"
    )


@pytest.mark.parametrize(
    ("params", "message"),
    [
        ({"unknown": 1}, "unknown detection viewer parameter"),
        ({"window_name": ""}, "window_name must be a non-empty"),
        ({"window_width": True}, "window_width must be an integer"),
        ({"window_height": 0}, "window_height must be between"),
        ({"wait_key_ms": 0}, "wait_key_ms must be between"),
        ({"wait_key_ms": 21}, "wait_key_ms must be between"),
        ({"box_thickness": 21}, "box_thickness must be between"),
        ({"font_scale": float("nan")}, "font_scale must be between"),
        ({"show_stats": "true"}, "show_stats must be a boolean"),
        ({"expected_inputs": "ppe"}, "expected_inputs must be a list"),
        ({"expected_inputs": []}, "must contain at least one"),
        ({"expected_inputs": ["ppe", "ppe"]}, "must not contain duplicate"),
        ({"expected_inputs": ["other"]}, "unsupported viewer port"),
        ({"max_pending_frames": 0}, "max_pending_frames must be between"),
    ],
)
def test_viewer_parameters_are_strict(params: dict[str, Any], message: str) -> None:
    with pytest.raises(ValueError, match=message):
        demo.OpenCvDetectionViewerSink(
            "viewer",
            params,
            request_stop=lambda: None,
        )


def test_box_clipping_skips_invalid_or_fully_outside_boxes() -> None:
    valid = Detection("valid", 0.9, BoundingBox(-2.2, 1.2, 20.0, 9.8))
    reversed_box = Detection("bad", 0.9, BoundingBox(8.0, 8.0, 2.0, 2.0))
    outside = Detection("outside", 0.9, BoundingBox(12.0, 2.0, 20.0, 5.0))
    not_finite = Detection("nan", 0.9, BoundingBox(float("nan"), 1, 2, 3))

    assert demo._clipped_box(valid, 10, 8) == (0, 1, 9, 7)
    assert demo._clipped_box(reversed_box, 10, 8) is None
    assert demo._clipped_box(outside, 10, 8) is None
    assert demo._clipped_box(not_finite, 10, 8) is None


@pytest.mark.parametrize(
    ("frame", "message"),
    [
        (_frame(codec="H265"), "requires a decoded"),
        (_frame(pixel_format="GRAY8"), "supports BGR8 or RGB8"),
        (_frame(data=b"too-short"), "packed frame has"),
        (
            ImageFrame(b"", 0, 2, "BGR8"),
            "dimensions must be positive integers",
        ),
        (
            ImageFrame(b"", 2, 2, "BGR8", codec=None),  # type: ignore[arg-type]
            "codec must be a string",
        ),
    ],
)
def test_display_frame_validation_is_actionable(
    frame: ImageFrame,
    message: str,
) -> None:
    with pytest.raises(demo.DetectionViewerError, match=message):
        demo._validate_display_frame(frame)


def test_viewer_rejects_missing_source_frame(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_cv2 = _FakeCv2()
    stop_requested = False

    def request_stop() -> None:
        nonlocal stop_requested
        stop_requested = True

    def load_module(name: str) -> Any:
        return fake_cv2 if name == "cv2" else object()

    async def scenario() -> None:
        viewer = demo.OpenCvDetectionViewerSink(
            "viewer",
            {},
            request_stop=request_stop,
            module_loader=load_module,
        )
        await viewer.setup(_context())
        result = DetectionResult((), "model@1", 1.0, source_frame=None)
        with pytest.raises(demo.DetectionViewerError, match="attach_frame: true"):
            await viewer.consume(_envelope(result))
        await viewer.stop()

    monkeypatch.setenv("DISPLAY", ":99")
    asyncio.run(scenario())
    assert stop_requested is False


@pytest.mark.parametrize("key", [ord("q"), ord("Q"), 27])
def test_viewer_quit_keys_work_before_any_detection_frame(
    key: int,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_cv2 = _FakeCv2(key=key)
    stop_calls = 0
    stop_event: asyncio.Event | None = None

    def request_stop() -> None:
        nonlocal stop_calls, stop_event
        stop_calls += 1
        assert stop_event is not None
        stop_event.set()

    def load_module(name: str) -> Any:
        return fake_cv2 if name == "cv2" else object()

    monkeypatch.setenv("DISPLAY", ":99")

    async def scenario() -> None:
        nonlocal stop_event
        stop_event = asyncio.Event()
        viewer = demo.OpenCvDetectionViewerSink(
            "viewer",
            {},
            request_stop=request_stop,
            module_loader=load_module,
        )
        await viewer.setup(_context())
        await viewer.start()
        await asyncio.wait_for(stop_event.wait(), timeout=0.2)
        await viewer.stop()
        await viewer.stop()

    asyncio.run(scenario())
    assert stop_calls == 1
    assert fake_cv2.shown == []
    assert fake_cv2.destroyed == ["CMVR PPE Detection"]


def test_viewer_window_close_requests_outer_application_stop(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_cv2 = _FakeCv2(visible=0.0)
    stop_calls = 0
    stop_event: asyncio.Event | None = None

    def request_stop() -> None:
        nonlocal stop_calls, stop_event
        stop_calls += 1
        assert stop_event is not None
        stop_event.set()

    monkeypatch.setenv("DISPLAY", ":99")

    async def scenario() -> None:
        nonlocal stop_event
        stop_event = asyncio.Event()
        viewer = demo.OpenCvDetectionViewerSink(
            "viewer",
            {},
            request_stop=request_stop,
            module_loader=lambda name: fake_cv2 if name == "cv2" else object(),
        )
        await viewer.setup(_context())
        await viewer.start()
        await asyncio.wait_for(stop_event.wait(), timeout=0.2)
        await viewer.stop()

    asyncio.run(scenario())
    assert stop_calls == 1


def test_unknown_visibility_does_not_close_window(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_cv2 = _FakeCv2(visible=-1.0)
    stop_calls = 0

    def request_stop() -> None:
        nonlocal stop_calls
        stop_calls += 1

    monkeypatch.setenv("DISPLAY", ":99")

    async def scenario() -> None:
        viewer = demo.OpenCvDetectionViewerSink(
            "viewer",
            {},
            request_stop=request_stop,
            module_loader=lambda name: fake_cv2 if name == "cv2" else object(),
        )
        await viewer.setup(_context())
        await viewer.start()
        await asyncio.sleep(0.03)
        await viewer.stop()

    asyncio.run(scenario())
    assert stop_calls == 0


def test_consume_displays_rendered_detection_frame(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_cv2 = _FakeCv2()
    monkeypatch.setattr(
        demo,
        "_render_detection_frame",
        lambda *args, **kwargs: "frame",
    )
    monkeypatch.setenv("DISPLAY", ":99")

    async def scenario() -> None:
        viewer = demo.OpenCvDetectionViewerSink(
            "viewer",
            {},
            request_stop=lambda: None,
            module_loader=lambda name: fake_cv2 if name == "cv2" else object(),
        )
        await viewer.setup(_context())
        await viewer.start()
        await viewer.consume(_envelope(_result()))
        await viewer.stop()

    asyncio.run(scenario())
    assert fake_cv2.shown == [("CMVR PPE Detection", "frame")]


def test_viewer_pairs_two_models_and_renders_one_combined_frame(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_cv2 = _FakeCv2()
    render_calls: list[tuple[ImageFrame, tuple[Detection, ...], str | None]] = []

    def render(
        frame: ImageFrame,
        detections: tuple[Detection, ...],
        **kwargs: Any,
    ) -> str:
        render_calls.append((frame, detections, kwargs["header"]))
        return "combined-frame"

    monkeypatch.setattr(demo, "_render_detection_frame", render)
    monkeypatch.setenv("DISPLAY", ":99")
    frame = _frame()
    ppe = _result(source_frame=frame)
    phone = _result(
        source_frame=frame,
        label="mobile_phone",
        model_id="yolov8n-mobile-phone@1",
        model_name="YOLOv8n Mobile Phone",
        inference_ms=87.5,
    )

    async def scenario() -> None:
        viewer = demo.OpenCvDetectionViewerSink(
            "viewer",
            {"expected_inputs": ["ppe", "phone"]},
            request_stop=lambda: None,
            module_loader=lambda name: fake_cv2 if name == "cv2" else object(),
        )
        await viewer.setup(_context())
        await viewer.start()
        # Arrival order must not change the configured overlay/header order.
        await viewer.consume(_envelope(phone), input_port="phone")
        assert fake_cv2.shown == []
        await viewer.consume(_envelope(ppe), input_port="ppe")
        await viewer.stop()

    asyncio.run(scenario())

    assert fake_cv2.shown == [
        ("CMVR PPE Detection", "combined-frame"),
    ]
    assert len(render_calls) == 1
    rendered_frame, detections, header = render_calls[0]
    assert rendered_frame is frame
    assert tuple(item.label for item in detections) == (
        "PPE:No-Helmet",
        "PHONE:mobile_phone",
    )
    assert header is not None
    assert "PPE boxes=1 12.5ms" in header
    assert "PHONE boxes=1 87.5ms" in header


def test_viewer_never_pairs_different_source_frame_objects(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_cv2 = _FakeCv2()
    monkeypatch.setattr(demo, "_render_detection_frame", lambda *args, **kwargs: None)
    monkeypatch.setenv("DISPLAY", ":99")

    async def scenario() -> None:
        viewer = demo.OpenCvDetectionViewerSink(
            "viewer",
            {"expected_inputs": ["ppe", "phone"]},
            request_stop=lambda: None,
            module_loader=lambda name: fake_cv2 if name == "cv2" else object(),
        )
        await viewer.setup(_context())
        await viewer.consume(_envelope(_result()), input_port="ppe")
        await viewer.consume(
            _envelope(
                _result(
                    label="mobile_phone",
                    model_id="yolov8n-mobile-phone@1",
                )
            ),
            input_port="phone",
        )
        assert len(viewer._pending_results) == 2
        await viewer.stop()

    asyncio.run(scenario())
    assert fake_cv2.shown == []


def test_viewer_never_pairs_different_traces_for_same_source_frame(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_cv2 = _FakeCv2()
    monkeypatch.setattr(demo, "_render_detection_frame", lambda *args, **kwargs: None)
    monkeypatch.setenv("DISPLAY", ":99")
    frame = _frame()

    async def scenario() -> None:
        viewer = demo.OpenCvDetectionViewerSink(
            "viewer",
            {"expected_inputs": ["ppe", "phone"]},
            request_stop=lambda: None,
            module_loader=lambda name: fake_cv2 if name == "cv2" else object(),
        )
        await viewer.setup(_context())
        await viewer.consume(
            _envelope(_result(source_frame=frame), trace_id="ppe-trace"),
            input_port="ppe",
        )
        await viewer.consume(
            _envelope(
                _result(
                    source_frame=frame,
                    label="mobile_phone",
                    model_id="yolov8n-mobile-phone@1",
                ),
                trace_id="phone-trace",
            ),
            input_port="phone",
        )
        assert len(viewer._pending_results) == 2
        await viewer.stop()

    asyncio.run(scenario())
    assert fake_cv2.shown == []


def test_viewer_pending_frame_cache_is_bounded(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_cv2 = _FakeCv2()
    monkeypatch.setenv("DISPLAY", ":99")

    async def scenario() -> None:
        viewer = demo.OpenCvDetectionViewerSink(
            "viewer",
            {
                "expected_inputs": ["ppe", "phone"],
                "max_pending_frames": 2,
            },
            request_stop=lambda: None,
            module_loader=lambda name: fake_cv2 if name == "cv2" else object(),
        )
        await viewer.setup(_context())
        for sequence in range(3):
            await viewer.consume(
                _envelope(
                    _result(),
                    sequence=sequence,
                    trace_id=f"frame-{sequence}",
                ),
                input_port="ppe",
            )
        assert len(viewer._pending_results) == 2
        assert viewer._pending_evicted == 1
        await viewer.stop()

    asyncio.run(scenario())


def test_linux_headless_session_fails_before_loading_opencv(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    modules_loaded: list[str] = []
    viewer = demo.OpenCvDetectionViewerSink(
        "viewer",
        {},
        request_stop=lambda: None,
        module_loader=lambda name: modules_loaded.append(name),
    )
    monkeypatch.setattr(demo.sys, "platform", "linux")
    monkeypatch.delenv("DISPLAY", raising=False)
    monkeypatch.delenv("WAYLAND_DISPLAY", raising=False)

    with pytest.raises(demo.DetectionViewerError, match="DISPLAY"):
        asyncio.run(viewer.setup(_context()))
    assert modules_loaded == []


def test_headless_opencv_build_is_rejected(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_cv2 = _FakeCv2(gui="NONE")
    viewer = demo.OpenCvDetectionViewerSink(
        "viewer",
        {},
        request_stop=lambda: None,
        module_loader=lambda name: fake_cv2 if name == "cv2" else object(),
    )
    monkeypatch.setenv("DISPLAY", ":99")

    with pytest.raises(demo.DetectionViewerError, match="no GUI backend"):
        asyncio.run(viewer.setup(_context()))
    assert fake_cv2.named == []


def test_render_preserves_bgr_and_converts_rgb() -> None:
    numpy = pytest.importorskip("numpy")

    class DrawingCv2(_FakeCv2):
        def __init__(self) -> None:
            super().__init__()
            self.operations: list[tuple[Any, ...]] = []

        def cvtColor(self, image: Any, code: int) -> Any:
            assert code == self.COLOR_RGB2BGR
            return image[:, :, ::-1].copy()

        def rectangle(self, *args: Any, **kwargs: Any) -> None:
            del kwargs
            _, top_left, bottom_right, _, thickness = args
            self.operations.append(
                ("rectangle", top_left, bottom_right, thickness)
            )

        def getTextSize(self, *args: Any, **kwargs: Any) -> tuple[tuple[int, int], int]:
            del args, kwargs
            return (10, 5), 1

        def putText(self, *args: Any, **kwargs: Any) -> None:
            del kwargs
            self.operations.append(("text", args[1]))

    cv2 = DrawingCv2()
    bgr = _frame(pixel_format="BGR8")
    rgb = _frame(pixel_format="RGB8")
    kwargs = {
        "cv2_module": cv2,
        "numpy_module": numpy,
        "box_thickness": 2,
        "font_scale": 0.6,
    }

    bgr_image = demo._render_detection_frame(
        bgr,
        _result().detections,
        header="stats",
        **kwargs,
    )
    rgb_image = demo._render_detection_frame(rgb, (), header=None, **kwargs)

    assert tuple(int(value) for value in bgr_image[0, 0]) == (10, 20, 30)
    assert tuple(int(value) for value in rgb_image[0, 0]) == (30, 20, 10)
    assert bgr.data == bytes((10, 20, 30)) * 4
    assert cv2.operations[:2] == [
        ("rectangle", (0, 0), (1, 1), -1),
        ("text", "stats"),
    ]
    assert ("rectangle", (0, 0), (1, 1), 2) in cv2.operations[2:]
    assert ("text", "No-Helmet 0.91") in cv2.operations[2:]


def test_run_demo_viewer_callback_stops_outer_application(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    applications: list[Any] = []

    class FakeApplication:
        def __init__(self, config: Any, registry: Any) -> None:
            del config
            self.registry = registry
            self.started_with: tuple[str, ...] | None = None
            self.stop_calls = 0
            applications.append(self)

        async def start(self, pipeline_ids: tuple[str, ...]) -> None:
            self.started_with = pipeline_ids
            viewer = self.registry.resolve(demo._VIEWER_PLUGIN_ID).factory(
                "viewer",
                {},
            )
            viewer._request_demo_stop("test")

        async def wait(self) -> None:
            await asyncio.Event().wait()

        async def stop(self, *, graceful: bool) -> None:
            assert graceful is True
            self.stop_calls += 1

    monkeypatch.setattr(demo, "EdgeAIApplication", FakeApplication)
    config = load_config(DEMO_CONFIG)

    assert asyncio.run(demo._run_demo(config, "detection_show")) == 0
    assert len(applications) == 1
    assert applications[0].started_with == ("detection_show",)
    assert applications[0].stop_calls >= 1