"""The concurrent acquisition path: no left-behind threads, and a stop/start cycle.

The worker drives the SDK's ``StreamPipeline``, which runs three threads of its own
(``orisys-pipeline-produce``, ``-flow``, ``-hhd``). Those threads own the sensor's
buffers and tracker state, so a session that leaves one running would corrupt the
next session's reference frame — the failure this test exists to catch.

No camera and no sensor bundle are needed: the fake ``orisys`` module below serves
``Sensor`` and ``finger.FingerRuntime`` the way the real package does, including the
compute methods ``StreamPipeline`` calls.
"""

from __future__ import annotations

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

import numpy as np
import pytest

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

PIPELINE_THREADS = ("orisys-pipeline-produce", "orisys-pipeline-flow", "orisys-pipeline-hhd")

#: Bound by :func:`_install` from the real SDK, before the fake device replaces the
#: ``orisys`` name.
StreamPipeline = None
FrameSnapshot = None


def _pipeline_threads() -> list[str]:
    return [t.name for t in threading.enumerate() if t.name in PIPELINE_THREADS]


class _Info:
    FNORMAL = "FNORMAL"
    FSHEARX = "FSHEARX"
    FSHEARY = "FSHEARY"
    FNORMAL_RAW = "FNORMAL_RAW"
    FSHEARX_RAW = "FSHEARX_RAW"
    FSHEARY_RAW = "FSHEARY_RAW"
    VRAW = "VRAW"
    VNORMAL = "VNORMAL"
    VSHEAR = "VSHEAR"
    DEPTH = "DEPTH"
    CENTROID = "CENTROID"
    CONTOUR = "CONTOUR"
    IMG = "IMG"


class _Source:
    """A capture source with a running background reader, so repeats are counted."""

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

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

    def __init__(self) -> None:
        self.status = self._Status()
        self.config = self._Config()
        self.running = True

    def read_latest(self, **kwargs):
        return True, np.zeros((8, 8, 3), dtype=np.uint8)

    def is_open(self) -> bool:
        return True

    def start_background(self) -> None:
        self.running = True

    def stop_background(self) -> None:
        self.running = False

    def open(self) -> None:
        return None

    def close(self) -> None:
        self.running = False


class _Sensor:
    """A device ``StreamPipeline`` can drive."""

    def __init__(self, *args, **kwargs) -> None:
        self.info = _Info()
        self.configured: dict = {}
        self.connected = True
        self.capture_source = _Source()
        self._capture_source = self.capture_source
        self.img = np.zeros((8, 8), dtype=np.uint8)
        self.frame = np.zeros((8, 8, 3), dtype=np.uint8)
        self.flow = None
        self.vnormal = None
        self.vshear = None
        self.depth_map = None
        self.centroid = None
        self.centroid_found = False
        self.contour = None
        self.compute_contact_valid = False
        self.valid_mask = None
        self.img_size_raw = (1920, 1080)
        self.force_calibrated = True
        self.calibration = None
        self.fnormal = 1.0
        self.fshearx = 0.0
        self.fsheary = 0.0
        self.ftor = 0.0
        self.force_unit = "N"
        self._force_raw = (9.0, 0.5, -0.5)
        self._frames = 0

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

    def read_info(self, *names):
        values = {
            _Info.FNORMAL: 1.0,
            _Info.FSHEARX: 0.0,
            _Info.FSHEARY: 0.0,
            _Info.FNORMAL_RAW: 9.0,
            _Info.FSHEARX_RAW: 0.5,
            _Info.FSHEARY_RAW: -0.5,
        }
        return tuple(values.get(str(name), None) for name in names)

    def compute_flow(self, img=None):
        self.flow = np.zeros((8, 8, 2), dtype=np.float32)

    def compute_hhd(self, flow=None):
        self._force_raw = (9.0, 0.5, -0.5)

    def compute_deformation(self, decompose=True, check_motion=False, **kwargs):
        self.compute_flow()
        if decompose:
            self.compute_hhd()
        return None

    def compute_contact(self, vnormal=None):
        self.compute_contact_valid = True
        return {}

    def get_img(self, **kwargs):
        self._frames += 1
        return self.img.copy()

    def warmup_deformation(self):
        return True

    def reset(self):
        return None

    def disconnect(self):
        self.connected = False


