ykcheungab
637 lines
20 KiB
Python
637 lines
20 KiB
Python
"""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
|