"""Which frame the acquisition loop counts, and at what rate.

``CaptureSource.read_latest`` returns the newest frame from its background reader
and, if that reader has not produced another one yet, the *same* frame again. The
loop must not treat the repeat as a new sample: doing so duplicated every
measurement (a 30 fps device recorded 63 rows per second) and the rail read the
loop's own polling rate as the device's frame rate.
"""

from __future__ import annotations

import sys
import threading
import time
import types
from pathlib import Path

APP_ROOT = Path(__file__).resolve().parents[2]
BOX_CONFIG = APP_ROOT / "apps" / "studio" / "config" / "sensors" / "box" / "sensor.json"

#: The fake device's own frame period: 30 frames per second, the rate the studio
#: sensor configs declare.
FRAME_PERIOD_S = 1.0 / 30.0
#: How many times the SDK hands the same frame back before the next one arrives.
REPEATS_PER_FRAME = 3


class _Status:
    """Stands in for ``orisys.capture.config.CaptureStatus``."""

    def __init__(self) -> None:
        self.last_seq = 0
        self.dropped_frames = 0
        #: What the driver reported back for the negotiated stream.
        self.width = 1920
        self.height = 1080
        self.fps = 30.0
        self.fourcc = "MJPG"
        self.backend_name = "MSMF"


class _CaptureConfig:
    """The requested stream, as ``CaptureConfig`` carries it."""

    def __init__(self) -> None:
        self.width = 1920
        self.height = 1080
        self.fps = 30.0
        self.fourcc = "MJPG"
        self.source = 0


class _CaptureSource:
    """A background reader that publishes a new frame on its own thread.

    The real one is a thread too: the sequence advances whether or not the
    consumer asks for a frame, which is exactly why an unchanged sequence means
    "no new frame" rather than "the consumer has not called yet". It also counts
    a drop whenever it replaces a frame nobody took, like
    ``orisys.capture.source.CaptureSource._background_loop`` does.
    """

    def __init__(self, period_s: float = FRAME_PERIOD_S) -> None:
        self.status = _Status()
        self.config = _CaptureConfig()
        self.running = True
        self._period = float(period_s)
        self._stop = threading.Event()
        self._taken_seq = 0
        self._thread = threading.Thread(target=self._publish, daemon=True)
        self._thread.start()

    def _publish(self) -> None:
        while not self._stop.is_set():
            if self._stop.wait(timeout=self._period):
                return
            # Replacing a frame that was never consumed is a drop on our side.
            if self.status.last_seq > self._taken_seq:
                self.status.dropped_frames += 1
            self.status.last_seq += 1

    def take(self) -> None:
        self._taken_seq = self.status.last_seq

    def close(self) -> None:
        self.running = False
        self._stop.set()
        self._thread.join(timeout=1.0)


#: Per-frame processing cost the fake device charges, so a test can make the
#: pipeline slower than the camera (the real box chain costs ~10 ms at 480p).
PROCESS_COST_S = 0.0


class _Info:
    """The channel names the worker reads off ``sensor.info``."""

    FNORMAL = "FNORMAL"
    FSHEARX = "FSHEARX"
    FSHEARY = "FSHEARY"
    FNORMAL_RAW = "FNORMAL_RAW"
    FSHEARX_RAW = "FSHEARX_RAW"
    FSHEARY_RAW = "FSHEARY_RAW"


#: Every sensor the fake SDK handed out, so a test can read its frame counter.
CREATED: list = []


class _Sensor:
    def __init__(self, *args, **kwargs) -> None:
        import numpy as np

        self.configured: dict = {}
        self.connected = True
        self.capture_source = _CaptureSource()
        self.frame = np.zeros((8, 8, 3), dtype=np.uint8)
        self.info = _Info()
        self.force_calibrated = True
        CREATED.append(self)

    def configure(self, **kwargs):
        self.configured = kwargs

    def read_info(self, *names):
        """The SDK's readout: one value per requested channel, in order."""
        values = {
            _Info.FNORMAL: 1.5,
            _Info.FSHEARX: 0.25,
            _Info.FSHEARY: -0.25,
            _Info.FNORMAL_RAW: 15.0,
            _Info.FSHEARX_RAW: 2.5,
            _Info.FSHEARY_RAW: -2.5,
        }
        return tuple(values.get(str(name), 0.0) for name in names)

    def get_img(self):
        # ``read_latest``: hand back the newest frame immediately, the same one
        # again when nothing new has arrived. A test can make this cost time, so
        # the loop can be put under a pipeline slower than the camera.
        if PROCESS_COST_S > 0:
            time.sleep(PROCESS_COST_S)
        self.capture_source.take()
        return self.frame

    def disconnect(self):
        self.connected = False
        self.capture_source.close()