class _Runtime:
    """A stand-in for ``FingerRuntime`` with the API the worker uses."""

    def __init__(self, *args, **kwargs) -> None:
        self._source_seq = 0
        self.tactile = types.SimpleNamespace(
            raw_sensor=_Sensor(),
            reset=lambda: None,
        )

    @classmethod
    def open(cls, **kwargs):
        return cls(**kwargs)

    @property
    def valid_mask(self):
        return np.ones((8, 8), dtype=bool)

    def read_source(self):
        return True, np.zeros((8, 8, 3), dtype=np.uint8)

    def read_source_stamped(self):
        self._source_seq += 1
        return types.SimpleNamespace(
            image=np.zeros((8, 8, 3), dtype=np.uint8),
            sequence=self._source_seq,
            host_mono_ns=time.perf_counter_ns(),
            host_utc_ns=time.time_ns(),
            media_time_ms=self._source_seq * 50.0,
        )

    def process_frame(self, frame, *, color=False):
        return np.zeros((8, 8), dtype=np.uint8)

    def warmup(self, *, color=False):
        return True, np.zeros((8, 8), dtype=np.uint8)

    def close(self):
        return None


def _install() -> dict:
    """Replace ``orisys`` with the fakes, keeping the real ``StreamPipeline``.

    ``StreamPipeline`` and ``FrameSnapshot`` are the SDK under test, so they are
    bound before the swap: the fake stands in for the device, not for the pipeline.
    The submodules the workers import lazily are bound too, because a module object
    placed in ``sys.modules`` is not a package and cannot resolve them itself.
    """
    import orisys.calibration  # noqa: F401 - bound so the workers can import it
    import orisys.finger  # noqa: F401
    import orisys.pipeline  # noqa: F401

    global StreamPipeline, FrameSnapshot
    from orisys.pipeline import FrameSnapshot as _Snapshot
    from orisys.pipeline import StreamPipeline as _Pipeline

    StreamPipeline, FrameSnapshot = _Pipeline, _Snapshot
    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 = _Runtime  # 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: dict) -> None:
    for name, original in saved.items():
        if original is None:
            sys.modules.pop(name, None)
        else:
            sys.modules[name] = original


def _worker(kind: str):
    from orisys_app_shared.acquisition.finger_worker import FingerAcquisitionWorker
    from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker

    return FingerAcquisitionWorker() if kind == "finger" else SdkAcquisitionWorker()


def _config(kind: str) -> Path:
    return FINGER_CONFIG if kind == "finger" else BOX_CONFIG


@pytest.mark.parametrize("source", ["recording.mp4", 0])
def test_finger_video_and_camera_use_the_same_packet_path(source, monkeypatch):
    """Both source forms reach FingerRuntime and publish the same packet contract."""
    opened: list[str | int] = []

    def open_runtime(cls, **kwargs):
        opened.append(kwargs["video"])
        return cls()

    saved = _install()
    monkeypatch.setattr(_Runtime, "open", classmethod(open_runtime))
    worker = _worker("finger")
    packets: list = []
    try:
        worker.add_sink(packets.append)
        worker.start(source, config_ref=FINGER_CONFIG)
        deadline = time.perf_counter() + 5.0
        while time.perf_counter() < deadline and not packets:
            time.sleep(0.02)
        assert opened == [source]
        assert packets
        packet = packets[0]
        assert packet.valid
        assert packet.force_calibrated
        assert packet.raw_frame.shape == (8, 8, 3)
        assert packet.processed_frame.shape == (8, 8, 3)
        assert packet.session_id == 1
        assert packet.frame_id == 0
        assert packet.source_frame_id == 1
        assert packet.capture_mono_ns is not None
        assert packet.capture_utc_ns is not None
        assert packet.timestamp == pytest.approx(packet.capture_utc_ns / 1e9)
        assert packet.time_basis == (
            "host_replay" if isinstance(source, str) else "host_camera"
        )
        assert packet.media_time_ms == 50.0
    finally:
        worker.stop(wait=True)
        _restore(saved)


