ykcheungab
fbf808d878
- What: 新增 apps/studio/src/orisys_studio/replay_controller.py：回放视图的**策略** （哪些入口能打开视频、transport 何时是播放何时是采集、从视频重算对面板与采集 管线做什么）从 app_window 搬进 ReplayController，共 30 个公开方法。replay.py 仍是 **引擎**（reader / 游标 / 异步深度重算），本模块是它外面的状态机。 它直接驱动的对象只有五个：toolbar 的回放控件、曲线图（时间轴）、2D 面板、三维面板的 来源标签、五个菜单项；其余依赖用 ReplayHost Protocol 写明（状态栏、轨道、视图切换、 显示单位、2D 唯一入口 _show_image_frame、采集管线 _start_run/_stop_run）——这些是 窗口级操作，所以留在窗口，并由 Protocol 把边界写清楚，而不是让控制器伸手进"上帝对象"。 - What: 播放时钟与定时器（33 ms）、_recompute_running、入口校验、菜单启用与提示、 重算开关的全部规则都随之上移；窗口侧保留过渡 shim（_load_recording / _load_replay_video / _render_replay_frame / _start_playback / _pause_playback / _on_play_shortcut / _stop_replay_activity / _replay_owns_image / _can_recompute / _play_timer / _recompute_running）供 tests、tools/ui_check.py、tools/ui_snapshot.py 继续驱动窗口。 窗口内部调用点（_apply_view_preset、_update_run_enabled、_sync_menu_checks、_on_run_toggled、 _on_stop_shortcut、_on_reset、菜单与工具栏连线）改为直接调用 self.replay_ctl.*。 删除窗口里已搬走的 _on_browse_video、_on_replay_position、_on_recording_changed、 _wire_replay、_on_recompute_toggled 等，以及随之为空的导入（time、QFileDialog、 SOURCE_REPLAY、PLAYBACK_TICK_MS）。顺带删掉一个从未接线的死方法 _on_replay_unit_toggled（与 _on_chart_unit_toggle 重复，无人调用）。 - What: 新增 apps/studio/tests/test_replay_controller.py：8 个用例，用假的 host/engine/ panel 直接测策略——播放在末尾从头开始、零长录制不播、playhead 跟墙钟并在末尾停下、 重算开关打开时 transport 启动管线而非播放、2D 面板归属、裸视频被接受/在实时视图被拒、 无标定录制回落到 raw、回放入口只在回放视图存在。 - Fix: 重算结束的 flag 由它自己的所有者清除（transport 暂停、开关关闭、stop_activity 三条路径），不再依赖窗口的 _stop_run 顺手清；效果是"重算结束后停止按钮正确变为不可用"， test_ui_smoke 里那个"手动补状态"的注释随之删除。 - Why: 回放是 app_window 的第二个独立变更驱动：录制格式变一次是引擎的事，操作员流程 变一次不该动 2500 行窗口。搬完之后回放策略可单独阅读、单独测，窗口只剩装配与转发。 - Verification: pytest shared/tests apps/studio/tests（offscreen）611 passed / 0 failed / 15 skipped（原 603 + 新增 8）；apps/studio/tools/ui_check.py 587 项检查 0 失败； .repro/locale_check.py 0 缺失。行为等价（搬运 + 一层转发）。 - Not in this change: C1 视图排列、C2 显示偏好、标定交互仍未拆分；docs 的行数/方法数 指标待全部切片落定后统一刷新。真机 GL 冒烟仍无法复跑（本机创建不出 GL 上下文， eglError 3005），13 例跳过、probe 用例按机器实况失败，与本次改动无关。
466 lines
14 KiB
Python
466 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
|