def _install() -> object:
    saved = {name: sys.modules.get(name) for name in ("orisys", "orisys.finger")}
    module = types.ModuleType("orisys")
    module.__version__ = "fake"
    module.Sensor = _Sensor  # type: ignore[attr-defined]
    finger_module = types.ModuleType("orisys.finger")
    finger_module.FingerRuntime = _Sensor  # type: ignore[attr-defined]
    module.finger = finger_module  # type: ignore[attr-defined]
    sys.modules["orisys"] = module
    sys.modules["orisys.finger"] = finger_module
    return saved


def _restore(saved) -> None:
    for name, original in saved.items():
        if original is None:
            sys.modules.pop(name, None)
        else:
            sys.modules[name] = original


def _collect(duration_s: float = 1.2):
    """Run the real worker loop against the fake device.

    Returns ``(packets, delivered)`` — what the loop recorded, and how many frames
    the fake capture source actually delivered in the same period.
    """
    from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker

    CREATED.clear()
    saved = _install()
    worker = SdkAcquisitionWorker()
    packets: list = []
    lock = threading.Lock()

    def sink(packet) -> None:
        with lock:
            packets.append(packet)

    try:
        worker.add_sink(sink)
        # Camera id 0, not "0": a string source is a file, and a file runs on the
        # video's timeline rather than the wall clock (see the file test below).
        worker.start(0, config_ref=BOX_CONFIG)
        time.sleep(duration_s)
        worker.stop(wait=True)
        delivered = CREATED[-1].capture_source.status.last_seq if CREATED else 0
    finally:
        _restore(saved)
    return packets, int(delivered)


def _rate(packets) -> float:
    span = float(packets[-1].mono_t) - float(packets[0].mono_t)
    return (len(packets) - 1) / span if span > 0 else 0.0


def test_the_same_frame_is_not_counted_twice():
    """One packet per delivered frame, whatever the loop's polling rate is."""
    packets, delivered = _collect()
    assert packets, "the loop emitted nothing at all"
    assert delivered > 10, f"the fake device only delivered {delivered} frames"
    assert len(packets) <= delivered * 1.5, (
        f"{len(packets)} packets for {delivered} delivered frames: the repeats "
        "are being recorded as samples"
    )
    assert len(packets) >= delivered * 0.7, (
        f"only {len(packets)} packets for {delivered} delivered frames"
    )


def test_a_30_fps_device_is_consumed_at_30_fps():
    """The loop must not add a timer tick of latency per frame.

    ``time.sleep(0.002)`` costs a full 15.6 ms on Windows, so a loop that sleeps
    before looking for the next frame spends ~47 ms per frame on a 30 fps device
    and records 16.7 fps with half the frames dropped.
    """
    packets, delivered = _collect(duration_s=1.5)
    assert packets, "the loop emitted nothing at all"
    rate = _rate(packets)
    assert 27.0 <= rate <= 31.5, (
        f"a 30 fps device was consumed at {rate:.1f} fps "
        f"({len(packets)} packets for {delivered} delivered frames)"
    )


def test_the_reported_rate_is_the_delivered_rate():
    """``packet.fps`` describes the frames, not how often the loop asked."""
    packets, delivered = _collect()
    reported = [float(packet.fps) for packet in packets if packet.fps]
    assert reported, "no packet carried a frame rate"
    measured = _rate(packets)
    settled = reported[-5:]
    assert all(value > 10.0 for value in settled), settled
    assert abs(settled[-1] - measured) < 0.25 * measured, (settled[-1], measured)


