feat(studio): 回放可以直接播放「一条视频」——不必先按「从视频重算」

- What: 新增 `record/playback.py::VideoOnlySource`(一个视频文件的时间轴:时长来自
  `CAP_PROP_FRAME_COUNT/FPS`,按位置解码、顺序前进时只 step 不 seek、越界钳到最后一帧;没有帧计数或帧率就
  拒绝而不是猜);`ReplayController` 增加 `open_video()`/`has_video()`/`is_video_only()`/`video_path()`,
  `duration_s`/`set_position`/`_refresh_video`/`close` 都覆盖这条路径,并在 `warnings_text()` 里说明
  "这段视频没有配套记录";`app_window._load_replay_video()` 改为按"这条视频是什么"处理:
  同目录有 `forces.csv` → 当记录打开(原行为),否则**当视频播放**(曲线为空、明确提示),只有连解码都不行
  才拒绝并给出 `从视频重算` 的出路;transport/`_render_replay_frame`/`_replay_owns_image`/状态栏 tooltip
  一律改用 `has_video()`(而查曲线的 `_apply_replay_source` 仍用 `is_loaded()`)。
- Why: 用户要求"只导入视频也能用回放按钮"。此前有录像必须走「从视频重算」才能动,因为一条裸视频被当成
  "不是记录"而拒绝(更早是当成空记录"成功"打开→零长度时间轴→按回放没反应)。这两种都不对:视频就是可以看的
  东西,只是没有力值;把它按视频播放、把"没有记录"如实写在标签与提示里,才既不骗人也不挡人。
- Verification: `shared/tests/test_playback.py` +3(VideoOnlySource:时长/按位置取帧/顺序前进/越界钳位/关闭后
  返回空/不可解码与无时间轴被拒;顺带修回我在插入新类时弄丢的 `list_recordings` 的 `base = Path(root)`);
  `test_ui_smoke.py` 改 1 增 1(裸视频可播放:`is_video_only`、transport 可用、tooltip 说明曲线为空、
  按下后位置推进且解出画面、停止有效;不可解码文件仍被拒并提示);ui_check 587 项 0 失败(回放段改为验证
  "裸视频能播");apps/studio 与 shared 全量通过;locale_check 0 缺失。
- Not in this change: 裸视频的曲线(没有 CSV 就没有力值,这是事实而不是待办);裸视频的三维形变(要形变得走
  「从视频重算」,面板标签在回放里仍然是"录像不含三维场");多段视频拼接播放。
