"""Unit tests for orisys.capture with fake VideoCapture objects (no hardware)."""

from __future__ import annotations

import threading
import time

import cv2
import numpy as np
import pytest

from orisys.capture import (
    MAX_SIMULTANEOUS_CAPTURES,
    CameraFilter,
    CameraInfo,
    CaptureConfig,
    CaptureHealth,
    CaptureSource,
    ReconnectPolicy,
    list_cameras,
    normalize_source,
    open_capture,
    pick_default_camera,
)
from orisys.capture.discovery import DEFAULT_LAPTOP_WEBCAM_PATTERNS
from orisys.capture.source import _apply_stream_props, _warmup_capture

MJPG = cv2.VideoWriter_fourcc(*"MJPG")
YUY2 = cv2.VideoWriter_fourcc(*"YUY2")


class FakeCap:
    """Deterministic stand-in for cv2.VideoCapture."""

    def __init__(
        self,
        frames=None,
        *,
        fail_after=None,
        fail_reads=0,
        open_ok=True,
        reset_fourcc_on_resize=False,
    ):
        self._frames = list(frames or [])
        self._idx = 0
        self._open = bool(open_ok)
        self._fail_after = fail_after
        self._fail_reads_remaining = int(fail_reads)
        self.read_count = 0
        self.grab_count = 0
        self.release_count = 0
        self.width = 4
        self.height = 3
        self.fps = 30.0
        self.fourcc = 0
        self.reset_fourcc_on_resize = reset_fourcc_on_resize
        self.set_calls: list[tuple[int, float]] = []

    def isOpened(self):
        return self._open

    def read(self):
        self.read_count += 1
        if not self._open:
            return False, None
        if self._fail_reads_remaining > 0:
            self._fail_reads_remaining -= 1
            return False, None
        if self._fail_after is not None and self.read_count > self._fail_after:
            self._open = False
            return False, None
        if self._idx >= len(self._frames):
            if not self._frames:
                return False, None
            frame = self._frames[-1]
        else:
            frame = self._frames[self._idx]
            self._idx += 1
        return True, frame.copy()

    def grab(self):
        self.grab_count += 1
        if not self._open:
            return False
        ok, _ = self.read()
        return ok

    def release(self):
        self.release_count += 1
        self._open = False

    def get(self, prop_id):
        if prop_id in (3,):  # CAP_PROP_FRAME_WIDTH
            return float(self.width)
        if prop_id in (4,):  # CAP_PROP_FRAME_HEIGHT
            return float(self.height)
        if prop_id in (5,):  # CAP_PROP_FPS
            return float(self.fps)
        if prop_id in (6,):  # CAP_PROP_FOURCC
            return float(self.fourcc)
        return 0.0

    def set(self, prop_id, value):
        self.set_calls.append((prop_id, float(value)))
        if prop_id in (3,):  # CAP_PROP_FRAME_WIDTH
            self.width = int(value)
            if self.reset_fourcc_on_resize:
                self.fourcc = YUY2
        if prop_id in (4,):  # CAP_PROP_FRAME_HEIGHT
            self.height = int(value)
            if self.reset_fourcc_on_resize:
                self.fourcc = YUY2
        if prop_id in (5,):  # CAP_PROP_FPS
            self.fps = float(value)
        if prop_id in (6,):  # CAP_PROP_FOURCC
            self.fourcc = int(value)
        return True


def _frame(val: int) -> np.ndarray:
    return np.full((3, 4, 3), val, dtype=np.uint8)


def _factory_from_caps(caps):
    lock = threading.Lock()
    state = {"i": 0}

    def factory(_config):
        with lock:
            if state["i"] >= len(caps):
                raise RuntimeError("No more fake captures")
            cap = caps[state["i"]]
            state["i"] += 1
            return cap, "FAKE"

    return factory


@pytest.mark.parametrize(
    "video,expected",
    [
        (None, 0),
        ("0", 0),
        (" 1 ", 1),
        (2, 2),
        ("/dev/video0", "/dev/video0"),
        (r".\clip.mp4" + "\\", r".\clip.mp4"),
    ],
)
def test_normalize_source(video, expected):
    assert normalize_source(video) == expected


def test_capture_config_from_camera_dict_ignores_deprecated_kwargs():
    cfg = CaptureConfig.from_camera_dict(
        0,
        {"resolution_raw": [1920, 1080], "fps": 30},
        mode="queue",
        queue_size=5,
        overflow="drop_newest",
        start_worker=False,
    )
    assert cfg.width == 1920
    assert cfg.height == 1080
    assert cfg.background is False


