ykcheungab
465 lines
14 KiB
Python
465 lines
14 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
|
|
|
|
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
|
|
|
|
|
|
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 = False
|
|
self._preview_open = False
|
|
self._session_active = False
|
|
self.notes: list[str] = []
|
|
self.frames: 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
|
|
|
|
# reads
|
|
def _in_replay_view(self) -> bool:
|
|
return self.view == "replay"
|
|
|
|
def _current_source_kind(self) -> str:
|
|
return "recompute"
|
|
|
|
# writes
|
|
def _apply_view_preset(self, preset: str) -> None:
|
|
self.presets.append(preset)
|
|
self.view = preset
|
|
|
|
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_image_frame(self, frame) -> None:
|
|
self.frames.append(frame)
|
|
|
|
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_run(self) -> None:
|
|
self.started += 1
|
|
|
|
def _stop_run(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
|
|
|
|
|
|
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 video_path(self):
|
|
return self._path
|
|
|
|
def video_frame_bgr(self):
|
|
return self._frame
|
|
|
|
def video_message(self) -> str:
|
|
return "no frame"
|
|
|
|
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 _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 = ReplayController(
|
|
engine=engine,
|
|
host=host,
|
|
toolbar=toolbar,
|
|
charts=charts,
|
|
image_panel=panel,
|
|
mesh_panel=mesh,
|
|
)
|
|
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_runs_a_recompute_pass_instead_of_playing(qt_app):
|
|
"""With the recompute switch on, play starts the *pipeline*, not 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.recompute_running
|
|
assert host.started == 1
|
|
assert not controller.is_playing(), "the recompute pass also started playback"
|
|
|
|
controller.on_transport_toggled(False)
|
|
assert not controller.recompute_running
|
|
assert host.stopped == ["user_stop"]
|
|
|
|
|
|
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()
|
|
|
|
host.view = "live"
|
|
assert not controller.owns_image(), "a live view does not hand the pane to a replay"
|
|
|
|
host.view = "replay"
|
|
controller.recompute_running = True
|
|
assert not controller.owns_image(), "the pass is drawing, not the replay"
|
|
|
|
|
|
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)
|
|
toolbar = parts[0]
|
|
|
|
host.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"
|
|
|
|
host.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)
|
|
|
|
|
|
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
|
|
|
|
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]
|
|
|
|
host.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")
|
|
|
|
host.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
|