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" / "debug" / "detection_viewer.yaml"
DEMO_SCRIPT = PROJECT_ROOT / "detect_server" / "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()


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) -> DetectionResult:
    return DetectionResult(
        detections=(
            Detection(
                "No-Helmet",
                0.91,
                BoundingBox(0.0, 0.0, 1.0, 1.0),
            ),
        ),
        model_id="construction-ppe-yolov8@1",
        model_name="Construction PPE YOLOv8s",
        inference_ms=12.5,
        source_frame=_frame() if source_frame is None else source_frame,
    )


def _envelope(payload: Any) -> Envelope[Any]:
    return Envelope(
        payload,
        schema_name="DetectionResult",
        schema_version=1,
        source_id="wrist_cam",
        sequence=7,
    )


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


def test_demo_config_compiles_with_only_camera_decoder_detector_and_viewer() -> 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",
        "detector",
        "viewer",
    }
    detector = config.pipelines["detection_show"].nodes["detector"]
    assert detector.params["max_fps"] == 10
    assert detector.params["attach_frame"] is True
    viewer_edge = next(
        edge
        for edge in config.pipelines["detection_show"].edges
        if edge.target == "viewer.input"
    )
    assert viewer_edge.source == "detector.detections"
    assert viewer_edge.qos.profile == "realtime_latest"
    assert viewer_edge.qos.capacity == 1
    assert viewer_edge.qos.overflow == "drop_oldest"


@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"),
    ],
)
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_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