def test_apply_stream_props_sets_size_before_fourcc():
    cap = FakeCap([_frame(1)], reset_fourcc_on_resize=True)
    _apply_stream_props(cap, width=1920, height=1080, fps=30, fourcc="MJPG")
    assert cap.fourcc == MJPG
    width_idx = [p for p, _ in cap.set_calls if p == 3]
    fourcc_calls = [v for p, v in cap.set_calls if p == 6]
    assert width_idx
    assert fourcc_calls
    assert cap.set_calls.index((3, 1920.0)) < cap.set_calls.index((6, float(MJPG)))


def test_warmup_capture_grabs():
    cap = FakeCap([_frame(1)])
    _warmup_capture(cap, grabs=3)
    assert cap.grab_count == 3


def test_capture_source_read_and_status():
    frames = [_frame(7)]
    cap = FakeCap(frames)
    src = CaptureSource(
        CaptureConfig(source=0, reconnect=ReconnectPolicy(enabled=False)),
        factory=lambda cfg: (cap, "FAKE"),
    )
    src.open()
    assert src.is_open()
    assert src.status.health == CaptureHealth.CONNECTED
    ok, frame = src.read_raw()
    assert ok and int(frame[0, 0, 0]) == 7
    src.release()
    assert src.status.health == CaptureHealth.CLOSED
    assert cap.release_count == 1


def test_auto_reconnect_after_failed_reads():
    first = FakeCap([_frame(1)], fail_after=1)
    second = FakeCap([_frame(9)])
    factory = _factory_from_caps([first, second])
    policy = ReconnectPolicy(
        enabled=True,
        fail_threshold=1,
        initial_backoff_s=0.01,
        max_backoff_s=0.01,
        stale_timeout_s=None,
        max_attempts=3,
    )
    src = CaptureSource(
        CaptureConfig(source=0, reconnect=policy, verbose=False),
        factory=factory,
    )
    src.open()
    ok, _ = src.read_raw()
    assert ok
    ok, _ = src.read_raw()
    assert not ok
    assert src.should_reconnect()
    assert src.reconnect()
    ok, frame = src.read_raw()
    assert ok and int(frame[0, 0, 0]) == 9
    assert src.status.health == CaptureHealth.CONNECTED
    assert src.status.reconnect_attempts >= 1


def test_reconnect_disabled_does_not_reopen():
    cap = FakeCap([_frame(1)], fail_after=1)
    src = CaptureSource(
        CaptureConfig(
            source=0,
            reconnect=ReconnectPolicy(enabled=False, fail_threshold=1),
        ),
        factory=lambda cfg: (cap, "FAKE"),
    )
    src.open()
    src.read_raw()
    src.read_raw()
    assert not src.should_reconnect()


def test_reconnect_attempts_exhausted():
    good = FakeCap([_frame(1)], fail_after=1)
    state = {"n": 0}

    def factory(_cfg):
        state["n"] += 1
        if state["n"] == 1:
            return good, "FAKE"
        raise RuntimeError("camera missing")

    policy = ReconnectPolicy(
        enabled=True,
        fail_threshold=1,
        initial_backoff_s=0.01,
        max_backoff_s=0.01,
        stale_timeout_s=None,
        max_attempts=2,
    )
    src = CaptureSource(
        CaptureConfig(source=0, reconnect=policy),
        factory=factory,
    )
    src.open()
    src.read_raw()
    src.read_raw()
    assert src.should_reconnect()
    assert not src.reconnect()
    assert not src.reconnect()
    assert src.status.health == CaptureHealth.FAILED


def test_background_thread_publishes_latest_frame():
    caps = [FakeCap([_frame(i) for i in range(5)])]
    src = open_capture(
        CaptureConfig(
            source=0,
            background=True,
            reconnect=ReconnectPolicy(enabled=False),
        ),
        factory=_factory_from_caps(caps),
        start_background=True,
    )
    try:
        deadline = time.perf_counter() + 2.0
        ok = False
        frame = None
        while time.perf_counter() < deadline:
            ok, frame = src.read(wait_s=0.05)
            if ok:
                break
        assert ok and frame is not None
        assert src.running
    finally:
        src.close()


