拆分 B：回放策略移入 ReplayController（app_window 2837→2499 行）
- 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 用例按机器实况失败，与本次改动无关。
This commit is contained in:
ykcheungab
@@ -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
|
||||
|
||||
@@ -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"
|
||||
@@ -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
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user