"""Acceptance for the replay view's policy, on its own.

``ReplayController`` is the replay view's state machine: what is open, what plays, and
what 「从视频重算」 does to the panels and to the run pipeline. It talks to the window
through the small ``ReplayHost`` interface and to the panels through their own, so both
can be stood in for. That is the point of the split: this policy used to be reachable
only by building a whole ``StudioMainWindow``, and now it is a unit.

The engine is faked here as well (the real one is covered by ``test_playback`` and the
window-level tests): these are the *rules*, not the decoder.
"""

from __future__ import annotations

import pytest
from orisys_app_shared.i18n import t
from orisys_studio import replay_controller
from orisys_studio.replay_controller import ReplayController

# --------------------------------------------------------------------- fakes


class _Signal:
    def __init__(self) -> None:
        self.slots: list = []

    def connect(self, slot) -> None:
        self.slots.append(slot)


class _Action:
    def __init__(self, checked: bool = False) -> None:
        self._checked = bool(checked)
        self.enabled: bool | None = None
        self.tooltips: list[str] = []
        self.triggered = _Signal()
        self.toggled = _Signal()

    def isChecked(self) -> bool:  # noqa: N802 - Qt's spelling
        return self._checked

    def setChecked(self, value: bool) -> None:  # noqa: N802
        self._checked = bool(value)

    def setEnabled(self, value: bool) -> None:  # noqa: N802
        self.enabled = bool(value)

    def setToolTip(self, text: str) -> None:  # noqa: N802
        self.tooltips.append(str(text))

    def blockSignals(self, flag: bool) -> None:  # noqa: N802
        pass


class _Button:
    def __init__(self) -> None:
        self.checked = False
        self.enabled = True

    def isChecked(self) -> bool:  # noqa: N802
        return self.checked

    def isEnabled(self) -> bool:  # noqa: N802
        return self.enabled

    def setChecked(self, value: bool) -> None:  # noqa: N802
        self.checked = bool(value)


class _Toolbar:
    def __init__(self, *, video_path: str = "", run_role: str = "playback") -> None:
        self.btn_run = _Button()
        self._video_path = video_path
        self._run_role = run_role
        self.run_checked: list[bool] = []
        self.refreshed = 0
        self.source_modes: list[str] = []
        self.recompute_enabled: list[bool] = []
        self.recompute_checked: list[bool] = []

    def run_role(self) -> str:
        return self._run_role

    def video_path(self) -> str:
        return self._video_path

    def set_video_path(self, path: str) -> None:
        self._video_path = str(path)

    def set_run_checked(self, value: bool) -> None:
        self.run_checked.append(bool(value))

    def refresh_labels(self) -> None:
        self.refreshed += 1

    def set_source_mode(self, mode: str) -> None:
        self.source_modes.append(str(mode))

    def set_recompute_enabled(self, value: bool) -> None:
        self.recompute_enabled.append(bool(value))

    def set_recompute_checked(self, value: bool) -> None:
        self.recompute_checked.append(bool(value))


class _Charts:
    def __init__(self, *, unit: str = "N", recording: bool = False) -> None:
        self._unit = unit
        self._recording = recording
        self.cursors: list[float] = []
        self.followed = 0
        self.shown = 0
        self.cleared = 0
        self.unit_fallbacks = 0

    def unit(self) -> str:
        return self._unit

    def set_cursor(self, position: float) -> None:
        self.cursors.append(float(position))

    def set_recording_source(self, reader) -> None:
        self._recording = True

    def clear_recording_source(self) -> None:
        self.cleared += 1
        self._recording = False

    def is_recording_source(self) -> bool:
        return self._recording

    def follow_latest(self) -> None:
        self.followed += 1

    def show_all(self) -> None:
        self.shown += 1

    def show_unit_fallback(self) -> None:
        self.unit_fallbacks += 1


class _ImagePanel:
    def __init__(self) -> None:
        self.kinds: list[str] = []
        self.cleared: list[str] = []

    def set_source_kind(self, kind: str) -> None:
        self.kinds.append(str(kind))

    def clear_preview(self, message: str) -> None:
        self.cleared.append(str(message))


class _Mesh:
    def __init__(self) -> None:
        self.kinds: list[str] = []
        self.reference_only = 0

    def set_source_kind(self, kind: str) -> None:
        self.kinds.append(str(kind))

    def show_reference_only(self) -> None:
        self.reference_only += 1


