orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

删除深度重算引擎；修好回放视频每帧重开文件导致的无法播放，运输控制改到状态轨主按钮 - What: 1) 彻底删除深度重算引擎：shared/.../record/depth.py（DepthRecomputer/DepthResult）与 depth_backends.py（BoxReplay/FingerReplay）、其专属测试 test_depth_replay.py，record/__init__ 不再导出，playback.py::replay_warnings 不再报 depth_recomputed；apps/studio/replay.py 由 515 行瘦身为只做读盘+游标+视频帧+提示，回放视图的 2D 面板只提供「原始图像」；session.json 的 depth.replay 由 recompute 改为 none；删除 31 个只剩深度使用的文案键（depth_*/warn_state_*/warn_reset_*），warn_config_changed 与 warn_sdk_changed 改为描述"由视频重算"；ui_check 深度段、studio 两个深度用例、test_playback 三例、drive_shared 的 test_depth_replay 与 reproduce.py 的深度段一并清理（shared 套件 150→134 项）。2) 无法播放的根因是 RecordingReader.video_frame_at 每帧都 cv2.VideoCapture 重开并 seek：小片段已 ~10 ms/帧，1080p H.264 每次 seek 需从关键帧解码（上百 ms），而播放每 33 ms 要一帧且全在 UI 线程上，视频因此看着不动；改为按分段缓存解码器、顺序帧用 read() 步进（实测 10 ms → 0.5 ms/帧）、跳转才 seek，新增 RecordingReader.close() 释放。运输控制按用户建议上移到状态轨主按钮：toolbar.set_run_role("playback") 使回放视图的主按钮变成「▶ 回放」/「⏸ 暂停」（实时视图仍是「▶ 运行」），点按以 1× 推进游标（墙钟计时，漏拍不变慢）、到末尾自动停，拖动游标/框选/滚轮/拖总览条经 ForcePlotsPanel.timeline_taken_over 自动暂停，空格键同效；曲线表头不再放第二个播放按钮；按钮样式在回放角色下不再套用"停止"的红色语义。 - Why: 用户要求先清掉深度重算引擎、并反馈"现在无法播放视频"，建议把上方播放按钮在历史回放形式下改为回放。 - Verification: ui_check.py 554 项 0 失败（新增 seek<250ms/步进<60ms 的解码预算、回放按钮文字随播放/暂停切换、手动拖游标即暂停、切回实时视图按钮恢复运行、按深度视图已移除）；test_ui_smoke.py 71 项 0 失败；drive_shared.py 134 项 0 失败（深度引擎用例随代码删除）；locale_studio.py 262 键齐全、zh==en；reproduce.py 退出 0；.repro/probe_play.py 实测按「回放」后 0.75 s 内游标推进 0.689 s、面板 12/12 帧都在更新；两张复核图（回放视图的「▶ 回放」、实时视图的「▶ 运行」）。 - Not in this change: 录制文件格式与 CSV 列序不变（raw_*.mp4 与 forces.csv 继续一起写）；「由视频重算」复核模式保留（它复用实时管线，是有状态、只能顺序前进的一遍）。