def test_the_camera_s_own_rate_is_reported_next_to_ours(monkeypatch):
    """A low frame rate must be attributable, not just visible.

    With a pipeline that fits the frame period the two rates agree. With one that
    does not, the *camera's* rate stays at 30 while ours falls and the source
    counts the frames nobody took — which is the whole difference between "the
    camera is slow" and "we are slow".
    """
    packets, _ = _collect(duration_s=1.5)
    assert packets, "the loop emitted nothing at all"
    camera = [float(packet.capture_fps) for packet in packets if packet.capture_fps]
    assert camera, "no packet carried the capture source's own rate"
    assert 28.0 <= camera[-1] <= 31.0, f"the fake camera publishes at 30 fps, got {camera[-1]}"
    assert packets[-1].capture_dropped == 0, (
        f"the loop dropped {packets[-1].capture_dropped} frames while keeping up"
    )

    # Now make the pipeline three times slower than the camera's own period.
    monkeypatch.setattr(sys.modules[__name__], "PROCESS_COST_S", 0.06)
    slow, _ = _collect(duration_s=2.0)
    try:
        assert slow, "the loop emitted nothing at all"
        ours = _rate(slow)
        assert ours < 22.0, f"a 60 ms pipeline still kept up at {ours:.1f} fps"
        assert 28.0 <= float(slow[-1].capture_fps) <= 31.0, (
            "the camera's rate was reported as slow too: "
            f"{slow[-1].capture_fps}"
        )
        assert slow[-1].capture_dropped > 5, (
            "the loop fell behind without recording a single dropped frame: "
            f"{slow[-1].capture_dropped}"
        )
    finally:
        monkeypatch.setattr(sys.modules[__name__], "PROCESS_COST_S", 0.0)


def test_a_low_camera_rate_is_recorded_as_a_negotiation(monkeypatch):
    """A camera that cannot do the configured rate must be visible as such.

    The pipeline cannot turn 16.7 fps into 30, so the only honest thing left is to
    record what was asked for and what the driver gave, and to have a one-line
    description ready for the operator.
    """
    from orisys_app_shared.acquisition.sdk_worker import (
        SdkAcquisitionWorker,
        short_capture_fallacy,
    )

    CREATED.clear()
    saved = _install()
    worker = SdkAcquisitionWorker()
    try:
        worker.start(0, config_ref=BOX_CONFIG)
        deadline = time.perf_counter() + 2.0
        while time.perf_counter() < deadline and not CREATED:
            time.sleep(0.01)
        assert CREATED, "the fake sensor was never constructed"
        source = CREATED[-1].capture_source
        source.status.fps = 16.67  # what the driver actually negotiated
        status = worker.capture_status()
        assert status["requested"]["fps"] == 30.0
        assert status["requested"]["width"] == 1920
        assert status["negotiated"]["fps"] == 16.67, status
        assert status["negotiated"]["fourcc"] == "MJPG"
        assert status["negotiated"]["backend"] == "MSMF"
        # The *measured* rate is the evidence: DSHOW echoes the requested fps back
        # as the "negotiated" one, so trusting that number would hide a camera
        # that advertises 30 and delivers 15 (measured on this project's camera).
        detail = short_capture_fallacy(status, measured_fps=16.67)
        assert "16.67" in detail and "30" in detail and "1920x1080" in detail, detail
        assert short_capture_fallacy(status, measured_fps=29.5) == ""
        # A driver reporting 30 while delivering half of it is the case that
        # matters, and it must not slip through.
        source.status.fps = 30.0
        echoing = worker.capture_status()
        assert short_capture_fallacy(echoing, measured_fps=15.0) != ""
        # No measured rate yet: nothing to say.
        assert short_capture_fallacy(echoing) == ""
        # ...and the session's own measurement is part of what gets recorded, so
        # a unit can be judged by the rate it really delivered.
        deadline = time.perf_counter() + 2.0
        while time.perf_counter() < deadline and not worker.capture_status().get("measured_fps"):
            time.sleep(0.05)
        measured = float(worker.capture_status().get("measured_fps") or 0.0)
        assert 20.0 <= measured <= 35.0, (
            f"the fake camera publishes at ~30 fps, recorded {measured}"
        )
    finally:
        worker.stop(wait=True)
        _restore(saved)


def test_a_file_source_runs_on_the_video_s_own_timeline():
    """A pass over a video is stamped with the video's time, not the wall clock."""
    from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker

    CREATED.clear()
    saved = _install()
    worker = SdkAcquisitionWorker()
    packets: list = []
    lock = threading.Lock()

    def sink(packet) -> None:
        with lock:
            packets.append(packet)

    try:
        worker.add_sink(sink)
        worker.start("recorded.mp4", config_ref=BOX_CONFIG)
        time.sleep(0.4)
        worker.stop(wait=True)
    finally:
        _restore(saved)
    assert len(packets) > 3, f"the pass produced {len(packets)} packets"
    steps = [
        round(packets[index + 1].mono_t - packets[index].mono_t, 6)
        for index in range(min(5, len(packets) - 1))
    ]
    assert packets[0].mono_t == 0.0, packets[0].mono_t
    assert all(abs(step - 1.0 / 30.0) < 1e-6 for step in steps), steps
    assert packets[-1].mono_t < 0.4 + 1.0, (
        f"the pass ran on the wall clock: {packets[-1].mono_t} s of video in "
        f"{0.4} s of real time"
    )