class _Host:
    """The window as the controller sees it: the host protocol, and nothing else."""

    def __init__(self, *, view: str = "replay") -> None:
        self.prefs: dict = {}
        self.view = view
        self._running_flag = False
        self._preview_open_flag = False
        self._session_active_flag = False
        self.notes: list[str] = []
        self.frames: list = []
        self.analysis_packets: list = []
        self.presets: list[str] = []
        self.units: list[str] = []
        self.source_modes: list[str] = []
        self.started = 0
        self.stopped: list[str] = []
        self.run_updates = 0
        self.rail_updates = 0
        self.markers = 0
        #: What the setup prompt answers for a bare video: a workspace directory, or
        #: ``None`` for "the operator declined". The window builds a real dialog; a fake
        #: has no dialog, so the answer is the field a test sets.
        self.workspace = None
        self.workspace_requests: list = []
        self.workspace_kwargs: dict = {}
        self.override_requests: list = []

    # AcquireGate reads (legacy private names kept as writeable fields for tests)
    @property
    def running(self) -> bool:
        return self._running_flag

    @running.setter
    def running(self, value: bool) -> None:
        self._running_flag = bool(value)

    @property
    def preview_open(self) -> bool:
        return self._preview_open_flag

    @preview_open.setter
    def preview_open(self, value: bool) -> None:
        self._preview_open_flag = bool(value)

    @property
    def session_active(self) -> bool:
        return self._session_active_flag

    @session_active.setter
    def session_active(self, value: bool) -> None:
        self._session_active_flag = bool(value)

    # compatibility for tests that still assign ``_running`` / friends
    @property
    def _running(self) -> bool:
        return self._running_flag

    @_running.setter
    def _running(self, value: bool) -> None:
        self._running_flag = bool(value)

    @property
    def _preview_open(self) -> bool:
        return self._preview_open_flag

    @_preview_open.setter
    def _preview_open(self, value: bool) -> None:
        self._preview_open_flag = bool(value)

    @property
    def _session_active(self) -> bool:
        return self._session_active_flag

    @_session_active.setter
    def _session_active(self, value: bool) -> None:
        self._session_active_flag = bool(value)

    def current_source_kind(self) -> str:
        return "recompute"

    def bare_video_workspace(self, video, **kwargs):
        self.workspace_requests.append(video)
        self.workspace_kwargs = dict(kwargs)
        return self.workspace

    def recording_analysis_overrides(self, recording):
        self.override_requests.append(recording)
        return False

    def analysis_service(self, recording, cancelled=None, *, on_progress=None):
        raise AssertionError("the test did not provide an analysis service")

    def set_display_unit(self, unit: str) -> None:
        self.units.append(unit)

    def set_source_mode(self, mode: str) -> None:
        self.source_modes.append(mode)

    def show_frame(self, frame) -> None:
        self.frames.append(frame)

    def show_analysis_packet(self, packet) -> None:
        self.deliver_measurement(packet, advance="flush", paint_2d=False)

    def deliver_measurement(
        self,
        packet,
        *,
        advance: str = "tick",
        arrow_policy=None,
        image_source: str | None = None,
        paint_2d: bool = True,
    ) -> None:
        if paint_2d:
            from orisys_app_shared.acquisition.image_backend import ImageFrame

            self.frames.append(
                ImageFrame(
                    getattr(packet, "processed_frame", None),
                    flow=getattr(packet, "flow", None),
                    depth=getattr(packet, "depth", None),
                    valid_mask=getattr(packet, "valid_mask", None),
                    source=image_source or "replay",
                    frame_id=int(getattr(packet, "frame_id", 0) or 0),
                    arrow_policy=arrow_policy,
                    result_available=True,
                )
            )
        self.analysis_packets.append(packet)

    def drop_render_marker(self) -> None:
        self.markers += 1

    def notify(self, text: str, ms: int = 0) -> None:
        self.notes.append(str(text))

    def start_acquire(self) -> None:
        self.started += 1

    def stop_acquire(self, *, reason: str = "user_stop", wait: bool = False) -> None:
        self.stopped.append(reason)

    def update_run_enabled(self) -> None:
        self.run_updates += 1

    def refresh_status_rail(self) -> None:
        self.rail_updates += 1

    def persist_prefs(self) -> None:
        pass


class _Reader:
    def __init__(self, *, newtons=None) -> None:
        self._newtons = newtons

    def segments(self):
        return []

    def info(self):
        return type("Info", (), {"newtons": self._newtons})()