Browse Source
This commit is contained in:
ykcheungab
2026-09-14 09:31:21 +08:00
parent 9b3c5fe46c
commit da8165cf03
19 changed files with 354 additions and 2056 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shared/src/orisys_app_shared/record/__init__.py
-3
View File
@@ -4,7 +4,6 @@ from orisys_app_shared.record.calibration_snapshot import (
list_calibration_snapshots,
store_calibration_snapshot,
)
from orisys_app_shared.record.depth import DepthRecomputer, DepthResult
from orisys_app_shared.record.experiment import (
EVENTS_HEADER,
FORCES_HEADER,
@@ -28,8 +27,6 @@ __all__ = [
"FORCES_HEADER",
"FRAMES_HEADER",
"RECORD_FORMAT_VERSION",
"DepthRecomputer",
"DepthResult",
"ExperimentRecorder",
"RecordConfig",
"RecordReport",
shared/src/orisys_app_shared/record/depth.py
-548
View File
@@ -1,548 +0,0 @@
"""Recompute depth for a recording, offline.
Depth is not stored with a recording (a per-frame depth stream would be several
GB per hour). It is a function of the frame and the sensor configuration, so a
replay re-runs the SDK over ``raw_*.mp4`` using the ``config_ref`` recorded in
``session.json``.
The honest failure modes are all surfaced rather than papered over:
* the SDK is not importable, or the recording's sensor config is gone;
* the recording predates the recorded config hash, so the recomputation is not
guaranteed to reproduce what was shown live;
* the segment cannot be decoded.
The caller is expected to show the returned ``error`` and to mark the depth pane
as unavailable instead of substituting some other image.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
import numpy as np
from orisys_app_shared.record.depth_backends import open_replay
from orisys_app_shared.record.playback import RecordingReader
logger = logging.getLogger(__name__)
@dataclass
class DepthResult:
"""Outcome of one recomputation attempt."""
depth: Optional[np.ndarray] = None
error: str = ""
frame_id: int = -1
t_rel_s: float = 0.0
#: What a replay could not reproduce exactly (never silently empty).
warnings: tuple[str, ...] = ()
#: Which recording this answer belongs to (the replay window compares it).
record_id: str = ""
@property
def ok(self) -> bool:
return self.depth is not None and not self.error
#: The replay started at a **reset**: the algorithm's state was cleared there, so
#: the frames since that point determine it exactly.
STATE_EXACT = "exact"
#: The replay is preceded by the recorded pre-roll: the intended reconstruction.
STATE_PREROLL = "preroll"
#: Neither: the state entering the target frame is unknown and may differ from
#: what the live run had.
STATE_APPROXIMATE = "approximate"
@dataclass(frozen=True)
class ReplayStep:
"""One leg of a replay: read ``source`` up to ``to_frame``.
``reset_after`` is the frame at which the algorithm's state was cleared live
(an operator ``zeroing``). The replay reaches that frame, then calls the
processor's ``reset()`` — which is what the live run did — and keeps reading.
"""
source: Path
to_frame: int
reset_after: Optional[int] = None
#: How many SDK reads a faithful replay may cost before it is declared
#: unrecoverable (5 minutes at 30 fps). A replay that re-reads an hour of video
#: to answer one cursor move is not a feature; saying so is.
MAX_REPLAY_FRAMES = 9000
@dataclass(frozen=True)
class ReplayPlan:
"""Where the SDK must start reading to reproduce one frame's state."""
segment: Path
#: The legs of the replay, in order. The last one ends at ``stop_at``.
steps: tuple[ReplayStep, ...] = ()
#: Frame within the final segment the caller asked for (1-based).
stop_at: int = 0
#: Frames to feed from another file before the first step.
preroll: Path | None = None
preroll_frames: int = 0
#: The ``zeroing`` frame the plan is anchored to, when there is one.
reset_frame_id: Optional[int] = None
#: ``reset`` / ``preroll`` / ``segment_start``.
anchor: str = "segment_start"
#: :data:`STATE_EXACT`, :data:`STATE_PREROLL` or :data:`STATE_APPROXIMATE`.
state: str = STATE_APPROXIMATE
warnings: tuple[str, ...] = ()
error: str = ""
@property
def frames_to_read(self) -> int:
"""How many SDK reads a faithful recomputation costs."""
return int(self.preroll_frames) + sum(
max(0, int(step.to_frame)) for step in self.steps
)
class DepthRecomputer:
"""Recompute depth maps for one recording, on demand."""
def __init__(self, reader: RecordingReader) -> None:
self.reader = reader
self._sensor = None
self._segment: Optional[Path] = None
#: How many frames the open sensor has already consumed in ``_segment``.
self._position = 0
#: The depth of ``_position``, so asking for the same frame twice is a
#: cache hit instead of a second (impossible) read.
self._last_depth: Optional[np.ndarray] = None
#: True when an earlier frame had to be re-read without a reset boundary
#: to clean up after it, so the state may not match the live run.
self._rewound_without_anchor = False
self.last_error = ""
self.last_warnings: list[str] = []
# ------------------------------------------------------------------ probing
@staticmethod
def sdk_available() -> tuple[bool, str]:
try:
import orisys # noqa: F401
return True, str(getattr(__import__("orisys"), "__version__", "unknown"))
except Exception as exc: # noqa: BLE001
return False, f"{type(exc).__name__}: {exc}"
def config_ref(self) -> str:
session = self.reader.session()
device = session.get("device") or {}
return str(device.get("config_ref") or "")
def resolved_config(self) -> str:
"""The exact configuration file to open, resolved as the live run did.
Reducing the path to a directory name or a stem could silently load a
*different* file with the same name while the hash check still compared
the recorded path — so the full resolved path is what gets opened.
"""
config_ref = self.config_ref()
try:
from orisys_app_shared.acquisition.sdk_worker import load_fixed_mode
return str(load_fixed_mode(config_ref).config_ref)
except Exception: # noqa: BLE001 - reported by the caller as a failure
return config_ref
def preflight(self) -> str:
"""Return an empty string when recomputation can be attempted."""
ok, detail = self.sdk_available()
if not ok:
return f"sdk_unavailable: {detail}"
config_ref = self.config_ref()
if not config_ref:
return "config_missing"
if not Path(config_ref).is_file():
return "config_file_missing"
if not self.reader.segments():
return "no_video"
return ""
# --------------------------------------------------------------- replay plan
def plan_for(self, frame) -> ReplayPlan:
"""Decide where replay must start so ``frame`` is reproducible.
The SDK's depth is the output of a stateful chain (optical-flow tracker,
temporal filters), so a frame cannot be recomputed in isolation. What the
plan states explicitly:
* **where** to start (``start_at``, plus a pre-roll file when there is one);
* which **reset boundary** the replay is anchored to, if any;
* whether the state is actually recoverable — :data:`STATE_EXACT` (the
replay starts at a reset), :data:`STATE_PREROLL` (the recorded pre-roll
precedes it) or :data:`STATE_APPROXIMATE` (neither: the depth map is
still produced, and the caller is told it may differ from live).
Finding a reset event is **not** by itself a recovered state: the reset
frame has to exist in the index *and* in the video. Every intervening
segment must be read with the same processor to preserve its state.
"""
warnings: list[str] = []
segment_path = self.reader.segment_path(frame.segment)
if segment_path is None:
return ReplayPlan(segment=Path(), error="segment_missing")
stop_at = max(1, int(frame.segment_frame))
reset_frame = self.reader.last_reset_frame(frame.frame_id)
if reset_frame is not None:
origin = self.reader.frame_of(reset_frame)
if origin is None:
warnings.append("reset_frame_not_indexed")
elif origin.dropped or not origin.in_video:
# The boundary frame is not in the video, so the state at that
# point cannot be rebuilt: replay from here is an approximation.
warnings.append("reset_frame_not_in_video")
elif any(
entry.dropped or not entry.in_video
for entry in self.reader.info().frames
if origin.frame_id <= entry.frame_id <= frame.frame_id
):
warnings.append("state_missing_frames")
elif origin.segment == frame.segment:
plan = ReplayPlan(
segment=segment_path,
steps=(
ReplayStep(
source=segment_path,
to_frame=stop_at,
reset_after=int(origin.segment_frame),
),
),
stop_at=stop_at,
reset_frame_id=int(reset_frame),
anchor="reset",
state=STATE_EXACT,
)
if plan.frames_to_read > MAX_REPLAY_FRAMES:
warnings.append("state_too_far_back")
else:
return plan
else:
# The reset is in an earlier segment. The processor survives a
# source swap, so the state *can* be carried: read the tail of
# the reset's segment, then continue in the target's.
steps = []
for segment in range(int(origin.segment), int(frame.segment) + 1):
source = self.reader.segment_path(segment)
last = stop_at if segment == frame.segment else self._last_segment_frame(segment)
if source is None or not source.is_file() or last is None:
steps = []
break
steps.append(ReplayStep(
source=source,
to_frame=int(last),
reset_after=int(origin.segment_frame) if segment == origin.segment else None,
))
if not steps:
warnings.append("state_across_segment")
else:
plan = ReplayPlan(
segment=segment_path,
steps=tuple(steps),
stop_at=stop_at,
reset_frame_id=int(reset_frame),
anchor="reset",
state=STATE_EXACT,
)
if plan.frames_to_read <= MAX_REPLAY_FRAMES:
return plan
# Too expensive: report it rather than freeze the window.
warnings.append("state_too_far_back")
start = self.reader.segment_start_frame(frame.segment)
if start is None:
return ReplayPlan(segment=Path(), error="frame_not_indexed")
preroll = self.reader.preroll_segment()
preroll_frames = self.reader.preroll_frames() if preroll is not None else 0
if preroll is not None and preroll_frames > 0 and frame.segment == 1:
# Recording started mid-stream: the frames before the first recorded
# one are the only history that can rebuild the tracker.
return ReplayPlan(
segment=segment_path,
steps=(ReplayStep(source=segment_path, to_frame=stop_at),),
stop_at=stop_at,
preroll=preroll,
preroll_frames=int(preroll_frames),
anchor="preroll",
state=STATE_PREROLL,
warnings=tuple(warnings),
)
# No reset and no pre-roll: the state entering this frame is unknown.
warnings.append("state_approximate")
return ReplayPlan(
segment=segment_path,
steps=(
ReplayStep(
source=segment_path,
to_frame=stop_at,
),
),
stop_at=stop_at,
anchor="segment_start",
state=STATE_APPROXIMATE,
warnings=tuple(dict.fromkeys(warnings)),
)
def _last_segment_frame(self, segment: int) -> Optional[int]:
"""The highest indexed frame number inside ``segment``."""
frames = [
int(entry.segment_frame)
for entry in self.reader.info().frames
if int(entry.segment) == int(segment)
]
return max(frames) if frames else None
# ----------------------------------------------------------------- recompute
def depth_at(self, t_rel_s: float) -> DepthResult:
"""Recompute the depth map for the frame at ``t_rel_s``."""
problem = self.preflight()
if problem:
self.last_error = problem
return DepthResult(error=problem, t_rel_s=float(t_rel_s))
frame = self.reader.frame_at(t_rel_s)
if frame is None:
return DepthResult(error="no_frame", t_rel_s=float(t_rel_s))
if frame.dropped:
# The row was lost to a full queue; the index row marks the hole.
return DepthResult(
error="frame_not_recorded",
frame_id=frame.frame_id,
t_rel_s=frame.t_rel_s,
)
if not frame.in_video:
# The queue dropped this frame; say so rather than show the previous
# frame as if it belonged to the cursor position.
return DepthResult(
error="frame_not_in_video", frame_id=frame.frame_id, t_rel_s=frame.t_rel_s
)
plan = self.plan_for(frame)
if plan.error:
self.last_error = plan.error
return DepthResult(
error=plan.error, frame_id=frame.frame_id, t_rel_s=frame.t_rel_s
)
try:
depth = self._replay(plan)
except Exception as exc: # noqa: BLE001 - reported to the caller
logger.exception("Depth recomputation failed")
self.last_error = f"{type(exc).__name__}: {exc}"
return DepthResult(
error=self.last_error,
frame_id=frame.frame_id,
t_rel_s=frame.t_rel_s,
warnings=plan.warnings,
record_id=self.record_id(),
)
self.last_warnings = list(plan.warnings)
if self._rewound_without_anchor and "state_approximate" not in self.last_warnings:
# The plan looked recoverable, but the read path had to restart a file
# without a reset boundary to clear the tracker afterwards.
self.last_warnings.append("state_approximate_after_rewind")
warnings = tuple(dict.fromkeys(self.last_warnings))
if depth is None:
self.last_error = "depth_not_available"
return DepthResult(
error=self.last_error,
frame_id=frame.frame_id,
t_rel_s=frame.t_rel_s,
warnings=warnings,
record_id=self.record_id(),
)
self.last_error = ""
return DepthResult(
depth=depth,
frame_id=frame.frame_id,
t_rel_s=frame.t_rel_s,
warnings=warnings,
record_id=self.record_id(),
)
def record_id(self) -> str:
"""Which recording these answers belong to."""
return str(getattr(self.reader, "directory", "") or "")
# ------------------------------------------------------------------ playback
def _replay(self, plan: ReplayPlan) -> Optional[np.ndarray]:
"""Execute the plan's steps, keeping the processor and swapping inputs.
The **processor is kept** across a source change: the SDK's tracker lives
on the object, so disposing of it would dispose of the state the pre-roll
(or the reset boundary) just rebuilt. A reset boundary is re-applied by
calling ``reset()`` where the operator's ``zeroing`` happened.
"""
self._rewound_without_anchor = False
if plan.preroll is not None and plan.preroll_frames > 0:
self._feed(plan.preroll, int(plan.preroll_frames))
depth: Optional[np.ndarray] = None
for step in plan.steps:
depth = self._run_step(plan, step)
return depth
def _run_step(self, plan: ReplayPlan, step: ReplayStep) -> Optional[np.ndarray]:
target = max(1, int(step.to_frame))
if self._sensor is None or self._segment != step.source:
if self._sensor is None or not self._swap_source(step.source):
self._open_segment(step.source)
elif self._position > target:
# Re-reading an earlier frame can only be done from the start of the
# file. Without a reset boundary to clean up after it, the tracker has
# now seen frames it should not have.
if not self._swap_source(step.source):
self._open_segment(step.source)
if plan.anchor != "reset":
self._rewound_without_anchor = True
last: Optional[np.ndarray] = (
self._last_depth if self._position == target else None
)
while self._position < target:
depth = self._read_one()
if depth is None:
raise RuntimeError(f"replay source ended before frame {target}: {step.source}")
last = depth
if step.reset_after is not None and self._position == int(step.reset_after):
# The operator's zeroing: the live run cleared the tracker here.
if not self._reset_processor():
self.last_warnings.append("reset_not_applied")
return last
def _reset_processor(self) -> bool:
replay = self._sensor
reset = getattr(replay, "reset", None)
if not callable(reset):
return False
try:
return bool(reset())
except Exception: # noqa: BLE001 - reported as a replay warning
logger.exception("Could not re-apply the reset boundary")
return False
def _depth_from_segment(
self, segment: Path, segment_frame: int
) -> Optional[np.ndarray]:
"""Recompute the map for ``segment_frame`` of ``segment``.
The SDK's video source only advances, so stepping costs one read per
frame — and the position is tracked, so asking for frame 2 and then
frame 3 reads two frames and then one, instead of two more. Asking twice
for the *same* frame is answered from the cached depth of that frame.
"""
target = max(1, int(segment_frame))
if self._segment != segment:
self._open_segment(segment)
return self._run_step(
ReplayPlan(segment=Path(segment), stop_at=target),
ReplayStep(source=Path(segment), to_frame=target),
)
# --------------------------------------------------------------------- video
def sensor_type(self) -> str:
device = self.reader.session().get("device") or {}
return str(device.get("sensor_type") or "box")
def _open_segment(self, segment: Path) -> None:
"""Open a segment on a **new** processor (the previous one is closed)."""
self._close()
self._sensor = open_replay(
self.sensor_type(),
Path(segment),
self.resolved_config(),
self._finger_export_dir(),
)
self._segment = Path(segment)
self._position = 0
self._last_depth = None
def _swap_source(self, segment: Path) -> bool:
"""Replace only the video source of the open processor.
Returns False when the backend cannot do it, in which case the caller
falls back to a full reopen (and the algorithm state is lost, which the
plan already reports as approximate).
"""
replay = self._sensor
if replay is None:
return False
try:
if not replay.set_source(Path(segment)):
return False
except Exception: # noqa: BLE001 - reported by the plan's warnings
logger.exception("Could not swap the replayed video source")
return False
self._segment = Path(segment)
self._position = 0
self._last_depth = None
return True
def _finger_export_dir(self):
"""The finger pipeline assets, resolved exactly as the live run did."""
try:
from orisys_app_shared.acquisition.finger_worker import (
resolve_finger_export_dir,
)
return resolve_finger_export_dir(self.resolved_config())
except Exception: # noqa: BLE001 - the SDK default is used instead
logger.exception("Could not resolve the finger export directory")
return None
def _feed(self, segment: Path, frames: int) -> None:
"""Consume ``frames`` frames from another file to rebuild state."""
if frames <= 0:
return
if self._sensor is None:
self._open_segment(segment)
elif self._segment != segment:
if not self._swap_source(segment):
self._open_segment(segment)
for _ in range(int(frames)):
if self._read_one() is None:
break
def _read_one(self) -> Optional[np.ndarray]:
"""Read one frame through the **live** processing chain (see ``process_chain``)."""
replay = self._sensor
if replay is None:
return None
depth = replay.read()
if depth is None:
return None
self._position += 1
self._last_depth = depth
return depth
def _close(self) -> None:
replay, self._sensor = self._sensor, None
self._segment = None
self._position = 0
self._last_depth = None
if replay is None:
return
try:
replay.close()
except Exception: # noqa: BLE001
pass
def close(self) -> None:
self._close()
def __enter__(self) -> "DepthRecomputer":
return self
def __exit__(self, *_exc) -> None:
self.close()
shared/src/orisys_app_shared/record/depth_backends.py
-194
View File
@@ -1,194 +0,0 @@
"""Replay backends: the live processing chain, pointed at a video file.
Each backend keeps **one** processor for its lifetime. The tracker state lives on
that object, so a replay must be able to change the *input source* without
throwing the processor away — that is what makes a pre-roll, or a zeroing that
happened in an earlier segment, reproducible at all.
The processing itself is not duplicated here: the reads go through
``acquisition.sdk_worker.process_chain``, the same function the live workers run.
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Optional
import numpy as np
logger = logging.getLogger(__name__)
class BoxReplay:
"""The box sensor's live chain, opened on one video file."""
def __init__(self, segment: Path, config_ref: str) -> None:
import orisys
self._config_ref = config_ref
self._sensor = orisys.Sensor(
str(segment),
config_name=config_ref,
verbose=False,
backend="auto",
)
def set_source(self, segment: Path) -> bool:
"""Point the same processor at another file.
The tracker lives on the ``Sensor`` object, so re-opening the sensor would
throw away the state the pre-roll just rebuilt. Only the capture is
replaced, and the tracker's reference frame is preserved.
"""
sensor = self._sensor
open_source = getattr(sensor, "_open_video_source", None)
if not callable(open_source):
return False
initial = getattr(sensor, "initial_img", None)
try:
import orisys
sensor.vid_src = orisys.capture.normalize_source(str(segment))
except Exception: # noqa: BLE001 - fall back to assigning the path
sensor.vid_src = str(segment)
sensor._is_live_camera = False
open_source(False, None)
if initial is not None:
# The tracker's reference frame belongs to the session, not to the
# file: replacing it would silently reset what "no contact" means.
sensor.initial_img = initial
return True
def read(self) -> Optional[np.ndarray]:
from orisys_app_shared.acquisition.sdk_worker import process_chain
image = self._sensor.get_img()
if image is None:
return None
# The same function the acquisition loop runs.
depth, error = process_chain(self._sensor)
if error:
logger.warning("Replay processing chain reported: %s", error)
return depth
def reset(self) -> bool:
"""Clear the tracker, exactly as the operator's zeroing did live."""
reset = getattr(self._sensor, "reset", None)
if not callable(reset):
return False
reset()
return True
def close(self) -> None:
self._sensor.disconnect()
class FingerReplay:
"""The finger's live chain: unroll the source frame, then process the tactile."""
def __init__(self, segment: Path, config_ref: str, export_dir) -> None:
from orisys.finger import FingerRuntime
from orisys_app_shared.acquisition.constants import (
FORCE_SAMPLE_RATE_HZ,
LOWPASS_CUTOFF_HZ,
)
self._config_ref = config_ref
self._export_dir = export_dir
kwargs: dict = {
"config": config_ref,
"video": str(segment),
"verbose": False,
}
if export_dir is not None:
kwargs["export_dir"] = str(export_dir)
self._runtime = FingerRuntime.open(**kwargs)
self._runtime.warmup(color=False)
tactile = getattr(self._runtime, "tactile", None)
if tactile is not None and hasattr(tactile, "configure"):
try:
# The same filter settings the acquisition loop applies.
tactile.configure(
lowpass_cutoff_hz=LOWPASS_CUTOFF_HZ,
lowpass_sample_rate_hz=FORCE_SAMPLE_RATE_HZ,
)
except Exception: # noqa: BLE001 - the defaults still work
logger.exception("Could not configure the replayed tactile sensor")
def set_source(self, segment: Path) -> bool:
"""Point the same runtime at another video, keeping the tactile pipeline."""
try:
from orisys.capture import (
CaptureConfig,
ReconnectPolicy,
is_live_camera,
normalize_source,
open_capture,
)
runtime = self._runtime
video = normalize_source(str(segment))
camera = dict(getattr(runtime.geometry, "config", {}).get("camera", {}))
capture_config = CaptureConfig.from_camera_dict(
video,
camera,
reconnect=ReconnectPolicy(enabled=False),
verbose=False,
background=False,
)
capture_source = open_capture(capture_config, start_background=False)
if not capture_source.is_open():
return False
if is_live_camera(video):
capture_source.start_background()
previous = runtime.capture_source
runtime.capture_source = capture_source
runtime.capture = capture_source.cap
runtime.capture_worker = capture_source
if previous is not None:
try:
previous.close()
except Exception: # noqa: BLE001
logger.exception("Could not close the previous finger source")
return True
except Exception: # noqa: BLE001 - reported as an approximate replay
logger.exception("Could not swap the finger video source")
return False
def read(self) -> Optional[np.ndarray]:
from orisys_app_shared.acquisition.sdk_worker import process_chain
ok, source_frame = self._runtime.read_source()
if not ok or source_frame is None:
return None
unrolled = self._runtime.process_frame(source_frame, color=False)
if unrolled is None:
return None
depth, error = process_chain(self._runtime.tactile)
if error:
logger.warning("Replay processing chain reported: %s", error)
return depth
def reset(self) -> bool:
"""Clear the tactile tracker, as the operator's zeroing did live."""
tactile = getattr(self._runtime, "tactile", None)
target = tactile if callable(getattr(tactile, "reset", None)) else None
if target is None:
sensor = getattr(tactile, "raw_sensor", None)
target = sensor if callable(getattr(sensor, "reset", None)) else None
if target is None:
return False
target.reset()
return True
def close(self) -> None:
self._runtime.close()
def open_replay(sensor_type: str, segment: Path, config_ref: str, export_dir=None):
"""The backend for a recorded sensor type."""
if str(sensor_type) == "finger":
return FingerReplay(segment, config_ref, export_dir)
return BoxReplay(segment, config_ref)
shared/src/orisys_app_shared/record/playback.py
+67 -14
View File
@@ -15,6 +15,7 @@ from __future__ import annotations
import csv
import json
import logging
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterator, Optional
@@ -144,6 +145,14 @@ class RecordingReader:
self.max_rows = int(max_rows or self.DEFAULT_MAX_ROWS)
self._index = RecordingIndex(self.directory, checkpoint_rows=checkpoint_rows)
self._overviews: dict[tuple[str, int], Overview] = {}
#: One open decoder per segment, kept between calls. Opening the file and
#: seeking costs 10 ms for a small clip and hundreds for 1080p H.264, and
#: playback asks for a frame 30 times a second: reopening per frame is
#: what made the video stand still.
self._capture = None
self._capture_segment: int | None = None
self._capture_index = -1
self._capture_lock = threading.Lock()
# ------------------------------------------------------------------ loading
def info(self) -> RecordingInfo:
@@ -348,24 +357,70 @@ class RecordingReader:
Returns ``(frame_bgr, RecordingFrame)`` or ``(None, frame)`` when the
frame was not written to video (the queue dropped it).
The decoder stays open and is *stepped* when the requested frame follows
the previous one, which is what playback does: a seek costs a keyframe
decode, a step costs one frame. Seeking is kept for jumps (dragging the
cursor).
"""
import cv2
frame = self.frame_at(t_rel_s)
if frame is None or not frame.in_video:
return None, frame
path = self.segment_path(frame.segment)
target = max(0, int(frame.segment_frame) - 1)
with self._capture_lock:
capture = self._capture_for(frame.segment)
if capture is None:
return None, frame
try:
if self._capture_index >= 0 and target == self._capture_index + 1:
ok, image = capture.read()
else:
capture.set(cv2.CAP_PROP_POS_FRAMES, target)
ok, image = capture.read()
except Exception: # noqa: BLE001 - a broken decode is not a crash
logger.exception("Could not decode frame %s", target)
self._release_capture()
return None, frame
# A failed read leaves the decoder's position unknown: start over next
# time rather than stepping from a frame that was never delivered.
self._capture_index = target if ok else -1
return (image if ok else None), frame
def _capture_for(self, segment: int):
"""The open decoder for ``segment``, opening (or switching) as needed."""
import cv2
if self._capture is not None and self._capture_segment == int(segment):
return self._capture
self._release_capture()
path = self.segment_path(int(segment))
if path is None:
return None, frame
return None
capture = cv2.VideoCapture(str(path))
if not capture.isOpened():
return None, frame
try:
capture.set(cv2.CAP_PROP_POS_FRAMES, max(0, frame.segment_frame - 1))
ok, image = capture.read()
return (image if ok else None), frame
finally:
capture.release()
return None
self._capture = capture
self._capture_segment = int(segment)
self._capture_index = -1
return capture
def _release_capture(self) -> None:
if self._capture is not None:
try:
self._capture.release()
except Exception: # noqa: BLE001 - releasing is best effort
logger.debug("Could not release the replay decoder", exc_info=True)
self._capture = None
self._capture_segment = None
self._capture_index = -1
def close(self) -> None:
"""Release the decoder. The reader is reusable afterwards."""
with self._capture_lock:
self._release_capture()
def forces_window(self, t0: float, t1: float) -> tuple[np.ndarray, np.ndarray]:
"""``(times, values)`` inside ``[t0, t1]`` for the raw triplet."""
@@ -445,16 +500,16 @@ class RecordingReader:
def replay_warnings(self, *, current_config_sha: str = "", sdk_version: str = "") -> list[str]:
"""What could stop this recording from replaying faithfully.
Depth is recomputed rather than stored, so a changed config or SDK means
the depth seen on replay may differ from what was shown live. That must
be stated, not discovered later.
A changed config or SDK means the recorded frames are read by a different
pipeline than the one that produced them; that must be stated, not
discovered later.
"""
session = self.info().session
warnings: list[str] = []
recorded_sha = str((session.get("device") or {}).get("config_sha256") or "")
if not current_config_sha:
# Compare against the config file as it is now, which is exactly what
# the recomputation would use.
# a recompute over the video would use.
current_config_sha = self.config_sha256(self.recorded_config_ref())
if recorded_sha and current_config_sha and recorded_sha != current_config_sha:
warnings.append("config_changed")
@@ -464,8 +519,6 @@ class RecordingReader:
warnings.append("sdk_changed")
if recorded_sha and not current_config_sha:
warnings.append("config_missing")
if not self.depth_is_stored():
warnings.append("depth_recomputed")
if self.info().dropped_rows:
warnings.append("rows_skipped")
if self.info().frames_dropped: