"""A manually confirmed site produces one traceable trial directory."""

from __future__ import annotations

import csv
import json
import time
from dataclasses import replace
from types import SimpleNamespace

import numpy as np
import pytest
from orisys_app_shared.acquisition.models import SdkPacket
from orisys_autocal.trial import (
    REFERENCE_HEADER,
    ManualTrial,
    load_path,
)

from orisys_autocal.sensor.uv_flat import FlatSurface, site_at, uv_origin


class Worker:
    def __init__(self) -> None:
        self.is_running = True
        self.sinks = []
        self.packet = SdkPacket(
            raw_frame=np.zeros((32, 32, 3), dtype=np.uint8),
            processed_frame=np.zeros((32, 32, 3), dtype=np.uint8),
            timestamp=time.time(),
            fps=30,
            process_ms=1,
            session_id=2,
            frame_id=1,
            mono_t=1.0,
            capture_mono_ns=time.perf_counter_ns(),
            capture_utc_ns=time.time_ns(),
            source_frame_id=7,
            time_basis="host_camera",
            normal_force_raw=1.25,
        )

    def latest(self):
        return self.packet

    def add_sink(self, sink):
        self.sinks.append(sink)

    def remove_sink(self, sink):
        self.sinks.remove(sink)

    def emit(self):
        for sink in self.sinks:
            sink(self.packet)

    def stop(self, *, wait=True):
        self.is_running = False


@pytest.fixture
def flat():
    mask = np.full((5, 7), -1, dtype=np.int32)
    mask[1:4, 2:5] = 2
    mask[4, 2] = 2  # A valid pixel may be below the centre-column origin.
    positions = np.zeros((5, 7, 3), dtype=np.float64)
    for y in range(5):
        for x in range(7):
            positions[y, x] = (x, y, 3)
    normals = np.zeros_like(positions)
    normals[:, :, 2] = 1
    return FlatSurface((5, 7), np.zeros((5, 7, 3), dtype=np.uint8), positions,
                       normals, mask, (0, 0), (1, 1))


@pytest.fixture
def clock(monkeypatch):
    from orisys_autocal import trial as trial_module

    class Clock:
        now = time.perf_counter_ns()

        def advance(self, seconds):
            self.now += round(seconds * 1e9)

    value = Clock()
    monkeypatch.setattr(trial_module, "time", SimpleNamespace(
        perf_counter_ns=lambda: value.now, time_ns=time.time_ns))
    return value


@pytest.fixture
def trial_setup(tmp_path, flat):
    bundle = tmp_path / "bundle"
    bundle.mkdir()
    (bundle / "mesh.npz").write_bytes(b"mesh")
    config = tmp_path / "sensor.json"
    config.write_text("{}", encoding="utf-8")
    return dict(root=tmp_path, site=site_at(flat, -1, 2), levels=[5, 10],
                hold_s=20, sensor_model="Finger D40", sensor_sn="F-001",
                bundle_path=str(bundle), source={"kind": "camera", "value": 0},
                config_ref=str(config), fps=30)


def test_uv_coordinates_are_centered_and_masked(flat):
    assert uv_origin(flat) == (3, 3)
    site = site_at(flat, -1, 2)
    assert site.raster_xy_px == (2, 1)
    assert site.position_sensor_mm == (2.0, 1.0, 3.0)
    assert site_at(flat, -1, -1).raster_xy_px == (2, 4)
    with pytest.raises(ValueError):
        site_at(flat, -3, 0)


def test_trial_persists_colour_region_outline_and_rejects_mismatched_mapping(trial_setup, flat):
    from orisys_autocal.sensor.uv_outline import outline_snapshot

    flat.textured = True
    flat.image_rgb[:] = 255
    flat.image_rgb[2:4, 2:5] = [20, 80, 200]
    snapshot = outline_snapshot(flat)
    worker = Worker()
    trial = ManualTrial(worker)
    with pytest.raises(ValueError, match="outline"):
        trial.start(**trial_setup, uv_outline={**snapshot, "mapping_sha256": "wrong"})
    directory = trial.start(**trial_setup, uv_outline=snapshot)
    worker.emit()
    trial.finish(aborted=True, reason="test ended")
    trial.recorder.stop(wait=True)
    trial.poll()
    manifest = json.loads((directory.parent.parent / "session.json").read_text(encoding="utf-8"))
    assert manifest["uv_outline"] == snapshot and snapshot["patch"]
    trial.finish_session()