def test_background_reconnect():
    first = FakeCap([_frame(1)], fail_after=2)
    second = FakeCap([_frame(42) for _ in range(20)])
    policy = ReconnectPolicy(
        enabled=True,
        fail_threshold=1,
        initial_backoff_s=0.01,
        max_backoff_s=0.02,
        stale_timeout_s=None,
    )
    src = open_capture(
        CaptureConfig(source=0, reconnect=policy, background=True),
        factory=_factory_from_caps([first, second]),
        start_background=True,
    )
    try:
        saw_reconnect = False
        saw_new = False
        deadline = time.perf_counter() + 3.0
        while time.perf_counter() < deadline:
            if src.status.reconnect_attempts > 0:
                saw_reconnect = True
            ok, frame = src.read(wait_s=0.05)
            if ok and frame is not None and int(frame[0, 0, 0]) == 42:
                saw_new = True
                break
            time.sleep(0.01)
        assert saw_reconnect
        assert saw_new
    finally:
        src.close()


def test_warmup_frames_discards_reads():
    cap = FakeCap([_frame(i) for i in range(10)])
    src = CaptureSource(
        CaptureConfig(source=0),
        factory=lambda cfg: (cap, "FAKE"),
    )
    src.open()
    src.warmup_frames(3)
    ok, frame = src.read_raw()
    assert ok and int(frame[0, 0, 0]) == 3


def test_camera_filter_excludes_laptop_webcams():
    cams = [
        CameraInfo(0, "DSHOW", 0, 640, 480, name="Integrated Webcam"),
        CameraInfo(1, "DSHOW", 0, 1920, 1080, name="Orisys Tactile Cam"),
        CameraInfo(2, "DSHOW", 0, 1280, 720, name="USB Camera"),
    ]
    filt = CameraFilter(exclude_laptop_webcams=True)
    kept = [c for c in cams if filt.matches(c)]
    assert [c.index for c in kept] == [1, 2]


def test_camera_filter_name_include_and_vid_pid():
    cams = [
        CameraInfo(0, "V4L2", 0, 640, 480, name="foo", vendor_id=0x1234, product_id=0x5678),
        CameraInfo(1, "V4L2", 0, 640, 480, name="orisys-finger", vendor_id=0xAAAA, product_id=0xBBBB),
    ]
    filt = CameraFilter(name_include=[r"orisys"], vid_pid_allowlist=[(0xAAAA, 0xBBBB)])
    kept = [c for c in cams if filt.matches(c)]
    assert len(kept) == 1 and kept[0].index == 1


def test_camera_filter_min_resolution():
    cams = [
        CameraInfo(0, "DSHOW", 0, 320, 240, name="tiny"),
        CameraInfo(1, "DSHOW", 0, 1280, 720, name="hd"),
    ]
    filt = CameraFilter(min_width=640, min_height=480)
    kept = [c for c in cams if filt.matches(c)]
    assert [c.index for c in kept] == [1]


def test_pick_default_camera():
    cams = [
        CameraInfo(1, "DSHOW", 0, 640, 480),
        CameraInfo(2, "DSHOW", 0, 640, 480),
    ]
    assert pick_default_camera(cams, preferred="2") == "2"
    assert pick_default_camera(cams, preferred=None) == "1"
    assert pick_default_camera([], preferred="0") is None


def test_laptop_patterns_nonempty():
    assert len(DEFAULT_LAPTOP_WEBCAM_PATTERNS) >= 5


def test_camera_info_label_includes_name():
    cam = CameraInfo(3, "DSHOW", 700, 1280, 720, name="Orisys")
    assert "Orisys" in cam.label()
    assert cam.source == "3"


def test_list_cameras_is_callable():
    assert callable(list_cameras)


def test_max_simultaneous_captures_constant():
    assert MAX_SIMULTANEOUS_CAPTURES == 10


def test_sensor_external_mode_still_works():
    from orisys import Sensor

    sensor = Sensor("external", config_name="box", verbose=False, backend="cpu")
    try:
        assert sensor._external_only
        assert sensor.capture_status is None
        assert sensor._capture_source is None
    finally:
        sensor.disconnect()


def test_capture_source_open_applies_stream_props_via_default_factory(monkeypatch):
    cap = FakeCap([_frame(1)], reset_fourcc_on_resize=True)

    def fake_open(*args, **kwargs):
        _apply_stream_props(
            cap,
            width=kwargs.get("width"),
            height=kwargs.get("height"),
            fps=kwargs.get("fps"),
            fourcc=kwargs.get("fourcc"),
        )
        _warmup_capture(cap)
        return cap, "FAKE"

    monkeypatch.setattr("orisys.capture.source.open_opencv_capture", fake_open)
    src = CaptureSource(CaptureConfig(source=0, width=1920, height=1080, fourcc="MJPG"))
    src.open()
    assert cap.fourcc == MJPG