@pytest.mark.parametrize(
    "kind, source", [("box", "0"), ("finger", "recording.mp4"), ("finger", 0)]
)
def test_a_stopped_session_leaves_no_pipeline_threads(kind, source):
    """``stop`` must join the three stages: they own the sensor's tracker state."""
    packets: list = []
    saved = _install()
    worker = _worker(kind)
    try:
        worker.add_sink(packets.append)
        worker.start(source, config_ref=_config(kind))
        deadline = time.perf_counter() + 5.0
        while time.perf_counter() < deadline and not packets:
            time.sleep(0.02)
        assert packets, "the session published nothing to stop"
        if kind == "finger" and source == 0:
            assert _pipeline_threads() == []
        else:
            assert _pipeline_threads(), "the fake session did not start the pipeline"
        worker.stop(wait=True)
        deadline = time.perf_counter() + 5.0
        while time.perf_counter() < deadline and _pipeline_threads():
            time.sleep(0.02)
        assert _pipeline_threads() == [], (
            f"{kind}: pipeline threads outlived stop(): {_pipeline_threads()}"
        )
    finally:
        worker.stop(wait=True)
        _restore(saved)


@pytest.mark.parametrize("kind", ["box", "finger"])
def test_a_stop_start_cycle_works_and_leaves_nothing_behind(kind):
    """One worker, two sessions: the second must run and the first must be gone."""
    saved = _install()
    worker = _worker(kind)
    try:
        for session in (1, 2):
            packets: list = []
            worker.add_sink(packets.append)
            worker.start("0", config_ref=_config(kind))
            deadline = time.perf_counter() + 5.0
            while time.perf_counter() < deadline and not packets:
                time.sleep(0.02)
            assert packets, f"session {session} published nothing"
            assert worker.latest() is packets[-1]
            worker.stop(wait=True)
            assert not worker.is_running
            assert worker.phase == "idle"
            worker.remove_sink(packets.append)
            deadline = time.perf_counter() + 5.0
            while time.perf_counter() < deadline and _pipeline_threads():
                time.sleep(0.02)
            assert _pipeline_threads() == [], _pipeline_threads()
    finally:
        worker.stop(wait=True)
        _restore(saved)