class _Engine:
    def __init__(
        self,
        *,
        duration: float = 1.0,
        position: float = 0.0,
        has_video: bool = False,
        loaded: bool = False,
        video_only: bool = False,
    ) -> None:
        self._duration = duration
        self._position = position
        self._has_video = has_video
        self._loaded = loaded
        self._video_only = video_only
        self._reader = None
        self._path = None
        self._frame = None
        self.positions: list[float] = []
        self.opened: list = []
        self.units: list[str] = []
        self.accept = True

    # state
    def has_video(self) -> bool:
        return self._has_video

    def is_loaded(self) -> bool:
        return self._loaded

    def is_video_only(self) -> bool:
        return self._video_only

    def duration_s(self) -> float:
        return self._duration

    def position(self) -> float:
        return self._position

    def set_position(self, position: float) -> None:
        self._position = float(position)
        self.positions.append(float(position))

    def reader(self):
        return self._reader

    def directory(self):
        """No recording directory: a stub opens a video, not a recorded session."""
        return None

    def video_path(self):
        return self._path

    def video_frame_bgr(self):
        return self._frame

    def video_message(self) -> str:
        return "no frame"

    def analysis(self):
        """No stored result in this stub: the derived views fall back to the video."""
        return None

    def set_unit(self, unit: str) -> None:
        self.units.append(str(unit))

    # entries
    def open_video(self, path) -> bool:
        self.opened.append(("video", path))
        if self.accept:
            self._has_video = True
            self._video_only = True
            self._path = path
        return self.accept

    def open_recording(self, path) -> bool:
        self.opened.append(("recording", path))
        if self.accept:
            self._has_video = True
            self._loaded = True
        return self.accept


def _wire_replay(controller, host):
    """Signals replace private start/stop/view calls on the shell."""
    controller.view_requested.connect(
        lambda preset: (
            host.presets.append(preset),
            setattr(host, "view", preset),
            controller.set_active_view(preset),
        )
    )
    controller.acquire_requested.connect(host.start_acquire)
    controller.acquire_stop_requested.connect(
        lambda reason: host.stop_acquire(reason=reason)
    )
    controller.set_active_view(host.view)
    return controller


def _controller(**kwargs):
    """A controller wired to fakes; returns ``(controller, host, engine, parts)``."""
    engine = kwargs.pop("engine", None) or _Engine()
    host = kwargs.pop("host", None) or _Host()
    toolbar = kwargs.pop("toolbar", None) or _Toolbar()
    charts = kwargs.pop("charts", None) or _Charts()
    panel = kwargs.pop("image_panel", None) or _ImagePanel()
    mesh = kwargs.pop("mesh_panel", None) or _Mesh()
    controller = _wire_replay(
        ReplayController(
            engine=engine,
            host=host,
            toolbar=toolbar,
            charts=charts,
            image_panel=panel,
            mesh_panel=mesh,
        ),
        host,
    )
    actions = {
        "open_video": _Action(),
        "playback": _Action(),
        "latest": _Action(),
        "play": _Action(),
        "recompute": _Action(),
    }
    controller.bind_actions(**actions)
    return controller, host, engine, (toolbar, charts, panel, mesh, actions)


# --------------------------------------------------------------------- tests


def test_starting_playback_at_the_end_starts_from_the_top(qt_app):
    engine = _Engine(duration=2.0, position=2.0, has_video=True)
    controller, host, engine, parts = _controller(engine=engine)
    toolbar = parts[0]

    controller.start_playback()
    assert engine.positions[0] == 0.0, "playing at the end did not rewind"
    assert controller.is_playing()
    assert toolbar.run_checked[-1] is True
    controller.pause_playback()


def test_a_zero_length_recording_is_not_something_to_play(qt_app):
    """Silently un-checking the button is how "pressing 回放 does nothing" felt."""
    engine = _Engine(duration=0.0, has_video=True)
    controller, host, engine, parts = _controller(engine=engine)
    toolbar = parts[0]

    controller.start_playback()
    assert not controller.is_playing()
    assert toolbar.run_checked[-1] is False
    assert t("playback.empty_recording") in host.notes


def test_the_playhead_follows_the_wall_clock_and_stops_at_the_end(qt_app, monkeypatch):
    engine = _Engine(duration=0.4, position=0.0, has_video=True)
    controller, host, engine, parts = _controller(engine=engine)
    clock = {"now": 100.0}
    monkeypatch.setattr(replay_controller.time, "perf_counter", lambda: clock["now"])

    controller.start_playback()
    clock["now"] = 100.1
    controller.on_play_tick()
    assert engine.position() == pytest.approx(0.1, abs=1e-6)

    clock["now"] = 100.9  # past the end
    controller.on_play_tick()
    assert engine.position() == pytest.approx(0.4, abs=1e-6)
    assert not controller.is_playing(), "the transport kept running past the end"


def test_the_transport_plays_even_when_the_recompute_switch_is_on(qt_app):
    """Analysis is a separate pass; the rail's play button always moves the cursor."""
    engine = _Engine(duration=1.0, has_video=True)
    controller, host, engine, parts = _controller(engine=engine)
    parts[4]["recompute"].setChecked(True)

    controller.on_transport_toggled(True)
    assert controller.is_playing(), "play must scrub the recording"
    assert host.started == 0, "play must not open an acquisition chain"

    controller.on_transport_toggled(False)
    assert not controller.is_playing()
    assert host.stopped == []