This commit is contained in:
ykcheungab
2026-09-14 20:21:13 +08:00
parent c9a28e7c3a
commit 6de6b8b703
10 changed files with 404 additions and 82 deletions
+4 -1
View File
@@ -103,6 +103,7 @@
"playback_pause_tip": "Pause playback (dragging the cursor, box zoom or the wheel also pauses)",
"playback_disabled": "Open a recorded video first (“Choose a recorded video…” at the top left)",
"playback_needs_recording": "This video has no recording beside it (no forces.csv): replay needs a recording; to just watch the video use View ▸ Recompute forces from the video",
"playback_video_only": "Play this video (no recording beside it: the picture plays, the force charts are empty)",
"reset": "Reset",
"stop_playback": "Stop",
"stop_playback_tip": "Stop the replay: plain playback and a “recompute from the video” pass both stop here (Esc does the same).",
@@ -213,7 +214,9 @@
"no_video": "This recording has no usable video segment",
"open_failed": "Could not open that recording: {error}",
"not_a_recording": "“{name}” is not a recording: its folder has no forces.csv (a recording is the video plus forces.csv and session.json). Replay needs a recording; to just watch this video, use “Recompute forces from the video”.",
"video_only_panel": "This is not a recording (no forces.csv beside it)\nTo watch this video, use\nView ▸ Recompute forces from the video",
"video_only_opened": "“{name}” has no recording beside it (no forces.csv), so it is playing as a video: the picture and the timeline work, the force charts are empty. For forces and deformation use View ▸ Recompute forces from the video.",
"video_only_notice": "This video has no recording beside it (no forces.csv): it plays, but the force charts and the recording details are empty.",
"video_only_panel": "This is not a recording, and this video cannot be decoded (no forces.csv beside it, and it does not read as playable video)\nYou can still try View ▸ Recompute forces from the video",
"empty_recording": "This recording has no playable picture (no raw_*.mp4 segment), so it cannot be replayed",
"none_found": "No recording found under outputs/recordings",
"frame_dropped": "This frame never reached the video (the record queue overflowed)",
+4 -1
View File
@@ -103,6 +103,7 @@
"playback_pause_tip": "暂停播放(拖动游标、框选或滚轮也会暂停)",
"playback_disabled": "先打开一段录制视频(左上「选择录制视频…」)",
"playback_needs_recording": "这段视频旁边没有记录(缺 forces.csv):回放需要一条记录;只想看视频请用 视图 ▸ 从视频重算力值",
"playback_video_only": "播放这段视频(没有配套记录:可以看画面,力曲线为空)",
"reset": "重置",
"stop_playback": "停止",
"stop_playback_tip": "停止回放:无论是普通播放还是「从视频重算」通道,都在这里停下(Esc 同效)。",
@@ -213,7 +214,9 @@
"no_video": "这段录制没有可用的视频分段",
"open_failed": "无法打开该录制: {error}",
"not_a_recording": "「{name}」不是一条记录:同目录没有 forces.csv(一条记录 = 录像 + forces.csv + session.json)。回放需要记录;只想看这段视频请用「从视频重算力值」。",
"video_only_panel": "这不是一条记录(同目录缺少 forces.csv)\n只想看这段视频请用\n视图 ▸ 从视频重算力值",
"video_only_opened": "「{name}」没有配套记录(无 forces.csv),已按视频播放:可以看画面与拖时间轴,力曲线为空。要得助力值/形变请用 视图 ▸ 从视频重算力值。",
"video_only_notice": "这段视频没有配套记录(同目录无 forces.csv):可以播放,但力曲线与记录信息为空。",
"video_only_panel": "这不是一条记录,而且无法解码这段视频(没有 forces.csv,也不像可播放的视频文件)\n可以试试 视图 ▸ 从视频重算力值",
"empty_recording": "这条记录没有可播放的画面(没有 raw_*.mp4 视频分段),因此无法回放",
"none_found": "outputs/recordings 下没有找到录制",
"frame_dropped": "该帧未被写入视频(记录队列曾溢出)",
+42 -23
View File
@@ -813,7 +813,7 @@ class StudioMainWindow(QMainWindow):
# 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.is_loaded():
if self._view_preset == "replay" and self.replay.has_video():
self.replay.set_position(float(t_rel_s))
def _on_open_recording(self) -> None:
@@ -835,25 +835,33 @@ class StudioMainWindow(QMainWindow):
self._load_replay_video(Path(chosen))
def _load_replay_video(self, video: Path) -> bool:
"""Open the recording a recorded video belongs to.
"""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. A video on its own is not one: it has no force rows and no
timeline. Opening its folder anyway "succeeded" (the reader tolerates a
directory with nothing in it) and then left the transport button playing a
zero-length timeline, which is why choosing a video and pressing 回放 did
nothing at all. So a bare video is refused with what is missing, and the rail
keeps its path so 「从视频重算」 can still run the chain over it.
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)
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.is_loaded():
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"))
@@ -870,7 +878,7 @@ class StudioMainWindow(QMainWindow):
"""Whether 「从视频重算」 has something to read: a recording, or a video file."""
if self._view_preset != "replay":
return False
if self.replay.is_loaded():
if self.replay.has_video():
return True
named = self.toolbar.video_path()
return bool(named) and Path(named).is_file()
@@ -907,7 +915,11 @@ class StudioMainWindow(QMainWindow):
def _refresh_replay_state(self) -> None:
"""Push the engine's state into the charts' mode and the rail."""
if self.replay.is_loaded() and not self.toolbar.video_path():
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.
@@ -944,7 +956,7 @@ class StudioMainWindow(QMainWindow):
self._stop_run(reason="user_stop")
self.toolbar.set_run_checked(bool(playing) and self._running)
return
if playing and self.replay.is_loaded():
if playing and self.replay.has_video():
self._start_playback()
else:
self._pause_playback()
@@ -1056,13 +1068,15 @@ class StudioMainWindow(QMainWindow):
self.mesh3d_panel.show_reference_only()
if self._view_preset == "replay":
self.mesh3d_panel.set_source_kind(self._current_source_kind())
if self._view_preset == "replay" and self.replay.is_loaded():
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()
def _render_replay_frame(self) -> None:
"""Draw what the 2D panel should show for the current replay cursor."""
if self._view_preset != "replay" or not self.replay.is_loaded():
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``).
@@ -1454,12 +1468,16 @@ class StudioMainWindow(QMainWindow):
def _update_run_enabled(self) -> None:
if self.toolbar.run_role() == "playback":
# In the replay view the button is a transport: it needs a recording,
# not a runnable sensor. While it plays it stays enabled so it can be
# paused.
loaded = self.replay.is_loaded()
# In the replay view the button is a transport: it needs something to
# play — a recording, or a bare video — not a runnable sensor. While it
# plays it stays enabled so it can be paused.
loaded = self.replay.has_video()
if loaded:
reason = ""
reason = (
t("toolbar.playback_video_only")
if self.replay.is_video_only()
else ""
)
elif self._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
@@ -1615,7 +1633,8 @@ class StudioMainWindow(QMainWindow):
# they live in the summary's tooltip (the bar that used to show them is
# gone: the two views share one layout).
tip_parts = [self.toolbar.video_path()]
if self._view_preset == "replay" and self.replay.is_loaded():
if self._view_preset == "replay" and self.replay.has_video():
# Includes the "this is a video, not a recording" note for a bare video.
tip_parts.append(self.replay.warnings_text())
self.toolbar.set_source_summary_tip(
"\n".join(part for part in tip_parts if part)
@@ -2810,7 +2829,7 @@ class StudioMainWindow(QMainWindow):
"""
return (
self._view_preset == "replay"
and self.replay.is_loaded()
and self.replay.has_video()
and not self._recompute_running
)
+69 -6
View File
@@ -14,13 +14,14 @@ is the acquisition path's job (see the studio's "recompute" mode).
from __future__ import annotations
import logging
from pathlib import Path
import numpy as np
from PyQt5.QtCore import QObject, pyqtSignal
from PyQt5.QtGui import QImage
from orisys_app_shared.i18n import t
from orisys_app_shared.record import RecordingReader
from orisys_app_shared.record import RecordingReader, VideoOnlySource
logger = logging.getLogger(__name__)
@@ -57,6 +58,8 @@ class ReplayController(QObject):
def __init__(self, parent=None) -> None:
super().__init__(parent)
self._reader: RecordingReader | None = None
#: A video with no recording beside it: playable, but it has no measurements.
self._video: VideoOnlySource | None = None
self._unit = "raw"
self._position = 0.0
#: The frame cache: one video image per cursor position.
@@ -67,8 +70,44 @@ class ReplayController(QObject):
# ------------------------------------------------------------- recording
def is_loaded(self) -> bool:
"""A *recording* is open: a video with its forces and session metadata."""
return self._reader is not None
def has_video(self) -> bool:
"""Whether there is anything to play at all — a recording, or a bare video.
The transport asks this, not ``is_loaded``: an operator who opens a clip from
a camera should be able to watch it. What the *charts* ask is ``is_loaded``,
because a bare video has no rows to draw.
"""
return self._reader is not None or self._video is not None
def is_video_only(self) -> bool:
return self._reader is None and self._video is not None
def video_path(self):
return self._video.path if self._video is not None else None
def open_video(self, path) -> bool:
"""Open a bare video for playback (no forces, no session metadata)."""
self._release_reader()
self._release_video()
try:
self._video = VideoOnlySource(Path(path))
except Exception: # noqa: BLE001 - reported through the status text
logger.exception("Could not open the video %s", path)
self._video = None
self._reset_frames()
self.recording_changed.emit(False)
self.status_changed.emit()
return False
self._reset_frames()
self._position = 0.0
self.recording_changed.emit(True)
self._refresh_warnings()
self.set_position(0.0)
return True
def directory(self):
return self._reader.directory if self._reader is not None else None
@@ -107,8 +146,9 @@ class ReplayController(QObject):
return True
def close(self) -> None:
"""Drop the recording (the decoder is released with it)."""
"""Drop whatever is open (the decoder is released with it)."""
self._release_reader()
self._release_video()
self._reset_frames()
self.recording_changed.emit(False)
self.status_changed.emit()
@@ -121,6 +161,14 @@ class ReplayController(QObject):
except Exception: # noqa: BLE001 - releasing is best effort
logger.debug("Could not close the recording reader", exc_info=True)
def _release_video(self) -> None:
video, self._video = self._video, None
if video is not None:
try:
video.close()
except Exception: # noqa: BLE001 - releasing is best effort
logger.debug("Could not close the video", exc_info=True)
def _reset_frames(self) -> None:
self._video_position = None
self._video_qimage = None
@@ -131,13 +179,15 @@ class ReplayController(QObject):
return self._position
def duration_s(self) -> float:
if self._reader is None:
return 0.0
if self._reader is not None:
return float(self._reader.info().duration_s)
if self._video is not None:
return float(self._video.duration_s)
return 0.0
def set_position(self, t_rel_s: float) -> None:
"""Move the single cursor: the video frame and the charts follow it."""
if self._reader is None:
if not self.has_video():
return
duration = self.duration_s()
self._position = max(0.0, min(float(t_rel_s), duration if duration > 0 else 0.0))
@@ -168,7 +218,7 @@ class ReplayController(QObject):
return self._video_reason
def _refresh_video(self) -> None:
if self._reader is None:
if self._reader is None and self._video is None:
self._video_position = None
self._video_qimage = None
self._video_reason = t("playback.none_found")
@@ -176,6 +226,15 @@ class ReplayController(QObject):
if self._video_position == self._position:
return
self._video_position = self._position
if self._video is not None:
image, _frame = self._video.video_frame_at(self._position)
if image is None:
self._video_qimage = None
self._video_reason = t("playback.frame_unreadable")
return
self._video_reason = ""
self._video_qimage = _bgr_to_qimage(image)
return
image, frame = self._reader.video_frame_at(self._position)
if image is None:
self._video_qimage = None
@@ -195,6 +254,10 @@ class ReplayController(QObject):
def _refresh_warnings(self) -> None:
self._warnings = ""
if self._reader is None:
if self._video is not None:
# Not a warning: a statement of what this file *is*, so nobody reads an
# empty chart as a measurement of zero.
self._warnings = t("playback.video_only_notice")
return
info = self._reader.info()
codes = self._reader.replay_warnings(sdk_version=_sdk_version())
@@ -691,6 +691,11 @@ class ToolbarWidget(QWidget):
key = "toolbar.playback_pause_tip" if self.btn_run.isChecked() else "toolbar.playback_tip"
if not self._run_available and self._run_tooltip_reason:
self.btn_run.setToolTip(self._run_tooltip_reason)
elif self._run_tooltip_reason:
# Usable, but with something the operator should know before pressing:
# a bare video plays with empty charts, and the tip says so instead of
# hiding that behind a second hover.
self.btn_run.setToolTip(f"{t(key)} — {self._run_tooltip_reason}")
else:
self.btn_run.setToolTip(t(key))
return
+67 -18
View File
@@ -2086,15 +2086,16 @@ def test_the_live_view_keeps_the_tracker_reset(make_window):
window.close()
def test_a_video_that_is_not_a_recording_is_refused_with_a_reason(
make_window, record_root
):
"""A bare video has no forces.csv, so it is not something this view can replay.
def test_a_bare_video_plays_even_though_it_is_not_a_recording(make_window, record_root):
"""A clip with no forces.csv beside it is still something to watch.
The reader tolerates a directory with nothing in it, so opening the video's own
folder as a recording used to *succeed* — and then the transport played a
zero-length timeline, which is what "pressing 回放 does nothing" was.
An operator who points at a video expects the transport to play it. Opening the
video's own folder as a *recording* used to "succeed" against the tolerant reader
and then play a zero-length timeline ("pressing 回放 does nothing"), so the fix is
not to refuse the file but to play it as what it is: a video, with empty charts.
"""
import time as _time
import cv2
import numpy as np
from PyQt5.QtWidgets import QApplication
@@ -2106,33 +2107,81 @@ def test_a_video_that_is_not_a_recording_is_refused_with_a_reason(
str(bare), cv2.VideoWriter_fourcc(*"mp4v"), 30.0, (64, 48)
)
for index in range(30):
writer.write(np.full((48, 64, 3), index, dtype=np.uint8))
writer.write(np.full((48, 64, 3), index * 8, dtype=np.uint8))
writer.release()
assert bare.is_file()
window = _show(make_window(), (1440, 880))
try:
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert not window.replay.is_loaded()
assert not window.replay.has_video()
assert window._load_replay_video(bare) is True, "a playable video was refused"
QApplication.processEvents()
assert window.replay.has_video(), "nothing is playable"
assert window.replay.is_video_only(), "a bare video was taken for a recording"
assert not window.replay.is_loaded(), (
"a bare video must not claim to carry a recording's measurements"
)
assert window.replay.duration_s() > 0.5, window.replay.duration_s()
assert window.toolbar.btn_run.isEnabled(), "the transport is not offered"
assert t("toolbar.playback_video_only") in window.toolbar.btn_run.toolTip(), (
window.toolbar.btn_run.toolTip()
)
assert t("playback.video_only_notice") in window.replay.warnings_text()
# The charts have nothing to read, and say so rather than drawing zeros.
assert not window.force_plots.is_recording_source()
assert window._load_replay_video(bare) is False, (
"a video with no recording beside it was accepted as a recording"
window.toolbar.btn_run.setChecked(True)
QApplication.processEvents()
deadline = _time.time() + 3.0
while _time.time() < deadline and window.replay.position() < 0.4:
QApplication.processEvents()
_time.sleep(0.05)
assert window.replay.position() >= 0.4, (
f"the transport did not play the video: {window.replay.position()}"
)
assert window._play_timer.isActive()
image = window.replay.video_image()
assert image is not None and not image.isNull(), "no frame was decoded"
window._stop_replay_activity()
QApplication.processEvents()
assert not window._play_timer.isActive()
finally:
window.close()
def test_a_file_that_is_not_a_video_at_all_is_refused_with_a_reason(
make_window, record_root
):
"""Not a recording *and* not decodable: say what is missing and offer recompute."""
from PyQt5.QtWidgets import QApplication
from orisys_app_shared.i18n import t
broken = Path(record_root) / "not_really_a_video.mp4"
broken.write_text("this is not a video at all\n", encoding="utf-8")
window = _show(make_window(), (1440, 880))
try:
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_replay_video(broken) is False, (
"a file that cannot be decoded was accepted for playback"
)
QApplication.processEvents()
assert not window.replay.is_loaded(), "an empty folder was opened as a recording"
assert not window.replay.has_video()
assert not window.toolbar.btn_run.isEnabled(), (
"the transport is offered although there is nothing to play"
)
reason = window.toolbar.btn_run.toolTip()
assert "forces.csv" in reason or "recording" in reason.lower(), reason
# And what was missing, in the panel that would have shown the frames.
body = window.arrow_panel.image_label.text()
assert "forces.csv" in body or "recording" in body.lower(), body
# The way forward for a video that is not a recording: recompute from it.
reason = window.statusBar().currentMessage()
assert "forces.csv" in reason or "记录" in reason, reason
# The way forward for a video with no recording: recompute from it.
assert window.act_recompute.isEnabled(), (
"a chosen video cannot be recomputed either, so the operator has no way to "
"look at it"
)
finally:
window.close()
+40 -19
View File
@@ -3001,9 +3001,9 @@ def main() -> int:
f"{replay_window.toolbar.source_summary.text()!r}",
)
# A video that is not a recording must be refused *with a reason*, because the
# reader tolerates an empty folder: accepting it left the transport playing a
# zero-length timeline, which looked like "pressing 回放 does nothing".
# A video with no recording beside it is still something to *watch*: the transport
# plays it, the charts stay empty, and the rail/tooltip say why. Refusing it (the
# older behaviour) left an operator pointing at a clip with a dead button.
import cv2 as _cv2
import numpy as _np
@@ -3011,8 +3011,8 @@ def main() -> int:
bare_writer = _cv2.VideoWriter(
str(bare_video), _cv2.VideoWriter_fourcc(*"mp4v"), 30.0, (64, 48)
)
for _index in range(10):
bare_writer.write(_np.zeros((48, 64, 3), dtype=_np.uint8))
for _index in range(30):
bare_writer.write(_np.full((48, 64, 3), _index * 8, dtype=_np.uint8))
bare_writer.release()
other_window = StudioMainWindow(
prefs={"language": "zh_CN", "theme": "dark", "sensor_type": "box"}
@@ -3023,34 +3023,55 @@ def main() -> int:
other_window.brand_bar.btn_playback.click()
app.processEvents()
check(
other_window._load_replay_video(bare_video) is False,
"a video with no forces.csv beside it was accepted as a recording",
other_window._load_replay_video(bare_video) is True,
"a playable video with no recording beside it was refused",
)
app.processEvents()
check(
other_window.replay.has_video() and other_window.replay.is_video_only(),
"the video is not playable as a video",
)
check(
not other_window.replay.is_loaded(),
"an empty folder was opened as a recording",
"a video with no forces.csv claimed to be a recording",
)
check(
not other_window.toolbar.btn_run.isEnabled(),
"the transport is offered although there is nothing to play",
other_window.toolbar.btn_run.isEnabled(),
"the transport is not offered for a video that can be played",
)
check(
"forces.csv" in other_window.toolbar.btn_run.toolTip()
or "记录" in other_window.toolbar.btn_run.toolTip(),
f"the transport does not say why it cannot play: "
t("toolbar.playback_video_only") in other_window.toolbar.btn_run.toolTip(),
f"the transport does not say the charts are empty: "
f"{other_window.toolbar.btn_run.toolTip()!r}",
)
check(
t("playback.video_only_notice") in other_window.replay.warnings_text(),
f"the video is not described as one: {other_window.replay.warnings_text()!r}",
)
check(
not other_window.force_plots.is_recording_source(),
"the charts claim to read a recording that does not exist",
)
other_window.toolbar.btn_run.setChecked(True)
app.processEvents()
_deadline = time.time() + 3.0
while time.time() < _deadline and other_window.replay.position() < 0.4:
app.processEvents()
time.sleep(0.05)
check(
other_window.replay.position() >= 0.4,
f"the transport did not play the video: {other_window.replay.position()}",
)
check(
other_window.replay.video_image() is not None,
"no frame was decoded from the video",
)
check(
other_window.act_recompute.isEnabled(),
"a chosen video cannot be recomputed either: the operator has no way to view it",
)
check(
"forces.csv" in other_window.arrow_panel.image_label.text()
or "记录" in other_window.arrow_panel.image_label.text(),
f"the panel does not say what is missing: "
f"{other_window.arrow_panel.image_label.text()!r}",
)
other_window._stop_replay_activity()
app.processEvents()
other_window.close()
app.processEvents()
replay_window.activateWindow()
@@ -19,6 +19,7 @@ from orisys_app_shared.record.playback import (
RecordingFrame,
RecordingInfo,
RecordingReader,
VideoOnlySource,
list_recordings,
)
from orisys_app_shared.record.visualization3d import (
@@ -43,6 +44,7 @@ __all__ = [
"RecordingFrame",
"RecordingInfo",
"RecordingReader",
"VideoOnlySource",
"Visualization3DRecord",
"build_visualization3d_block",
"format_value",
@@ -602,6 +602,84 @@ def decimate_window(window: ForceWindow, max_points: int, *, unit: str = "raw")
)
class VideoOnlySource:
"""One video file with nothing beside it: watch it, but it has no measurements.
A recording is a directory (video + forces + session metadata); a *video* is what
an operator often has — a clip from a camera, a file somebody sent, footage whose
``forces.csv`` was never written. This serves the same questions
:class:`RecordingReader` serves for a recording — how long is it, what is at *t* —
so the replay view can play it with the transport instead of demanding that the
file be something it is not.
What it deliberately does **not** do: invent force rows, a session id or a config
hash. ``curve_source()`` is ``None`` and the caller is expected to say that the
charts have nothing to read. Recomputing forces from this video is the acquisition
path's job, not this one's.
"""
def __init__(self, path: str | Path) -> None:
self.path = Path(path)
if not self.path.is_file():
raise FileNotFoundError(f"Not a video file: {self.path}")
import cv2
self._capture = cv2.VideoCapture(str(self.path))
if not self._capture.isOpened():
self._capture.release()
self._capture = None
raise ValueError(f"Could not open the video: {self.path}")
self._fps = float(self._capture.get(cv2.CAP_PROP_FPS) or 0.0)
frames = float(self._capture.get(cv2.CAP_PROP_FRAME_COUNT) or 0.0)
if not (self._fps > 0.0) or not (frames > 0.0):
# Without a frame count or a rate there is no timeline to scrub, and a
# guessed one would put the cursor somewhere the picture is not.
self._capture.release()
self._capture = None
raise ValueError(
f"The video reports no frame count or frame rate: {self.path.name}"
)
self.duration_s = frames / self._fps
self._index = -1
self._lock = threading.Lock()
def video_frame_at(self, t_rel_s: float) -> tuple[Optional[np.ndarray], None]:
"""The frame at ``t_rel_s``: ``(bgr, None)`` or ``(None, None)``.
The second element is always ``None``: there is no frame *record* to hand back,
which is the difference between this and a recording (and the reason the
status line for a dropped frame does not apply here).
"""
import cv2
if self._capture is None:
return None, None
target = int(max(0.0, float(t_rel_s)) * self._fps)
target = max(0, min(target, int(round(self.duration_s * self._fps)) - 1))
with self._lock:
try:
if self._index >= 0 and target == self._index + 1:
ok, image = self._capture.read()
else:
self._capture.set(cv2.CAP_PROP_POS_FRAMES, target)
ok, image = self._capture.read()
except Exception: # noqa: BLE001 - a broken decode is not a crash
logger.exception("Could not decode %s at frame %s", self.path, target)
self._index = -1
return None, None
self._index = target if ok else -1
return (image if ok else None), None
def close(self) -> None:
with self._lock:
if self._capture is not None:
try:
self._capture.release()
except Exception: # noqa: BLE001 - releasing is best effort
pass
self._capture = None
def list_recordings(root: str | Path) -> list[Path]:
base = Path(root)
if not base.is_dir():
+79
View File
@@ -22,6 +22,7 @@ from orisys_app_shared.record import (
ExperimentRecorder,
RecordConfig,
RecordingReader,
VideoOnlySource,
list_recordings,
)
@@ -235,6 +236,84 @@ def test_list_recordings_finds_directories_newest_first(recording):
assert recording in found
# ------------------------------------------------------------- a video on its own
def _write_clip(path: Path, frames: int = 60, fps: float = 30.0) -> Path:
"""A tiny real video, so the decode path is exercised rather than mocked."""
import cv2
writer = cv2.VideoWriter(
str(path), cv2.VideoWriter_fourcc(*"mp4v"), fps, (64, 48)
)
assert writer.isOpened(), "this build cannot write mp4v"
for index in range(frames):
writer.write(np.full((48, 64, 3), index * 4, dtype=np.uint8))
writer.release()
return path
def test_a_video_with_no_recording_is_playable(tmp_path):
"""A clip with nothing beside it answers the same questions a recording does."""
clip = _write_clip(tmp_path / "clip.mp4")
source = VideoOnlySource(clip)
try:
assert source.duration_s == pytest.approx(2.0, abs=0.2)
first, extra = source.video_frame_at(0.0)
later, _ = source.video_frame_at(1.5)
assert first is not None and later is not None
assert first.shape == (48, 64, 3)
assert float(later.mean()) > float(first.mean()), (
"the clip gets brighter over time; the same frame came back twice"
)
# Stepping forward is the playback case and must not need a seek.
stepped, _ = source.video_frame_at(1.5 + 1.0 / 30.0)
assert stepped is not None
# Past the end clamps to the last frame instead of failing.
assert source.video_frame_at(99.0)[0] is not None
# There is deliberately no frame *record*: a video has no measurements.
assert extra is None
finally:
source.close()
assert source.video_frame_at(0.0) == (None, None), "closed source still served frames"
def test_a_file_that_is_not_a_video_is_refused(tmp_path):
broken = tmp_path / "broken.mp4"
broken.write_text("not a video at all", encoding="utf-8")
with pytest.raises(ValueError):
VideoOnlySource(broken)
with pytest.raises(FileNotFoundError):
VideoOnlySource(tmp_path / "missing.mp4")
def test_a_video_without_a_timeline_is_refused(tmp_path, monkeypatch):
"""No frame count or rate means no timeline, and a guessed one would lie.
The input is a decoder that reports neither, rather than a hand-made file: this
build will not even write a video with a zero frame rate.
"""
import cv2
clip = _write_clip(tmp_path / "clip.mp4")
real_capture = cv2.VideoCapture
class NoTimeline:
def __init__(self, *args, **kwargs) -> None:
self._real = real_capture(*args, **kwargs)
def isOpened(self) -> bool: # noqa: N802 - OpenCV spelling
return True
def get(self, prop): # noqa: ANN001 - OpenCV spelling
return 0.0
def release(self) -> None:
self._real.release()
monkeypatch.setattr(cv2, "VideoCapture", NoTimeline)
with pytest.raises(ValueError):
VideoOnlySource(clip)
if __name__ == "__main__": # pragma: no cover - manual runner
import shutil as _shutil