def test_manual_trial_records_video_stamps_and_load_unload_events(trial_setup, clock):
    worker = Worker()
    trial = ManualTrial(worker)
    assert load_path([5, 10]) == [
        ("zero", 0), ("load", 5), ("load", 10), ("unload", 5), ("unload", 0)]
    directory = trial.start(**trial_setup)
    worker.emit()
    for index, value in enumerate((0, 5, 10, 5, 0)):
        assert trial.step_index == index
        trial.confirm(value)
        assert trial.step_index == index
        with pytest.raises(RuntimeError, match="hold"):
            trial.confirm(value)
        with pytest.raises(RuntimeError):
            trial.finish()
        clock.advance(19)
        trial.poll()
        assert trial.holding
        assert trial.remaining_s == pytest.approx(1)
        clock.advance(3)  # Late UI polling must not extend the reference window.
        trial.poll()
        assert trial.step_index == index + 1
        assert not trial.holding
        clock.advance(4)  # Adjustment time must have no constant-force label.
    trial.finish()
    assert trial.recorder.stop(wait=True).ok
    assert trial.poll() == "complete"
    assert directory.name == "trial-001"
    assert "Finger-D40_F-001" in directory.parent.parent.name
    assert len(list(directory.glob("raw_*.mp4"))) == 1
    with (directory / "frames.csv").open(encoding="utf-8", newline="") as handle:
        row = next(csv.DictReader(handle))
    assert row["t_host_mono_ns"] == str(worker.packet.capture_mono_ns)
    assert row["source_frame_id"] == "7"
    assert row["time_basis"] == "host_camera"
    with (directory / "reference.csv").open(encoding="utf-8", newline="") as handle:
        reader = csv.DictReader(handle)
        assert tuple(reader.fieldnames) == REFERENCE_HEADER
        references = list(reader)
    assert [row["branch"] for row in references] == ["zero", "load", "load", "unload", "unload"]
    assert [float(row["F_ref_normal_N"]) for row in references] == [0, 5, 10, 5, 0]
    assert all(row["F_ref_x_N"] == row["F_ref_y_N"] == row["F_ref_z_N"] == "" for row in references)
    assert all(row["stability_verified"] == "0" and row["hold_complete"] == "1" for row in references)
    assert all(int(r["t_end_host_mono_ns"]) - int(r["t_start_host_mono_ns"]) == 20_000_000_000 for r in references)
    assert all(int(a["t_end_host_mono_ns"]) < int(b["t_start_host_mono_ns"]) for a, b in zip(references, references[1:]))
    with (directory / "session.json").open(encoding="utf-8") as handle:
        manifest = json.load(handle)
    assert manifest["trial_status"] == "complete"
    assert manifest["site"]["uv_xy_px"] == [-1, 2]
    assert len(manifest["steps"]) == 5
    assert manifest["uv_definition"]["mapping_sha256"] == trial_setup["site"].mapping_sha256
    session_path = directory.parent.parent / "session.json"
    root = json.loads(session_path.read_text(encoding="utf-8"))
    assert root["status"] == "in_progress"
    assert root["trials"][0]["status"] == "complete"
    assert trial.finish_session()[1] == "complete"
    assert json.loads(session_path.read_text(encoding="utf-8"))["status"] == "complete"