def test_a_file_that_yields_no_frames_ends_the_session_with_an_error(monkeypatch):
    """A video this build cannot decode must report it, not wait forever.

    The end-of-file branch needs at least one frame to have arrived (``saw_frame``),
    so a file that never produces one used to leave the loop spinning: no frames, no
    error, and the window in "connecting" until somebody pressed Stop. The shipping
    deadline is ten seconds; it is shortened here, because the point is the behaviour
    rather than the wait.
    """
    import orisys_app_shared.acquisition.sdk_worker as worker_module

    monkeypatch.setattr(worker_module, "FILE_FIRST_FRAME_TIMEOUT_S", 0.3)
    # A sensor that never produces a frame: what an unreadable codec looks like.
    monkeypatch.setattr(_Sensor, "get_img", lambda self: None)

    CREATED.clear()
    saved = _install()
    errors: list = []
    packets: list = []
    worker = worker_module.SdkAcquisitionWorker(on_error=errors.append)
    try:
        worker.add_sink(packets.append)
        worker.start("unreadable.mp4", config_ref=BOX_CONFIG)
        deadline = time.perf_counter() + 10.0
        while time.perf_counter() < deadline and worker.is_running:
            time.sleep(0.05)
        assert not worker.is_running, (
            "the worker is still running on a file that produced no frames"
        )
        assert worker.consume_end_reason() == "error", worker.consume_end_reason()
        assert packets == []
        assert errors and "no frames" in errors[0], errors
    finally:
        worker.stop(wait=True)
        _restore(saved)


def test_a_paused_worker_stops_producing_but_keeps_the_device_open():
    """Pause is not stop: compute stops, the capture keeps running.

    It lives in this file because the fake device that can answer the question (does the
    source keep publishing while the loop takes no frame?) is the harness above.
    """
    from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker

    CREATED.clear()
    saved = _install()
    worker = SdkAcquisitionWorker()
    packets: list = []
    lock = threading.Lock()

    def sink(packet) -> None:
        with lock:
            packets.append(packet)

    try:
        worker.add_sink(sink)
        worker.start(0, config_ref=BOX_CONFIG)
        time.sleep(0.4)
        before = len(packets)
        assert before > 0, "the fake device produced nothing to pause"
        assert not worker.paused

        worker.pause()
        assert worker.paused
        time.sleep(0.4)
        during = len(packets)
        # The frames the source published across the pause: the device is demonstrably
        # alive, which is what makes this different from ``stop``.
        source_seq_at_pause = CREATED[-1].capture_source.status.last_seq
        assert during == before, (
            f"a paused worker emitted {during - before} packet(s): the computation was "
            "supposed to stop"
        )

        worker.resume()
        assert not worker.paused
        time.sleep(0.4)
        after = len(packets)
        assert after > during, "the worker did not resume producing packets"
        assert CREATED[-1].capture_source.status.last_seq > source_seq_at_pause, (
            "the capture source stopped publishing while paused — the device was released "
            "or the reader was starved, and neither is what pause means"
        )
        # Resuming continues from the newest frame rather than replaying the pause: the
        # packet that follows the pause is dated after it, not before.
        assert float(packets[after - 1].mono_t) > 0.0
        stop_requests = getattr(worker, "_stop").is_set()
        assert not stop_requests, "pausing must not set the stop flag"
        worker.stop(wait=True)
    finally:
        _restore(saved)


def test_a_new_session_starts_unpaused():
    """Pausing is a statement about one session, not a sticky property of the device."""
    from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker

    CREATED.clear()
    saved = _install()
    worker = SdkAcquisitionWorker()
    try:
        worker.add_sink(lambda _packet: None)
        worker.start(0, config_ref=BOX_CONFIG)
        time.sleep(0.3)
        worker.pause()
        assert worker.paused
        worker.stop(wait=True)

        assert not worker.paused, "stop cleared the pause flag"
        worker.start(0, config_ref=BOX_CONFIG)
        assert not worker.paused, "the next session started paused"
        worker.stop(wait=True)
    finally:
        _restore(saved)


if __name__ == "__main__":  # pragma: no cover - manual runner
    failures = 0
    for name, function in sorted(globals().items()):
        if name.startswith("test_") and callable(function):
            try:
                function()
                print(f"  ok   {name}")
            except Exception as exc:  # noqa: BLE001
                failures += 1
                print(f"  FAIL {name}: {type(exc).__name__}: {exc}")
    print(f"failures: {failures}")
    raise SystemExit(1 if failures else 0)