From fbf808d878ba5c43faa2a9b27d532dfa798fe87a Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Tue, 15 Sep 2026 09:38:40 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8B=86=E5=88=86=20B=EF=BC=9A=E5=9B=9E?= =?UTF-8?q?=E6=94=BE=E7=AD=96=E7=95=A5=E7=A7=BB=E5=85=A5=20ReplayControlle?= =?UTF-8?q?r=EF=BC=88app=5Fwindow=202837=E2=86=922499=20=E8=A1=8C=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 用例按机器实况失败，与本次改动无关。 --- apps/studio/src/orisys_studio/app_window.py | 516 +++----------- .../src/orisys_studio/replay_controller.py | 641 ++++++++++++++++++ apps/studio/tests/test_replay_controller.py | 465 +++++++++++++ apps/studio/tests/test_ui_smoke.py | 10 +- 4 files changed, 1201 insertions(+), 431 deletions(-) create mode 100644 apps/studio/src/orisys_studio/replay_controller.py create mode 100644 apps/studio/tests/test_replay_controller.py diff --git a/apps/studio/src/orisys_studio/app_window.py b/apps/studio/src/orisys_studio/app_window.py index 3bfdc62..0f4ecd5 100644 --- a/apps/studio/src/orisys_studio/app_window.py +++ b/apps/studio/src/orisys_studio/app_window.py @@ -5,7 +5,6 @@ from __future__ import annotations import logging import subprocess import sys -import time from pathlib import Path from typing import Any @@ -16,7 +15,6 @@ from PyQt5.QtWidgets import ( QAction, QActionGroup, QApplication, - QFileDialog, QFrame, QHBoxLayout, QLabel, @@ -34,7 +32,6 @@ from orisys_app_shared.acquisition.image_backend import ( SOURCE_LIVE, SOURCE_PREVIEW, SOURCE_RECOMPUTE, - SOURCE_REPLAY, ImageFrame, plan_image, render_image, @@ -48,6 +45,7 @@ from orisys_app_shared.record import RecordConfig from orisys_app_shared.record.sample import to_record_sample from orisys_studio.recording import RecordingCoordinator, record_config_from_prefs from orisys_studio.replay import ReplayEngine +from orisys_studio.replay_controller import ReplayController from orisys_studio import session_metadata from orisys_studio.session_controller import SessionController from orisys_studio.scene3d.prefs import settings_to_prefs @@ -114,10 +112,6 @@ VIEW_SHOWS_3D = ("custom", "live", "replay") #: nothing. The table describes the two-dimensional layout; the pane is optional, and #: "not in the table" must not mean "standing at its minimum width". DEFAULT_3D_PANE_WIDTH = 340 -#: Playback tick while a recording is playing. 33 ms is one frame at 30 fps; the -#: cursor is advanced from the wall clock, so a missed tick does not slow the -#: recording down. -PLAYBACK_TICK_MS = 33 #: Kept under its historical name for ``tools/ui_snapshot.py``, which imports it from @@ -164,8 +158,8 @@ class StudioMainWindow(QMainWindow): self._view_preset = preset if preset in VIEW_PRESETS else "custom" self._show_finger_3d = bool(self.prefs.get("show_finger_3d", True)) self._show_fps = bool(self.prefs.get("show_fps", False)) - #: True while the run over the video of the recompute mode is going. - self._recompute_running = False + # Whether a recompute pass is running is the replay controller's state; the + # window reads and writes it through the ``_recompute_running`` shim below. self._preview = None self._preview_timer = QTimer(self) self._preview_timer.setInterval(33) @@ -174,9 +168,9 @@ class StudioMainWindow(QMainWindow): self._last_rendered_view_mode: str | None = None # ---- replay --------------------------------------------------------- # One engine for the history view: the recording reader, the cursor and - # the asynchronous depth recomputation. It renders nothing itself; the - # transport policy around it will move into a ReplayController (see - # docs/architecture/studio-architecture.md, suggestion 1). + # the asynchronous depth recomputation. It renders nothing itself; the policy + # around it (transport, entry points, the recompute pass) lives in + # ``ReplayController``, which is built once the panels it drives exist. self.replay = ReplayEngine(self) # ---- record channel ------------------------------------------------- # Fed from the acquisition thread via the worker's sink list, so it does @@ -267,11 +261,18 @@ class StudioMainWindow(QMainWindow): # under the panels: the two views share one layout, and the transport is # the rail's primary button — 「运行」while measuring, 「回放」while # replaying (see ``_on_run_toggled`` / ``toolbar.set_run_role``). - self._play_timer = QTimer(self) - self._play_timer.setInterval(PLAYBACK_TICK_MS) - self._play_timer.timeout.connect(self._on_play_tick) - self._play_started_at = 0.0 - self._play_from = 0.0 + # The replay policy needs the panels (built above) and the menus (built below): + # it is created here, between the two. + self.replay_ctl = ReplayController( + engine=self.replay, + host=self, + toolbar=self.toolbar, + charts=self.force_plots, + image_panel=self.arrow_panel, + mesh_panel=self.mesh3d_panel, + parent=self, + ) + self.replay_ctl.wire() self.setStatusBar(_NoticeBar()) # The status bar carries **one-off notices** only (a recording finished, a @@ -285,7 +286,6 @@ class StudioMainWindow(QMainWindow): self._build_menus() self._wire_toolbar() self._wire_sensor_picker() - self._wire_replay() self._restore_prefs_into_ui() # The 3D pane's sensor and its persisted display settings, before anything # asks it to draw: the pane resolves its asset from the sensor, and restoring @@ -687,8 +687,8 @@ class StudioMainWindow(QMainWindow): self.toolbar.camera_id_changed.connect(self._on_camera_id_changed) self.toolbar.sensor_clicked.connect(self._open_sensor_dialog) self.toolbar.use_camera_clicked.connect(self._on_use_camera) - self.toolbar.replay_video_clicked.connect(self._on_open_recording) - self.toolbar.recompute_clicked.connect(self._on_recompute_button) + self.toolbar.replay_video_clicked.connect(self.replay_ctl.on_open_recording) + self.toolbar.recompute_clicked.connect(self.replay_ctl.on_recompute_button) self.toolbar.run_toggled.connect(self._on_run_toggled) self.toolbar.record_toggled.connect(self._on_record_toggled) self.toolbar.reset_clicked.connect(self._on_reset) @@ -699,390 +699,68 @@ class StudioMainWindow(QMainWindow): self.sensor_picker.sensor_type_changed.connect(self._on_sensor_type_changed) self.sensor_picker.config_changed.connect(self._on_config_changed) self.arrow_panel.view_mode_changed.connect(self._on_image_view_mode_changed) - self.arrow_panel.view_mode_changed.connect(lambda _mode: self._render_replay_frame()) + self.arrow_panel.view_mode_changed.connect(lambda _mode: self.replay_ctl.render_frame()) # The unit chip on the chart header switches the same unit everywhere. self.force_plots.unit_toggle_requested.connect(self._on_chart_unit_toggle) # The gear on the chart header opens the same dialog as 视图 ▸ 曲线图设置. self.force_plots.plot_settings_requested.connect(self._on_plot_settings) - def _wire_replay(self) -> None: - """The controller feeds the views; the chart is the only timeline.""" - self.replay.position_changed.connect(self._on_replay_position) - self.replay.frames_changed.connect(self._render_replay_frame) - self.replay.status_changed.connect(self._refresh_replay_state) - self.replay.recording_changed.connect(self._on_recording_changed) - # Dragging the shared cursor on the charts scrubs the recording too. - self.force_plots.cursor_moved.connect(self._on_cursor_moved) - self.force_plots.timeline_taken_over.connect(self._pause_playback) - def _on_cursor_moved(self, t_rel_s: float) -> None: - if self._recompute_running: - # The recompute pass owns the timeline: its samples are arriving in - # order and the cursor follows them. - return - if self._view_preset == "replay" and self.replay.has_video(): - self.replay.set_position(float(t_rel_s)) + """The charts' shared cursor scrubs the recording (through the controller).""" + self.replay_ctl.on_cursor_moved(t_rel_s) - def _on_open_recording(self) -> None: - """Choose a recorded video and switch to the view that replays it. - - The recording a video belongs to is its own directory, so the picker - takes the *video* (what the operator recognises) and the charts pair it - with the ``forces.csv`` written beside it. - """ - start = str(recordings_dir()) - chosen, _filter = QFileDialog.getOpenFileName( - self, - t("playback.open"), - start, - t("playback.video_filter"), - ) - if not chosen: - return - self._load_replay_video(Path(chosen)) - - def _on_recompute_button(self) -> None: - """The replay view's own entry for the processed picture (see ``recompute``).""" - self.act_recompute.setChecked(not self.act_recompute.isChecked()) - - def _sync_replay_actions(self) -> None: - """Replay entries exist in the replay view, and nowhere else. - - Opening a recording from the *live* view used to leave the window in a state - nothing described: the source became a video file, the 2D panel did not change - (the replay renderer only draws in the replay view) and the Run button acquired - from that file. The entries are therefore disabled outside the replay view, with - the reason in the tooltip, instead of quietly doing half of it. - """ - allowed = self._view_preset == "replay" - reason = "" if allowed else t("toolbar.replay_view_only") - for action in (self.act_open_video, self.act_playback, self.act_playback_latest): - action.setEnabled(allowed) - action.setToolTip(reason) - # The rail's entries are in the replay group already; the recompute button also - # has to follow whether there is anything to recompute *from*. - self.toolbar.set_recompute_enabled(allowed and self._can_recompute()) - self.toolbar.set_recompute_checked(self.act_recompute.isChecked()) + # -------------------------------------------------------- replay (forwards) + # The replay policy lives in ``ReplayController``. These forward so the call sites + # still in this file — and the tests / ``tools/ui_check`` / ``tools/ui_snapshot``, + # which drive the window — keep working while the rest of the split lands. Each is + # a one-line delegation; they are deleted as the call sites move to + # ``self.replay_ctl.*``. + def _load_recording(self, directory: Path) -> bool: + return self.replay_ctl.load_recording(directory) def _load_replay_video(self, video: Path) -> bool: - """Open what the chosen video *is*: a recording, or just a video. - - A **recording** is a directory that holds ``forces.csv`` beside the video — the - same rule ``list_recordings`` uses, so the two cannot disagree about what a - recording is. Anything else is still worth watching: an operator who points at - a clip expects the transport to play it, not to be told to press a different - button first. So a bare video is opened for playback (no forces, no session - metadata, and the rail says so), and only a file that cannot be decoded at all - is refused — with what is missing, and with 「从视频重算」 still offered, because - that is the only way to get *numbers* out of a video that has no recording. - """ - video = Path(video) - if self._view_preset != "replay": - # Defensive: the entries are disabled outside the replay view, so this only - # happens programmatically — and refusing is better than leaving the window - # with a video source and no replay renderer. - self._notify(t("toolbar.replay_view_only")) - return False - folder = video.parent if video.is_file() else video - if not (folder / "forces.csv").is_file(): - if video.is_file() and self.replay.open_video(video): - self.toolbar.set_video_path(str(video)) - self.prefs["video_path"] = str(video) - self._set_source_mode("video") - self._render_replay_frame() - self._notify(t("playback.video_only_opened", name=video.name)) - self._refresh_replay_state() - return True - if video.is_file(): - self.toolbar.set_video_path(str(video)) - self.prefs["video_path"] = str(video) - self._set_source_mode("video") - if not self.replay.has_video(): - # Nothing else is on screen, so the reason belongs in the panel that - # would have shown the frames — not only in a status line that expires. - self.arrow_panel.clear_preview(t("playback.video_only_panel")) - self._refresh_replay_state() - self._notify(t("playback.not_a_recording", name=video.name)) - return False - if not self._load_recording(folder): - return False - self.toolbar.set_video_path(str(video)) - self._refresh_status_rail() - return True - - def _can_recompute(self) -> bool: - """Whether 「从视频重算」 has something to read: a recording, or a video file.""" - if self._view_preset != "replay": - return False - if self.replay.has_video(): - return True - named = self.toolbar.video_path() - return bool(named) and Path(named).is_file() - - def _load_recording(self, directory: Path) -> bool: - if self._view_preset != "replay": - self._apply_view_preset("replay") - opened = self.replay.open_recording(Path(directory)) - if not opened: - self._notify(t("playback.open_failed", error=Path(directory).name)) - return False - self._refresh_replay_state() - return True - - def _open_latest_recording(self) -> None: - """Switch to the replay view on the most recent recording.""" - from orisys_app_shared.record import list_recordings - - if self._view_preset != "replay": - self._notify(t("toolbar.replay_view_only")) - return - found = list_recordings(recordings_dir()) - if not found: - self._notify(t("playback.none_found")) - return - self._load_recording(Path(found[-1])) - - def _on_recording_changed(self, opened: bool) -> None: - """A recording was opened (or dropped): re-point everything that reads it.""" - self._refresh_replay_state() - if not opened: - self._pause_playback() - return - if self._view_preset == "replay": - self._apply_replay_source(True) - self._render_replay_frame() - - def _refresh_replay_state(self) -> None: - """Push the engine's state into the charts' mode and the rail.""" - if self.replay.is_video_only() and not self.toolbar.video_path(): - path = self.replay.video_path() - if path is not None: - self.toolbar.set_video_path(str(path)) - elif self.replay.is_loaded() and not self.toolbar.video_path(): - # Opened without naming a video (the file menu's "latest"): name the - # segment that is on screen so the rail states the same thing the - # picker would. - reader = self.replay.reader() - segments = sorted(reader.segments()) if reader is not None else [] - if segments: - self.toolbar.set_video_path(str(segments[0])) - self.toolbar.set_run_checked(self._play_timer.isActive()) - # Loading a recording is what makes a recompute pass possible at all — but a - # chosen video file is enough to recompute *from*, even when it is not a - # recording: that is the way to look at a video that has no forces.csv. - self.act_recompute.setEnabled(self._can_recompute()) - self._sync_replay_actions() - self._update_run_enabled() - self._refresh_status_rail() - - def _on_replay_position(self, position_s: float) -> None: - self.force_plots.set_cursor(float(position_s)) - - def _on_replay_unit_toggled(self) -> None: - """The chart's unit chip drives the same unit the replay reads.""" - self._set_display_unit("raw" if self.force_plots.unit() == "N" else "N") - - # ------------------------------------------------------------- playback - def _on_transport_toggled(self, playing: bool) -> None: - """The rail's primary button in the replay view: play or pause.""" - if self._recompute_active(): - # A recompute pass is a *run*: play starts the pipeline over the - # video, pause stops it. It can only go forward (the chain is - # stateful), so playing again after the end starts from the top. - if playing: - self._recompute_running = True - self._start_run() - elif self._recompute_running: - self._stop_run(reason="user_stop") - self.toolbar.set_run_checked(bool(playing) and self._running) - return - if playing and self.replay.has_video(): - self._start_playback() - else: - self._pause_playback() - - def _start_playback(self) -> None: - duration = self.replay.duration_s() - if duration <= 0: - # A zero-length timeline is not something to play, and silently un-checking - # the button is how "pressing 回放 does nothing" felt to the operator. - self.toolbar.set_run_checked(False) - self._notify(t("playback.empty_recording")) - return - position = self.replay.position() - if position >= duration - 0.02: - # At the end: playing again means playing from the start. - self._set_replay_position(0.0) - position = 0.0 - self._play_from = float(position) - self._play_started_at = time.perf_counter() - self._play_timer.start() - self.toolbar.set_run_checked(True) - self.toolbar.refresh_labels() - # The stop button appears with the playback: one place decides what the rail - # offers, so it is told here rather than kept in step in two places. - self._update_run_enabled() - - def _pause_playback(self) -> None: - if self._play_timer.isActive(): - self._play_timer.stop() - if self.toolbar.btn_run.isChecked(): - self.toolbar.set_run_checked(False) - self.toolbar.refresh_labels() - self._update_run_enabled() - - def _on_play_tick(self) -> None: - """Advance the playhead from the wall clock, not from the tick count.""" - duration = self.replay.duration_s() - elapsed = time.perf_counter() - self._play_started_at - position = self._play_from + elapsed - if position >= duration: - self._set_replay_position(duration) - self._pause_playback() - return - self._set_replay_position(position) - - def _set_replay_position(self, position_s: float) -> None: - """Move the playhead without the position signal echoing back into it.""" - self.replay.set_position(float(position_s)) - - def _recompute_active(self) -> bool: - return bool(self.act_recompute.isChecked()) - - def _replay_video_segment(self) -> Path | None: - """The recorded video a recompute pass should read.""" - reader = self.replay.reader() - segments = sorted(reader.segments()) if reader is not None else [] - if segments: - return Path(segments[0]) - named = self.toolbar.video_path() - if named and Path(named).is_file(): - return Path(named) - return None - - def _on_recompute_toggled(self, checked: bool) -> None: - """Off: read the recorded CSV. On: re-run the pipeline over the video. - - The CSV is the session's own measurement. Re-running the vision chain - over the recorded (lossy) video answers a different question — "what - would this algorithm version say about that footage" — and it can only - run *forward*, because the chain is stateful, so it is a pass rather than - a scrub. - """ - wanted = bool(checked) and self._can_recompute() - if self.act_recompute.isChecked() != wanted: - self.act_recompute.blockSignals(True) - self.act_recompute.setChecked(wanted) - self.act_recompute.blockSignals(False) - self._pause_playback() - if wanted: - segment = self._replay_video_segment() - if segment is None: - self._notify(t("playback.no_video")) - self.act_recompute.blockSignals(True) - self.act_recompute.setChecked(False) - self.act_recompute.blockSignals(False) - return - # The charts stop reading the CSV; the pipeline's own samples become - # the data, and the 2D panel shows what it produces. The pass starts - # here so the switch is not a blank chart waiting for a second click. - self.force_plots.clear_recording_source() - self.arrow_panel.set_source_kind("recompute") - self.toolbar.set_source_mode("video") - self.toolbar.set_video_path(str(segment)) - self._refresh_status_rail() - self._recompute_running = True - # The 3D pane now says "recomputed", not "replayed": the pass is what - # feeds it, through the ordinary acquisition sink. - self.mesh3d_panel.set_source_kind(self._current_source_kind()) - # Playing is what starts the pass: one path in, so the button and the - # menu cannot start it twice. - self.toolbar.set_run_checked(True) - self.toolbar.refresh_labels() - self._on_transport_toggled(True) - return - if self._recompute_running: - self._stop_run(reason="user_stop") - # The pass is over, so the last recomputed frame must go: leaving it on - # screen would show a measurement nothing is producing any more. - self.mesh3d_panel.show_reference_only() - if self._view_preset == "replay": - self.mesh3d_panel.set_source_kind(self._current_source_kind()) - if self.replay.is_loaded(): - self._apply_replay_source(True) - # A bare video has no charts to point at the recording, but it does have - # frames to show again now that the pass is over. - self._render_replay_frame() + return self.replay_ctl.load_video(video) def _render_replay_frame(self) -> None: - """Draw what the 2D panel should show for the current replay cursor. + self.replay_ctl.render_frame() - A recording stores the raw camera image and nothing else, so the frame this - builds has no flow and no depth — the plan then says so, in one line, and the - panel offers exactly the one view the frame can be. The processed picture of the - same footage is what 「从视频重算」 produces, through the *same* backend. - """ - if self._view_preset != "replay" or not self.replay.has_video(): - return - if self._recompute_running: - # The pipeline is drawing its own frames (see ``_replay_owns_image``). - return - image = self.replay.video_frame_bgr() - if image is None: - self.arrow_panel.clear_preview(self.replay.video_message()) - return - self._show_image_frame( - ImageFrame( - image, - source=SOURCE_REPLAY, - t_rel_s=float(self.replay.position()), - ) - ) + def _start_playback(self) -> None: + self.replay_ctl.start_playback() - def _apply_replay_source(self, active: bool) -> None: - """Point the charts at the recording (or back at the live session).""" - if active and self.replay.is_loaded(): - reader = self.replay.reader() - self.force_plots.set_recording_source(reader) - # A recording is shown in a unit it actually carries: newton only when - # it has a calibrated axis somewhere, otherwise the charts would be - # empty and every card a dash. The operator's preference is kept when - # the recording can honour it. - wanted = "N" if self.force_plots.unit() == "N" else "raw" - unit = self._replay_unit_for(reader) - self.replay.set_unit(unit) - self._set_display_unit(unit) - if unit != wanted: - self._notify(t("status.units_raw_fallback")) - self.force_plots.show_all() - self.arrow_panel.set_source_kind("replay") - return - if self.force_plots.is_recording_source(): - self.force_plots.clear_recording_source() - self.force_plots.follow_latest() - self.arrow_panel.set_source_kind("live") - # The panel held a recorded frame: the next live packet must repaint it, - # so the "already rendered" marker is dropped. + def _pause_playback(self) -> None: + self.replay_ctl.pause_playback() + + def _on_play_shortcut(self) -> None: + self.replay_ctl.on_play_shortcut() + + def _stop_replay_activity(self) -> None: + self.replay_ctl.stop_activity() + + def _replay_owns_image(self) -> bool: + return self.replay_ctl.owns_image() + + def _can_recompute(self) -> bool: + return self.replay_ctl.can_recompute() + + def _in_replay_view(self) -> bool: + return self._view_preset == "replay" + + def _drop_render_marker(self) -> None: + """A recorded frame was on the panel: the next live packet must repaint it.""" self._last_rendered_view_mode = None - if not self._running and not self._preview_open: - self.arrow_panel.clear_preview(t("panels.arrow_empty")) - def _replay_unit_for(self, reader) -> str: - """The unit a recording can be shown in: raw unless it has newtons. + @property + def _play_timer(self): + return self.replay_ctl.play_timer + + @property + def _recompute_running(self) -> bool: + return self.replay_ctl.recompute_running + + @_recompute_running.setter + def _recompute_running(self, running: bool) -> None: + self.replay_ctl.recompute_running = running - ``raw`` is always present (it is what the device measured); the calibrated - triplet only exists for the axes the recording actually calibrated. Asking - for newton on a raw-only recording draws three empty charts, so the same - fallback the live view uses applies here. - """ - if self.force_plots.unit() != "N": - return "raw" - info = reader.info() - newtons = getattr(info, "newtons", None) - if newtons is None: - return "raw" - values = np.asarray(newtons, dtype=float) - return "N" if values.size and bool(np.isfinite(values).any()) else "raw" def _restore_prefs_into_ui(self) -> None: raw_id = str(self.prefs.get("camera_id") or "0").strip() @@ -1142,17 +820,14 @@ class StudioMainWindow(QMainWindow): self.file_menu = menu_bar.addMenu("") self.act_open_video = QAction(self) - self.act_open_video.triggered.connect(self._on_browse_video) self.file_menu.addAction(self.act_open_video) self.act_reveal_config = QAction(self) self.act_reveal_config.triggered.connect(self._on_reveal_config) self.file_menu.addAction(self.act_reveal_config) self.file_menu.addSeparator() self.act_playback = QAction(self) - self.act_playback.triggered.connect(self._on_open_recording) self.file_menu.addAction(self.act_playback) self.act_playback_latest = QAction(self) - self.act_playback_latest.triggered.connect(self._open_latest_recording) self.file_menu.addAction(self.act_playback_latest) # The transport is on the chart header; the space bar is the editing @@ -1160,7 +835,6 @@ class StudioMainWindow(QMainWindow): self.act_play = QAction(self) self.act_play.setShortcut("Space") self.act_play.setShortcutContext(Qt.ApplicationShortcut) - self.act_play.triggered.connect(self._on_play_shortcut) self.addAction(self.act_play) self.view_menu = menu_bar.addMenu("") @@ -1202,7 +876,6 @@ class StudioMainWindow(QMainWindow): self.act_recompute = QAction(self) self.act_recompute.setCheckable(True) self.act_recompute.setChecked(False) - self.act_recompute.toggled.connect(self._on_recompute_toggled) self.view_menu.addAction(self.act_recompute) self.view_menu.addSeparator() @@ -1260,6 +933,15 @@ class StudioMainWindow(QMainWindow): self.act_about.triggered.connect(self._show_about) self.help_menu.addAction(self.act_about) + # The replay entries exist now: hand them over, and the controller connects + # them (it owns what they mean — see ``ReplayController.bind_actions``). + self.replay_ctl.bind_actions( + open_video=self.act_open_video, + playback=self.act_playback, + latest=self.act_playback_latest, + play=self.act_play, + recompute=self.act_recompute, + ) self._sync_menu_checks() def _sync_menu_checks(self) -> None: @@ -1275,7 +957,7 @@ class StudioMainWindow(QMainWindow): self.act_finger_3d.blockSignals(True) self.act_finger_3d.setChecked(bool(self._show_finger_3d) if is_finger else False) self.act_finger_3d.blockSignals(False) - self.act_recompute.setEnabled(self._can_recompute()) + self.act_recompute.setEnabled(self.replay_ctl.can_recompute()) self._apply_arrow_rotation_for_sensor() def _current_source_kind(self) -> str: @@ -1386,7 +1068,7 @@ class StudioMainWindow(QMainWindow): # The rail shows the source this view needs, and only that source: a # camera while measuring, a recorded video while replaying. It is the one # visible difference between the two views. - self._pause_playback() + self.replay_ctl.pause_playback() self.toolbar.set_source_view("replay" if preset == "replay" else "live") # The rail's primary button is this view's main action: acquire while # measuring, play back while replaying; and the second button is that view's @@ -1400,16 +1082,16 @@ class StudioMainWindow(QMainWindow): # History browsing is the point of this view: the charts read the # open recording, and the chart cursor is its timeline. Without a # recording the charts stay on the live session and follow it. - self._apply_replay_source(True) - self._refresh_replay_state() + self.replay_ctl.apply_source(True) + self.replay_ctl.refresh_state() elif preset == "live": - self._apply_replay_source(False) + self.replay_ctl.apply_source(False) # Leaving the replay must not leave its video on the rail: the live # view is a camera, and the summary says so (see ``_refresh_status_rail``). - self._on_recompute_toggled(False) + self.replay_ctl.on_recompute_toggled(False) self.brand_bar.set_active_view(preset if preset != "custom" else None) self._sync_menu_checks() - self._sync_replay_actions() + self.replay_ctl.sync_actions() self._update_run_enabled() self._refresh_status_rail() if persist: @@ -1430,7 +1112,7 @@ class StudioMainWindow(QMainWindow): if self.replay.is_video_only() else "" ) - elif self._can_recompute(): + elif self.replay_ctl.can_recompute(): # A video was chosen but it is not a recording: the transport cannot # play it, and the tooltip says which button can show it instead of # leaving the operator pressing a dead button. @@ -1438,12 +1120,12 @@ class StudioMainWindow(QMainWindow): else: reason = t("toolbar.playback_disabled") self.toolbar.set_run_tooltip_reason(reason) - self.toolbar.set_run_enabled(loaded or self._play_timer.isActive()) + self.toolbar.set_run_enabled(loaded or self.replay_ctl.is_playing()) # The stop button is offered whenever there is something to stop; the # operator must never have to guess whether that is the case. if self.toolbar.reset_role() == "stop": self.toolbar.set_reset_enabled( - self._play_timer.isActive() + self.replay_ctl.is_playing() or self._recompute_running or self._session_active ) @@ -2206,8 +1888,8 @@ class StudioMainWindow(QMainWindow): from the keyboard. """ if self.toolbar.run_role() == "playback": - if self._play_timer.isActive() or self._session_active: - self._stop_replay_activity() + if self.replay_ctl.is_playing() or self._session_active: + self.replay_ctl.stop_activity() return if not self._session_active: return @@ -2225,26 +1907,6 @@ class StudioMainWindow(QMainWindow): self._refresh_status_rail() self._persist_prefs() - def _on_browse_video(self) -> None: - """Open a recorded video for replay (file menu). - - Choosing a video is the replay view's source selection, so this takes the - same path as the rail's button: a recorded video is something to *play*, - not a second way to acquire. - """ - if self._running or self._preview_open: - return - start = self.toolbar.video_path() or str(Path.home()) - path, _ = QFileDialog.getOpenFileName( - self, - t("menu.open_video"), - start, - t("playback.video_filter"), - ) - if not path: - return - self._load_replay_video(Path(path)) - def _on_reveal_config(self) -> None: sensor_type = self.sensor_picker.sensor_type() path = sensor_json_path(sensor_type) @@ -2346,7 +2008,7 @@ class StudioMainWindow(QMainWindow): def _on_run_toggled(self, checked: bool) -> None: if self.toolbar.run_role() == "playback": # The primary button of the replay view is its transport. - self._on_transport_toggled(bool(checked)) + self.replay_ctl.on_transport_toggled(bool(checked)) return if checked: self._start_run() @@ -2525,7 +2187,7 @@ class StudioMainWindow(QMainWindow): if self.toolbar.reset_role() == "stop": # In the replay view that button is the operator's way out of a replay or # a recompute pass, whatever state the transport is in. - self._stop_replay_activity() + self.replay_ctl.stop_activity() return if self._session_phase != "running" or self._active_worker is None: return diff --git a/apps/studio/src/orisys_studio/replay_controller.py b/apps/studio/src/orisys_studio/replay_controller.py new file mode 100644 index 0000000..fcf97c8 --- /dev/null +++ b/apps/studio/src/orisys_studio/replay_controller.py @@ -0,0 +1,641 @@ +"""The replay view's policy: what is open, what is playing, and what a pass is. + +``replay.py`` holds the *engine* — the recording reader, the cursor, the asynchronous +depth recomputation. This module holds the *policy* around it, which used to live in +``app_window.py``: which entry points open a video and which refuse it, what the +transport does when the operator presses play, when the button is a transport and when +it is an acquire button, and what "从视频重算" does to the panels and to the run +pipeline. + +The split is by change driver, not by convenience: the engine changes when the +*recording format* changes, this controller when the *operator's workflow* changes, +and the window when the furniture changes. + +What this controller drives directly: the toolbar's replay controls, the charts (the +timeline), the 2D panel, the 3D pane's source label, and the five replay actions. What +it borrows from the window is named by :class:`ReplayHost` — the status line, the rail, +the view switch, the display unit, the single 2D sink, and the run pipeline a recompute +pass is. Those are window-level operations, so the window keeps them and this module +states its dependence on them out loud instead of reaching into a god object. +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional, Protocol + +import numpy as np +from PyQt5.QtCore import QObject, QTimer +from PyQt5.QtWidgets import QFileDialog + +from orisys_app_shared.acquisition.image_backend import SOURCE_REPLAY, ImageFrame +from orisys_app_shared.i18n import t +from orisys_app_shared.paths import recordings_dir + +logger = logging.getLogger(__name__) + +#: Playback tick while a recording is playing. 33 ms is one frame at 30 fps; the +#: cursor is advanced from the wall clock, so a missed tick does not slow the +#: recording down. +PLAYBACK_TICK_MS = 33 + + +class ReplayHost(Protocol): + """What the replay controller needs from the window. + + The names are the window's own (its transitional shims included), because the + window is the only implementation today. They are listed here so the boundary is + explicit and so a test can stand in for the window with a fake. + """ + + #: Operator preferences (the chosen video path is remembered here). + prefs: dict + + @property + def _running(self) -> bool: ... + + @property + def _preview_open(self) -> bool: ... + + @property + def _session_active(self) -> bool: ... + + def _in_replay_view(self) -> bool: ... + + def _apply_view_preset(self, preset: str) -> None: ... + + def _set_display_unit(self, unit: str) -> None: ... + + def _set_source_mode(self, mode: str) -> None: ... + + def _show_image_frame(self, frame: ImageFrame) -> None: ... + + def _current_source_kind(self) -> str: ... + + def _drop_render_marker(self) -> None: ... + + def _notify(self, text: str, ms: int = ...) -> None: ... + + def _start_run(self) -> None: ... + + def _stop_run(self, *, reason: str = ..., wait: bool = ...) -> None: ... + + def _update_run_enabled(self) -> None: ... + + def _refresh_status_rail(self) -> None: ... + + +@dataclass +class ReplayActions: + """The five menu/shortcut entries whose meaning depends on the replay state.""" + + open_video: Any + playback: Any + latest: Any + play: Any + recompute: Any + + +class ReplayController(QObject): + """The replay view's state machine and the panels it drives.""" + + def __init__( + self, + *, + engine: Any, + host: ReplayHost, + toolbar: Any, + charts: Any, + image_panel: Any, + mesh_panel: Any, + parent: Optional[QObject] = None, + ) -> None: + super().__init__(parent) + self._engine = engine + self._host = host + self._toolbar = toolbar + self._charts = charts + self._image_panel = image_panel + self._mesh_panel = mesh_panel + self._actions: Optional[ReplayActions] = None + #: True while a recompute pass is running: it is a *run* over a recording, so + #: the acquire pipeline owns the panels and the transport stops the pass. + self._recompute_running = False + self._play_from = 0.0 + self._play_started_at = 0.0 + self.play_timer = QTimer(self) + self.play_timer.setInterval(PLAYBACK_TICK_MS) + self.play_timer.timeout.connect(self.on_play_tick) + + # ------------------------------------------------------------------ wiring + def wire(self) -> None: + """Connect the engine and the charts. Called once, after construction.""" + self._engine.position_changed.connect(self.on_replay_position) + self._engine.frames_changed.connect(self.render_frame) + self._engine.status_changed.connect(self.refresh_state) + self._engine.recording_changed.connect(self.on_recording_changed) + # Dragging the shared cursor on the charts scrubs the recording too. + self._charts.cursor_moved.connect(self.on_cursor_moved) + self._charts.timeline_taken_over.connect(self.pause_playback) + + def bind_actions( + self, + *, + open_video: Any, + playback: Any, + latest: Any, + play: Any, + recompute: Any, + ) -> None: + """Take the replay entries once the menus exist, and connect them. + + The controller owns what these entries *mean*, so it also owns their signals: + one place decides whether opening a video is allowed and what it does. + """ + self._actions = ReplayActions( + open_video=open_video, + playback=playback, + latest=latest, + play=play, + recompute=recompute, + ) + open_video.triggered.connect(self.on_browse_video) + playback.triggered.connect(self.on_open_recording) + latest.triggered.connect(self.open_latest_recording) + play.triggered.connect(self.on_play_shortcut) + recompute.toggled.connect(self.on_recompute_toggled) + + # ------------------------------------------------------------------- state + @property + def engine(self) -> Any: + return self._engine + + @property + def recompute_running(self) -> bool: + return self._recompute_running + + @recompute_running.setter + def recompute_running(self, running: bool) -> None: + self._recompute_running = bool(running) + + def is_playing(self) -> bool: + return self.play_timer.isActive() + + def recompute_active(self) -> bool: + """Whether the operator asked for a recompute pass (the switch is on).""" + if self._actions is None: + return False + return bool(self._actions.recompute.isChecked()) + + def owns_image(self) -> bool: + """True while the 2D panel shows the replay, not the live camera. + + Without this the live renderer overwrote the replay's frame on every packet + *and* on every preview tick, so a replay running next to an active session (or + a preview) looked as if it had opened the camera instead of the recording. A + recompute pass is the pipeline drawing again, so it owns the panel while it + runs. + """ + return ( + self._host._in_replay_view() + and self._engine.has_video() + and not self._recompute_running + ) + + # ------------------------------------------------------------------ entries + def on_cursor_moved(self, t_rel_s: float) -> None: + if self._recompute_running: + # The recompute pass owns the timeline: its samples are arriving in + # order and the cursor follows them. + return + if self._host._in_replay_view() and self._engine.has_video(): + self._engine.set_position(float(t_rel_s)) + + def on_open_recording(self) -> None: + """Choose a recorded video and switch to the view that replays it. + + The recording a video belongs to is its own directory, so the picker takes the + *video* (what the operator recognises) and the charts pair it with the + ``forces.csv`` written beside it. + """ + start = str(recordings_dir()) + chosen, _filter = QFileDialog.getOpenFileName( + self._host, + t("playback.open"), + start, + t("playback.video_filter"), + ) + if not chosen: + return + self.load_video(Path(chosen)) + + def on_browse_video(self) -> None: + """Open a recorded video for replay (file menu). + + Choosing a video is the replay view's source selection, so this takes the same + path as the rail's button: a recorded video is something to *play*, not a + second way to acquire. + """ + if self._host._running or self._host._preview_open: + return + start = self._toolbar.video_path() or str(Path.home()) + path, _ = QFileDialog.getOpenFileName( + self._host, + t("menu.open_video"), + start, + t("playback.video_filter"), + ) + if not path: + return + self.load_video(Path(path)) + + def on_recompute_button(self) -> None: + """The replay view's own entry for the processed picture.""" + if self._actions is None: + return + self._actions.recompute.setChecked(not self._actions.recompute.isChecked()) + + def on_play_shortcut(self) -> None: + """Space plays or pauses the recording, if there is one to move through.""" + if self._toolbar.run_role() != "playback" or not self._toolbar.btn_run.isEnabled(): + return + self._toolbar.btn_run.setChecked(not self._toolbar.btn_run.isChecked()) + + # ------------------------------------------------------------------ opening + def sync_actions(self) -> None: + """Replay entries exist in the replay view, and nowhere else. + + Opening a recording from the *live* view used to leave the window in a state + nothing described: the source became a video file, the 2D panel did not change + (the replay renderer only draws in the replay view) and the Run button acquired + from that file. The entries are therefore disabled outside the replay view, with + the reason in the tooltip, instead of quietly doing half of it. + """ + if self._actions is None: + return + allowed = self._host._in_replay_view() + reason = "" if allowed else t("toolbar.replay_view_only") + for action in ( + self._actions.open_video, + self._actions.playback, + self._actions.latest, + ): + action.setEnabled(allowed) + action.setToolTip(reason) + # The rail's entries are in the replay group already; the recompute button also + # has to follow whether there is anything to recompute *from*. + self._toolbar.set_recompute_enabled(allowed and self.can_recompute()) + self._toolbar.set_recompute_checked(self._actions.recompute.isChecked()) + + def can_recompute(self) -> bool: + """Whether 「从视频重算」 has something to read: a recording, or a video file.""" + if not self._host._in_replay_view(): + return False + if self._engine.has_video(): + return True + named = self._toolbar.video_path() + return bool(named) and Path(named).is_file() + + def load_video(self, video: Path) -> bool: + """Open what the chosen video *is*: a recording, or just a video. + + A **recording** is a directory that holds ``forces.csv`` beside the video — the + same rule ``list_recordings`` uses, so the two cannot disagree about what a + recording is. Anything else is still worth watching: an operator who points at + a clip expects the transport to play it, not to be told to press a different + button first. So a bare video is opened for playback (no forces, no session + metadata, and the rail says so), and only a file that cannot be decoded at all + is refused — with what is missing, and with 「从视频重算」 still offered, because + that is the only way to get *numbers* out of a video that has no recording. + """ + video = Path(video) + if not self._host._in_replay_view(): + # Defensive: the entries are disabled outside the replay view, so this only + # happens programmatically — and refusing is better than leaving the window + # with a video source and no replay renderer. + self._host._notify(t("toolbar.replay_view_only")) + return False + folder = video.parent if video.is_file() else video + if not (folder / "forces.csv").is_file(): + if video.is_file() and self._engine.open_video(video): + self._toolbar.set_video_path(str(video)) + self._host.prefs["video_path"] = str(video) + # The window's own switch, which also remembers the choice: the file + # menu's source selector and this path must leave the same state. + self._host._set_source_mode("video") + self.render_frame() + self._host._notify(t("playback.video_only_opened", name=video.name)) + self.refresh_state() + return True + if video.is_file(): + self._toolbar.set_video_path(str(video)) + self._host.prefs["video_path"] = str(video) + self._host._set_source_mode("video") + if not self._engine.has_video(): + # Nothing else is on screen, so the reason belongs in the panel that + # would have shown the frames — not only in a status line that expires. + self._image_panel.clear_preview(t("playback.video_only_panel")) + self.refresh_state() + self._host._notify(t("playback.not_a_recording", name=video.name)) + return False + if not self.load_recording(folder): + return False + self._toolbar.set_video_path(str(video)) + self._host._refresh_status_rail() + return True + + def load_recording(self, directory: Path) -> bool: + if not self._host._in_replay_view(): + self._host._apply_view_preset("replay") + opened = self._engine.open_recording(Path(directory)) + if not opened: + self._host._notify(t("playback.open_failed", error=Path(directory).name)) + return False + self.refresh_state() + return True + + def open_latest_recording(self) -> None: + """Switch to the replay view on the most recent recording.""" + from orisys_app_shared.record import list_recordings + + if not self._host._in_replay_view(): + self._host._notify(t("toolbar.replay_view_only")) + return + found = list_recordings(recordings_dir()) + if not found: + self._host._notify(t("playback.none_found")) + return + self.load_recording(Path(found[-1])) + + def on_recording_changed(self, opened: bool) -> None: + """A recording was opened (or dropped): re-point everything that reads it.""" + self.refresh_state() + if not opened: + self.pause_playback() + return + if self._host._in_replay_view(): + self.apply_source(True) + self.render_frame() + + def refresh_state(self) -> None: + """Push the engine's state into the charts' mode and the rail.""" + if self._engine.is_video_only() and not self._toolbar.video_path(): + path = self._engine.video_path() + if path is not None: + self._toolbar.set_video_path(str(path)) + elif self._engine.is_loaded() and not self._toolbar.video_path(): + # Opened without naming a video (the file menu's "latest"): name the + # segment that is on screen so the rail states the same thing the + # picker would. + reader = self._engine.reader() + segments = sorted(reader.segments()) if reader is not None else [] + if segments: + self._toolbar.set_video_path(str(segments[0])) + self._toolbar.set_run_checked(self.play_timer.isActive()) + # Loading a recording is what makes a recompute pass possible at all — but a + # chosen video file is enough to recompute *from*, even when it is not a + # recording: that is the way to look at a video that has no forces.csv. + if self._actions is not None: + self._actions.recompute.setEnabled(self.can_recompute()) + self.sync_actions() + self._host._update_run_enabled() + self._host._refresh_status_rail() + + # ---------------------------------------------------------------- transport + def on_replay_position(self, position_s: float) -> None: + self._charts.set_cursor(float(position_s)) + + def on_transport_toggled(self, playing: bool) -> None: + """The rail's primary button in the replay view: play or pause.""" + if self.recompute_active(): + # A recompute pass is a *run*: play starts the pipeline over the + # video, pause stops it. It can only go forward (the chain is + # stateful), so playing again after the end starts from the top. + if playing: + self._recompute_running = True + self._host._start_run() + elif self._recompute_running: + # The flag is this controller's, so it is cleared here as well as in + # the session stop below: a pass that ended must not keep claiming to + # be running just because the stop came from the transport. + self._recompute_running = False + self._host._stop_run(reason="user_stop") + self._toolbar.set_run_checked(bool(playing) and self._host._running) + return + if playing and self._engine.has_video(): + self.start_playback() + else: + self.pause_playback() + + def start_playback(self) -> None: + duration = self._engine.duration_s() + if duration <= 0: + # A zero-length timeline is not something to play, and silently un-checking + # the button is how "pressing 回放 does nothing" felt to the operator. + self._toolbar.set_run_checked(False) + self._host._notify(t("playback.empty_recording")) + return + position = self._engine.position() + if position >= duration - 0.02: + # At the end: playing again means playing from the start. + self.set_position(0.0) + position = 0.0 + self._play_from = float(position) + self._play_started_at = time.perf_counter() + self.play_timer.start() + self._toolbar.set_run_checked(True) + self._toolbar.refresh_labels() + # The stop button appears with the playback: one place decides what the rail + # offers, so it is told here rather than kept in step in two places. + self._host._update_run_enabled() + + def pause_playback(self) -> None: + if self.play_timer.isActive(): + self.play_timer.stop() + if self._toolbar.btn_run.isChecked(): + self._toolbar.set_run_checked(False) + self._toolbar.refresh_labels() + self._host._update_run_enabled() + + def on_play_tick(self) -> None: + """Advance the playhead from the wall clock, not from the tick count.""" + duration = self._engine.duration_s() + elapsed = time.perf_counter() - self._play_started_at + position = self._play_from + elapsed + if position >= duration: + self.set_position(duration) + self.pause_playback() + return + self.set_position(position) + + def set_position(self, position_s: float) -> None: + """Move the playhead without the position signal echoing back into it.""" + self._engine.set_position(float(position_s)) + + def stop_activity(self) -> None: + """Stop whatever the replay view is doing. The operator's escape hatch. + + Two different things can be running there — plain playback of the recorded + timeline, and a recompute pass over the video — and the person in front of the + screen should not have to work out which one is stuck before they can stop it. + So this stops both, clears a recomputed 3D picture, and says what it stopped. + Safe to press when nothing is running (it says so rather than doing nothing + silently, which is what made a stuck state feel stuck). + """ + stopped: list[str] = [] + if self._recompute_running or self._host._session_active: + self._recompute_running = False + self._host._stop_run(reason="user_stop") + stopped.append(t("toolbar.stopped_pass")) + if self.play_timer.isActive(): + self.pause_playback() + stopped.append(t("toolbar.stopped_playback")) + self._mesh_panel.show_reference_only() + self._host._update_run_enabled() + if stopped: + self._host._notify(t("toolbar.stopped", what=", ".join(stopped))) + else: + self._host._notify(t("toolbar.nothing_to_stop")) + + # ---------------------------------------------------------------- recompute + def video_segment(self) -> Optional[Path]: + """The recorded video a recompute pass should read.""" + reader = self._engine.reader() + segments = sorted(reader.segments()) if reader is not None else [] + if segments: + return Path(segments[0]) + named = self._toolbar.video_path() + if named and Path(named).is_file(): + return Path(named) + return None + + def on_recompute_toggled(self, checked: bool) -> None: + """Off: read the recorded CSV. On: re-run the pipeline over the video. + + The CSV is the session's own measurement. Re-running the vision chain over the + recorded (lossy) video answers a different question — "what would this algorithm + version say about that footage" — and it can only run *forward*, because the + chain is stateful, so it is a pass rather than a scrub. + """ + wanted = bool(checked) and self.can_recompute() + if self._actions is not None and self._actions.recompute.isChecked() != wanted: + self._actions.recompute.blockSignals(True) + self._actions.recompute.setChecked(wanted) + self._actions.recompute.blockSignals(False) + self.pause_playback() + if wanted: + segment = self.video_segment() + if segment is None: + self._host._notify(t("playback.no_video")) + if self._actions is not None: + self._actions.recompute.blockSignals(True) + self._actions.recompute.setChecked(False) + self._actions.recompute.blockSignals(False) + return + # The charts stop reading the CSV; the pipeline's own samples become the + # data, and the 2D panel shows what it produces. The pass starts here so + # the switch is not a blank chart waiting for a second click. + self._charts.clear_recording_source() + self._image_panel.set_source_kind("recompute") + self._toolbar.set_source_mode("video") + self._toolbar.set_video_path(str(segment)) + self._host._refresh_status_rail() + self._recompute_running = True + # The 3D pane now says "recomputed", not "replayed": the pass is what feeds + # it, through the ordinary acquisition sink. + self._mesh_panel.set_source_kind(self._host._current_source_kind()) + # Playing is what starts the pass: one path in, so the button and the menu + # cannot start it twice. + self._toolbar.set_run_checked(True) + self._toolbar.refresh_labels() + self.on_transport_toggled(True) + return + if self._recompute_running: + # The flag is cleared here, not only by the session stop: it belongs to this + # controller, and a pass that ended must stop claiming to be running. + self._recompute_running = False + self._host._stop_run(reason="user_stop") + # The pass is over, so the last recomputed frame must go: leaving it on + # screen would show a measurement nothing is producing any more. + self._mesh_panel.show_reference_only() + if self._host._in_replay_view(): + self._mesh_panel.set_source_kind(self._host._current_source_kind()) + if self._engine.is_loaded(): + self.apply_source(True) + # A bare video has no charts to point at the recording, but it does have + # frames to show again now that the pass is over. + self.render_frame() + + # ------------------------------------------------------------------ drawing + def render_frame(self) -> None: + """Draw what the 2D panel should show for the current replay cursor. + + A recording stores the raw camera image and nothing else, so the frame this + builds has no flow and no depth — the plan then says so, in one line, and the + panel offers exactly the one view the frame can be. The processed picture of the + same footage is what 「从视频重算」 produces, through the *same* backend. + """ + if not self._host._in_replay_view() or not self._engine.has_video(): + return + if self._recompute_running: + # The pipeline is drawing its own frames (see ``owns_image``). + return + image = self._engine.video_frame_bgr() + if image is None: + self._image_panel.clear_preview(self._engine.video_message()) + return + self._host._show_image_frame( + ImageFrame( + image, + source=SOURCE_REPLAY, + t_rel_s=float(self._engine.position()), + ) + ) + + def apply_source(self, active: bool) -> None: + """Point the charts at the recording (or back at the live session).""" + if active and self._engine.is_loaded(): + reader = self._engine.reader() + self._charts.set_recording_source(reader) + # A recording is shown in a unit it actually carries: newton only when + # it has a calibrated axis somewhere, otherwise the charts would be + # empty and every card a dash. The operator's preference is kept when + # the recording can honour it. + wanted = "N" if self._charts.unit() == "N" else "raw" + unit = self.unit_for(reader) + self._engine.set_unit(unit) + self._host._set_display_unit(unit) + if unit != wanted: + self._host._notify(t("status.units_raw_fallback")) + self._charts.show_all() + self._image_panel.set_source_kind("replay") + return + if self._charts.is_recording_source(): + self._charts.clear_recording_source() + self._charts.follow_latest() + self._image_panel.set_source_kind("live") + # The panel held a recorded frame: the next live packet must repaint it, so the + # "already rendered" marker is dropped. + self._host._drop_render_marker() + if not self._host._running and not self._host._preview_open: + self._image_panel.clear_preview(t("panels.arrow_empty")) + + def unit_for(self, reader: Any) -> str: + """The unit a recording can be shown in: raw unless it has newtons. + + ``raw`` is always present (it is what the device measured); the calibrated + triplet only exists for the axes the recording actually calibrated. Asking for + newton on a raw-only recording draws three empty charts, so the same fallback + the live view uses applies here. + """ + if self._charts.unit() != "N": + return "raw" + info = reader.info() + newtons = getattr(info, "newtons", None) + if newtons is None: + return "raw" + values = np.asarray(newtons, dtype=float) + return "N" if values.size and bool(np.isfinite(values).any()) else "raw" diff --git a/apps/studio/tests/test_replay_controller.py b/apps/studio/tests/test_replay_controller.py new file mode 100644 index 0000000..0cb649b --- /dev/null +++ b/apps/studio/tests/test_replay_controller.py @@ -0,0 +1,465 @@ +"""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 diff --git a/apps/studio/tests/test_ui_smoke.py b/apps/studio/tests/test_ui_smoke.py index 553b37f..36a06f6 100644 --- a/apps/studio/tests/test_ui_smoke.py +++ b/apps/studio/tests/test_ui_smoke.py @@ -2046,10 +2046,12 @@ def test_the_replay_view_offers_a_stop_that_always_works(make_window, replay_rec ) # Nothing running: it says so instead of doing nothing at all. - # (The stubbed ``_stop_run`` above records the reason but does not clear the flag - # the real one clears, so the state is completed by hand here.) - window._recompute_running = False - window.toolbar.btn_reset.click() + # (The pass is over, so the stop button is correctly disabled — the escape hatch is + # therefore exercised the way Esc reaches it, through the controller.) + assert not window.toolbar.btn_reset.isEnabled(), ( + "a stop is still offered after the pass was stopped" + ) + window.replay_ctl.stop_activity() QApplication.processEvents() assert t("toolbar.nothing_to_stop") in window.statusBar().currentMessage(), ( window.statusBar().currentMessage()