def test_retries_and_other_sites_share_one_session(trial_setup, clock):
    worker = Worker()
    trial = ManualTrial(worker)
    directories = []
    for site in (trial_setup["site"], trial_setup["site"],
                 replace(trial_setup["site"], u_px=0, raster_xy_px=(3, 1), position_sensor_mm=(3, 1, 3))):
        directory = trial.start(**{**trial_setup, "site": site})
        directories.append(directory)
        worker.emit()
        trial.confirm(0)
        clock.advance(3)
        trial.finish(aborted=True, reason="operator ended early")
        trial.recorder.stop(wait=True)
        assert trial.poll() == "incomplete"
        rows = list(csv.DictReader((directory / "reference.csv").open(encoding="utf-8")))
        assert rows[0]["hold_complete"] == "0"
    assert len({d.parent.parent for d in directories}) == 1
    root = trial.session.document
    assert len(root["sites"]) == 2
    assert [t["repeat_index"] for t in root["trials"]] == [1, 2, 1]
    assert [t["site_id"] for t in root["trials"]] == ["site-001", "site-001", "site-002"]
    with pytest.raises(ValueError, match="finish this session"):
        trial.start(**{**trial_setup, "sensor_sn": "F-002"})
    with pytest.raises(ValueError, match="finish this session"):
        trial.start(**{**trial_setup, "site": replace(site, mapping_sha256="other")})
    assert trial.finish_session()[1] == "incomplete"
    new_directory = trial.start(**trial_setup)
    assert new_directory.parent.parent != directories[0].parent.parent
    trial.finish(aborted=True)
    trial.recorder.stop(wait=True)
    trial.poll()
    trial.finish_session()


def test_input_end_aborts_active_hold(trial_setup, clock):
    worker = Worker()
    worker.packet = replace(worker.packet, time_basis="host_replay")
    trial = ManualTrial(worker)
    directory = trial.start(**trial_setup)
    assert not trial.reference_alignment_possible
    worker.emit()
    trial.confirm(0)
    clock.advance(2)
    worker.is_running = False
    trial.poll()
    trial.recorder.stop(wait=True)
    assert trial.poll() == "incomplete"
    assert trial.entry["reason"] == "input or writer stopped"
    assert json.loads((directory / "session.json").read_text(encoding="utf-8"))["steps"][0]["status"] == "incomplete"
    trial.finish_session()


def test_a_trial_recorded_without_video_finishes_complete(trial_setup, clock):
    """Data-only recording: the force rows are the evidence and the verdict says so."""
    from orisys_autocal.recording_options import RecordingOptions

    worker = Worker()
    trial = ManualTrial(worker)
    directory = trial.start(**{**trial_setup, "recording": RecordingOptions(video=False)})

    for index, (_branch, target) in enumerate(trial.steps):
        trial.confirm(target)
        # Two frames inside the second half of the hold: the summary's own rule.
        # Each frame carries its own id: a repeated id has no single acquisition
        # time and the review would exclude it.
        for delay, frame_id in ((15, index * 2 + 2), (1, index * 2 + 3)):
            clock.advance(delay)
            worker.packet = replace(
                worker.packet, capture_mono_ns=clock.now,
                frame_id=frame_id, source_frame_id=frame_id,
            )
            worker.emit()
        clock.advance(4)
        trial.poll()
        assert trial.step_index == index + 1

    trial.finish()
    assert trial.recorder.stop(wait=True).ok
    assert trial.poll() == "complete"
    assert not list(directory.glob("raw_*.mp4"))
    root = json.loads(
        (directory.parent.parent / "session.json").read_text(encoding="utf-8")
    )
    assert root["recording"] == {
        "video": False, "codec": "mp4v", "lossless_frames": False}

    from orisys_autocal.summary import build_summary

    summary = build_summary(directory.parent.parent)
    assert all(row["eligible"] for row in summary["windows"]), [
        (row["excluded_reason"], row["capture_status"], row["samples_valid"])
        for row in summary["windows"]
    ]
    assert all(row["capture_status"] == "complete" for row in summary["windows"])
    trial.finish_session()


def test_recording_settings_are_locked_by_the_first_trial(trial_setup, clock):
    """One session, one recording choice: the manifest must not change mid-session."""
    from orisys_autocal.recording_options import RecordingOptions

    worker = Worker()
    trial = ManualTrial(worker)
    directory = trial.start(
        **{**trial_setup, "recording": RecordingOptions(video=False)}
    )
    root = json.loads(
        (directory.parent.parent / "session.json").read_text(encoding="utf-8")
    )
    assert root["recording"] == {
        "video": False, "codec": "mp4v", "lossless_frames": False}

    worker.emit()
    trial.finish(aborted=True, reason="test ended")
    trial.recorder.stop(wait=True)
    trial.poll()

    with pytest.raises(ValueError, match="locked"):
        trial.start(**{**trial_setup, "recording": RecordingOptions(video=True)})
    # The locked settings are what a later trial of this session uses.
    second = trial.start(
        **{**trial_setup, "recording": RecordingOptions(video=False)}
    )
    assert trial.recording.video is False
    assert second.parent.parent == directory.parent.parent
    trial.recorder.stop(wait=True)
    trial.poll()
    trial.finish_session()


def test_the_capture_page_no_longer_drives_a_manual_trial(make_window):
    """D1: a new capture is a one-point queue in the ``sites/`` format.

    The ManualTrial flow stays only so old ``trials/`` sessions keep reading; the
    capture page owns a Campaign plus the workflow controller. The two-page
    single/multi-point split it used to switch between is gone.
    """
    window = make_window()
    page = window.calibration_panel
    assert not hasattr(page, "trial")
    assert not hasattr(page, "multipoint")
    assert page.campaign is not None
    assert page.controller is not None
    window.close()


def test_the_capture_page_keeps_the_picked_uv_point_while_the_live_view_is_visible(
    make_window, qt_app
):
    """Both images are on the page at once; picking on the map keeps the point."""
    from orisys_autocal.sensor.uv_flat import FlatSurface

    window = make_window()
    window.resize(1440, 900)
    window.show()
    qt_app.processEvents()
    window.workspace.set_current_key("sensor")
    qt_app.processEvents()
    mask = np.full((5, 7), -1, dtype=np.int32)
    mask[1:4, 2:5] = 2
    positions = np.zeros((5, 7, 3), dtype=np.float64)
    positions[1, 2] = (2, 1, 3)
    flat = FlatSurface(
        size_hw=(5, 7),
        image_rgb=np.zeros((5, 7, 3), dtype=np.uint8),
        positions_mm=positions,
        normals=np.zeros_like(positions),
        face_index=mask,
        uv_min=(0, 0),
        uv_max=(1, 1),
    )
    page = window.calibration_panel
    page.set_flat(flat, "bundle")
    page._picked(2, 1)
    assert (page.u_spin.value(), page.v_spin.value()) == (-1, 2)
    assert page._site.raster_xy_px == (2, 1)
    assert "2.000" in page.position_value.text()
    # The UV map and the live 2-D image are visible in the same row: selecting one
    # used to hide the other behind a tab.
    assert page.flat_view.isVisible() and page.live_view.isVisible()
    window.close()


def test_the_session_row_opens_the_folder_the_summary_and_the_path(
    make_window, tmp_path, monkeypatch
):
    """The operator can reach the recording without copying a path by hand."""
    from PySide6.QtWidgets import QApplication
    from orisys_app_shared.i18n import t

    from evidence_helpers import multipoint_session
    from orisys_autocal.summary import build_summary
    from orisys_autocal.summary_export import publish_summary
    from orisys_autocal.ui import calibration_panel as panel_module

    window = make_window()
    panel = window.calibration_panel
    opened: list = []
    monkeypatch.setattr(
        panel_module, "reveal", lambda path, select=None: opened.append((path, select)) or True
    )

    # Nothing has been recorded yet: each action is disabled and says why.
    panel._refresh()
    assert not panel.open_session_button.isEnabled()
    assert not panel.open_summary_button.isEnabled()
    assert not panel.copy_path_button.isEnabled()
    assert panel.open_session_button.toolTip() == t("cap.block.no_directory")
    assert panel.open_summary_button.toolTip() == t("cap.block.no_summary")

    root = multipoint_session(tmp_path / "session", sites=("P1",), repeats=1, levels=(5.0,))
    publish_summary(root, build_summary(root))
    # The page finds the session the campaign is recording into.
    panel.campaign.directory = root
    panel._refresh()

    assert panel._session_directory() == root
    assert panel.open_session_button.isEnabled()
    assert panel.open_summary_button.isEnabled()
    assert panel.copy_path_button.isEnabled()
    assert panel.open_session_button.toolTip() == t("cap.action.open_session")

    panel.open_session_button.click()
    assert opened[-1][0] == root
    panel.open_summary_button.click()
    assert opened[-1][0] == root / "summary"
    panel.copy_path_button.click()
    assert QApplication.clipboard().text() == str(root)
    window.close()