def test_the_packet_carries_every_field_the_ui_reads():
    """The mapping from a snapshot: forces, raws, arrays, ids and provenance."""
    import orisys_app_shared.acquisition.models as models

    saved = _install()
    try:
        snapshot = FrameSnapshot(
            seq=7,
            timestamp=1000.5,
            age_ms=12.0,
            fps=30.0,
            unit="N",
            normal_force=1.25,
            shear_x=0.5,
            shear_y=-0.5,
            normal_force_raw=9.0,
            shear_x_raw=0.4,
            shear_y_raw=-0.4,
            torque=0.0,
            image=np.zeros((4, 4), dtype=np.uint8),
            flow=np.zeros((4, 4, 2), dtype=np.float32),
            depth=np.ones((4, 4), dtype=np.float32),
            centroid=np.array([2.0, 3.0]),
            contour=None,
            produce_ms=1.0,
            flow_ms=2.0,
            hhd_ms=3.0,
        )
        packet = models.from_snapshot(
            snapshot,
            _Sensor(),
            session_id=3,
            frame_id=7,
            mono_t=0.25,
            capture_fps=30.0,
            capture_dropped=2,
            raw_frame=np.zeros((8, 8, 3), dtype=np.uint8),
            valid_mask=np.ones((4, 4), dtype=bool),
            contact_analysed=True,
            contact_found=True,
        )
        assert packet.frame_id == 7
        assert packet.session_id == 3
        assert packet.mono_t == pytest.approx(0.25)
        assert packet.fps == 30.0
        # The stage costs add up to the frame's own cost; the SDK measures them.
        assert packet.process_ms == pytest.approx(6.0)
        assert (packet.normal_force, packet.shear_x, packet.shear_y) == (1.25, 0.5, -0.5)
        assert (
            packet.normal_force_raw,
            packet.shear_x_raw,
            packet.shear_y_raw,
        ) == (9.0, 0.4, -0.4)
        assert packet.force_unit == "N"
        assert packet.capture_fps == 30.0
        assert packet.capture_dropped == 2
        assert packet.processed_frame.shape == (4, 4, 3)
        assert packet.raw_frame.shape == (8, 8, 3)
        assert packet.flow is not None and packet.depth is not None
        assert packet.valid_mask is not None
        assert packet.flow_valid is True and packet.flow_frame_id == 7
        assert packet.depth_valid is True and packet.depth_frame_id == 7
        assert packet.forces_valid is True and packet.valid is True
        assert packet.contact is not None
        assert packet.contact.centroid_uv_xy == (2.0, 3.0)
        assert packet.contact.found is True and packet.contact.frame_id == 7
    finally:
        _restore(saved)


def test_a_failed_chain_marks_the_frame_rather_than_claiming_its_forces():
    """A compute error must leave the frame's forces unclaimable, not silently trusted."""
    import orisys_app_shared.acquisition.models as models

    saved = _install()
    try:
        snapshot = FrameSnapshot(
            seq=0,
            timestamp=1000.0,
            age_ms=1.0,
            fps=0.0,
            unit="raw",
            normal_force=1.0,
            shear_x=0.0,
            shear_y=0.0,
            normal_force_raw=1.0,
            shear_x_raw=0.0,
            shear_y_raw=0.0,
            torque=0.0,
            image=np.zeros((4, 4), dtype=np.uint8),
            flow=None,
            depth=None,
            centroid=np.array(0),
            contour=None,
        )
        packet = models.from_snapshot(
            snapshot,
            _Sensor(),
            frame_id=0,
            contact_analysed=True,
            contact_found=False,
            compute_error="compute: ValueError: boom",
        )
        assert packet.valid is False
        assert packet.stale_forces is True
        assert packet.forces_valid is False
        assert packet.flow_valid is False and packet.depth_valid is False
        assert packet.contact is not None
        assert packet.contact.found is False and packet.contact.valid is False
        assert packet.contact.centroid_uv_xy is None
    finally:
        _restore(saved)


def test_the_finger_producer_reports_the_camera_rate(monkeypatch):
    """The first live frame asks the producer for ``capture_fps``.

    That call is what ended a finger session with
    ``'_FingerProducer' object has no attribute 'capture_fps'``. A file has no
    delivered-frame counter, so it reports nothing; a live source reports the
    frames its counter advanced over the wall-clock between two reads.
    """
    from orisys_app_shared.acquisition import finger_worker

    clock = {"t": 1000.0}
    monkeypatch.setattr(finger_worker.time, "perf_counter", lambda: clock["t"])

    file_source = types.SimpleNamespace(running=False, status=types.SimpleNamespace(last_seq=0))
    file_runtime = types.SimpleNamespace(capture_source=file_source)
    file_producer = finger_worker._FingerProducer(file_runtime, is_file=True)
    assert file_producer.capture_fps() == 0.0

    status = types.SimpleNamespace(last_seq=10, dropped_frames=0)
    live = types.SimpleNamespace(running=True, status=status)
    producer = finger_worker._FingerProducer(
        types.SimpleNamespace(capture_source=live), is_file=False
    )
    assert producer.capture_fps() == 0.0
    clock["t"] = 1001.0
    producer.capture_seq = 40
    assert producer.capture_fps() == pytest.approx(30.0)