def test_the_replay_policy_owns_the_image_only_in_the_replay_view(qt_app):
    controller, host, engine, parts = _controller(
        engine=_Engine(has_video=True, loaded=True)
    )
    assert controller.owns_image()

    controller.set_active_view("live")
    assert not controller.owns_image(), "a live view does not hand the pane to a replay"

    controller.set_active_view("replay")
    # An analysis pass is drawn by this controller too, so ownership stays with replay.
    assert controller.owns_image()


def test_a_bare_video_is_opened_as_a_video_and_refused_outside_the_view(qt_app, tmp_path):
    bare = tmp_path / "clip.mp4"
    bare.write_bytes(b"not really a video")
    engine = _Engine()
    controller, host, engine, parts = _controller(engine=engine)

    controller.set_active_view("live")
    assert controller.load_video(bare) is False
    assert host.notes[-1] == t("toolbar.replay_view_only")
    assert engine.opened == [], "a video was opened from the live view"

    controller.set_active_view("replay")
    assert controller.load_video(bare) is True, "a playable clip was refused"
    assert engine.opened == [("video", bare)]
    assert host.source_modes[-1] == "video"
    assert host.prefs["video_path"] == str(bare)
    # Opening a bare video asks how to analyse it — once, after it is on screen. The answer
    # here is "declined", and that must leave today's behaviour intact: the video plays,
    # and no workspace is claimed as this timeline's input.
    assert host.workspace_requests == [bare], "the setup prompt was never asked"
    assert controller.recording_directory() is None
    assert controller.result_state()[1] == t("playback.analysis_setup_declined")


def test_a_bare_video_uses_the_workspace_the_prompt_answers_with(qt_app, tmp_path):
    """The generated workspace *is* the recording directory for everything downstream."""
    bare = tmp_path / "clip.mp4"
    bare.write_bytes(b"not really a video")
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    (workspace / "session.json").write_text("{}", encoding="utf-8")
    host = _Host()
    host.workspace = workspace
    controller, host, engine, parts = _controller(host=host)

    assert controller.load_video(bare) is True
    assert host.workspace_requests == [bare]
    assert controller.recording_directory() == workspace, (
        "the pass would have read the video's own folder instead of the workspace"
    )

    # Dropping what is open must drop the claim with it: the workspace is keyed by the
    # video, so reopening finds it, but it is not this timeline's input any more.
    controller.on_recording_changed(False)
    assert controller.recording_directory() is None


def test_the_charts_fall_back_to_raw_when_a_recording_has_no_newtons(qt_app):
    engine = _Engine(loaded=True, has_video=True)
    engine._reader = _Reader(newtons=None)
    controller, host, engine, parts = _controller(
        engine=engine, charts=_Charts(unit="N")
    )

    controller.apply_source(True)
    assert engine.units[-1] == "raw"
    assert host.units[-1] == "raw"
    assert t("status.units_raw_fallback") in host.notes
    assert parts[1].unit_fallbacks == 1

    engine._reader = _Reader(newtons=[1.0, 2.0, 3.0])
    controller.apply_source(True)
    assert engine.units[-1] == "N", "a calibrated recording must keep newtons"


def test_the_replay_entries_exist_only_in_the_replay_view(qt_app):
    controller, host, engine, parts = _controller()
    actions = parts[4]

    controller.set_active_view("live")
    controller.sync_actions()
    assert [
        actions[name].enabled for name in ("open_video", "playback", "latest")
    ] == [False, False, False]
    # The recompute *button* on the rail is what this syncs (the menu entry follows the
    # engine's state in ``refresh_state``).
    assert parts[0].recompute_enabled[-1] is False
    assert actions["open_video"].tooltips[-1] == t("toolbar.replay_view_only")

    controller.set_active_view("replay")
    engine._has_video = True
    controller.sync_actions()
    assert actions["open_video"].enabled is True
    assert actions["open_video"].tooltips[-1] == ""
    assert parts[0].recompute_enabled[-1] is True


def test_analyze_opens_a_remembered_video_that_is_not_loaded_yet(qt_app, tmp_path):
    """A toolbar path alone used to notify ``playback.no_video`` and skip the dialog.

    The rail can show a file name from a previous session while the engine has not opened
    anything. 「分析视频」 must call ``load_video`` so the setup prompt appears — the same
    path as picking the file again in the explorer.
    """
    bare = tmp_path / "1.mp4"
    bare.write_bytes(b"not really a video")
    engine = _Engine(has_video=False, video_only=False)
    toolbar = _Toolbar(video_path=str(bare))
    controller, host, engine, parts = _controller(engine=engine, toolbar=toolbar)

    controller.on_recompute_toggled(True)
    assert engine.opened == [("video", bare)], "the remembered file was not opened"
    assert host.workspace_requests == [bare], "the setup dialog was not shown"
    assert t("playback.no_video") not in host.notes