ykcheungab
1aa454311f
Replay moved the cursor onto the result's timeline, so the recording's own picture was not offered at all and ``use_recording()`` had no caller. Adopting a result now keeps the cursor on the recording, opens the panel on a derived view, and serves scrubs, playback and display changes from the cache (Part B §9). Verified: apps/studio 455 items / 0 failed / 18 skipped, shared 427 / 0 / 1, autocal 358, ui_check 628 checks / 0 failures, lint gate clean.
1198 lines
45 KiB
Python
1198 lines
45 KiB
Python
"""Part B: playback served from an analysis result.
|
|
|
|
What is being defended, in the order the plan asked for it:
|
|
|
|
1. **one timeline.** The charts, the 2D canvas and the 3D pane read one cursor, and a
|
|
direct seek to frame *N* produces the same frame as playing forward to *N*;
|
|
2. **analysis on open.** A recording opens on a compatible stored result, and runs the
|
|
pass (with a progress and cancel surface) only when there is none;
|
|
3. **a display change writes nothing.** The flow threshold, the colour range and the
|
|
idle dots are reads: the result directory set is unchanged;
|
|
4. **an incomplete result is never served.** A cancelled or failed pass is reported as
|
|
cancelled or failed, and playback is not offered from it;
|
|
5. **frames are addressed by id.** An id the recording does not hold is refused, never
|
|
coerced into an index.
|
|
|
|
The fixtures use real footage — the 33-frame box recording under ``dist/`` and the
|
|
shipped finger configuration — driven through the **finger** chain, which is what gives
|
|
the result a validity mask and therefore a static candidate grid. The pass is O(5 s)
|
|
and is shared per module. Where the SDK or the footage is absent the tests skip rather
|
|
than fake a result: a synthetic result would not defend the id span (20…52), the
|
|
keyframe identity or the sample round trip.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from orisys_app_shared.acquisition.arrows import depth_map_to_bgr, flow_colormap
|
|
from orisys_app_shared.acquisition.image_backend import (
|
|
ARROW_PURE,
|
|
ARROW_STITCHED,
|
|
DEPTH,
|
|
DERIVED_MODES,
|
|
RAW_NOTE_KEY,
|
|
SOURCE_ANALYSIS,
|
|
SOURCE_REPLAY,
|
|
ImageFrame,
|
|
plan_image,
|
|
render_image,
|
|
)
|
|
from orisys_app_shared.analysis.cache import ResultCache, scan_results
|
|
from orisys_app_shared.analysis.reader import (
|
|
AnalysisIncompleteError,
|
|
AnalysisNotFoundError,
|
|
AnalysisReader,
|
|
)
|
|
from orisys_app_shared.analysis.service import AnalysisService, ChainSpec
|
|
from orisys_app_shared.i18n import t
|
|
from orisys_app_shared.scene3d.contracts import DisplaySettings
|
|
from orisys_app_shared.scene3d.visual_policy import (
|
|
policy_for_settings,
|
|
select_flow_samples,
|
|
)
|
|
from orisys_studio.analysis_worker import AnalysisWorker, report_from
|
|
from orisys_studio.replay import MODE_ANALYSIS, MODE_RECORDING
|
|
from orisys_studio.replay_analysis import (
|
|
AnalysisSession,
|
|
ResultLibrary,
|
|
ResultPolicyError,
|
|
)
|
|
|
|
REPO = Path(__file__).resolve().parents[3]
|
|
STUDIO = REPO / "dist" / "OrisysStudio"
|
|
RECORDING = (
|
|
STUDIO
|
|
/ "outputs"
|
|
/ "recordings"
|
|
/ "20260930T015028Z_session0002_40cb8e7441434feab48bb646afdba011"
|
|
)
|
|
FINGER_CONFIG = STUDIO / "config" / "sensors" / "finger" / "sensor.json"
|
|
FINGER_EXPORT = STUDIO / "config" / "sensors" / "finger" / "assets" / "30fps-0703"
|
|
|
|
#: The evidence recording's own frame ids. Deliberately not 0…32: an implementation
|
|
#: that used an index would pass every test written against a 0-based recording.
|
|
FIRST_FRAME_ID = 20
|
|
LAST_FRAME_ID = 52
|
|
|
|
|
|
def _require_inputs() -> None:
|
|
missing = [
|
|
str(path)
|
|
for path in (RECORDING / "frames.csv", RECORDING / "raw_0001.mp4", FINGER_CONFIG)
|
|
if not path.exists()
|
|
]
|
|
if missing:
|
|
pytest.skip("analysis inputs are not present: " + ", ".join(missing))
|
|
pytest.importorskip("orisys", reason="the SDK is needed to run the analysis pass")
|
|
|
|
|
|
def _finger_chain() -> ChainSpec:
|
|
return ChainSpec(
|
|
sensor_type="finger", config_ref=FINGER_CONFIG, export_dir=FINGER_EXPORT
|
|
)
|
|
|
|
|
|
def _service(root: Path, **kwargs) -> AnalysisService:
|
|
return AnalysisService(RECORDING, chain=_finger_chain(), cache=ResultCache(root), **kwargs)
|
|
|
|
|
|
# ------------------------------------------------------------------------ fixtures
|
|
@pytest.fixture(scope="module")
|
|
def pass_root(tmp_path_factory) -> Path:
|
|
"""One real analysis pass over the evidence recording, in its own cache root."""
|
|
_require_inputs()
|
|
|
|
root = tmp_path_factory.mktemp("replay-analysis-b")
|
|
outcome = _service(root).run()
|
|
assert outcome.complete, outcome.reason
|
|
return root
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def reader(pass_root: Path):
|
|
reader = AnalysisReader(_result_directory(pass_root))
|
|
yield reader
|
|
reader.close()
|
|
|
|
|
|
def _result_directory(root: Path) -> Path:
|
|
directories = [entry.directory for entry in scan_results(root) if entry.complete]
|
|
assert directories, f"no complete result under {root}"
|
|
return directories[0]
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def session(reader) -> AnalysisSession:
|
|
service = _service(Path(reader.directory).parent)
|
|
return AnalysisSession(
|
|
reader,
|
|
policy=service.policy,
|
|
display_policy=policy_for_settings(DisplaySettings()),
|
|
session_id=2,
|
|
recording=RECORDING,
|
|
)
|
|
|
|
|
|
def _same_samples(left, right) -> dict[str, bool]:
|
|
"""Compare two ``FlowSamples`` (or two frames that carry one) array by array."""
|
|
left = getattr(left, "samples", left)
|
|
right = getattr(right, "samples", right)
|
|
return {
|
|
name: bool(
|
|
np.array_equal(
|
|
np.asarray(getattr(left, name)), np.asarray(getattr(right, name))
|
|
)
|
|
)
|
|
for name in ("rows", "cols", "dx", "dy", "active")
|
|
}
|
|
|
|
|
|
# ------------------------------------------------------------------- fakes (Qt)
|
|
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 = "") -> None:
|
|
self.btn_run = _Button()
|
|
self._video_path = video_path
|
|
self.run_checked: list[bool] = []
|
|
self.source_modes: list[str] = []
|
|
self.recompute_enabled: list[bool] = []
|
|
self.recompute_checked: list[bool] = []
|
|
|
|
def run_role(self) -> str:
|
|
return "playback"
|
|
|
|
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:
|
|
pass
|
|
|
|
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) -> None:
|
|
self.cursors: list[float] = []
|
|
self._recording = False
|
|
self.cleared = 0
|
|
self.unit_value = "raw"
|
|
|
|
def unit(self) -> str:
|
|
return self.unit_value
|
|
|
|
def set_cursor(self, position: float) -> None:
|
|
self.cursors.append(float(position))
|
|
|
|
def set_recording_source(self, reader) -> None:
|
|
self._recording = reader is not None
|
|
|
|
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:
|
|
pass
|
|
|
|
def show_all(self) -> None:
|
|
pass
|
|
|
|
|
|
class _ImagePanel:
|
|
def __init__(self) -> None:
|
|
self.kinds: list[str] = []
|
|
self.cleared: list[str] = []
|
|
#: The 2D mode selector's current value. ``recorded`` is what the panel starts
|
|
#: on for a recording, so a stub that never sets it behaves as before.
|
|
self._mode = "recorded"
|
|
|
|
def set_source_kind(self, kind: str) -> None:
|
|
self.kinds.append(str(kind))
|
|
|
|
def view_mode(self) -> str:
|
|
return self._mode
|
|
|
|
def set_view_mode(self, mode: str) -> None:
|
|
self._mode = str(mode)
|
|
|
|
def set_plan(self, plan) -> None:
|
|
self.plan = plan
|
|
|
|
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
|
|
self._settings = DisplaySettings()
|
|
|
|
def set_source_kind(self, kind: str) -> None:
|
|
self.kinds.append(str(kind))
|
|
|
|
def show_reference_only(self) -> None:
|
|
self.reference_only += 1
|
|
|
|
def display_settings(self) -> DisplaySettings:
|
|
return self._settings
|
|
|
|
def set_display_settings(self, settings: DisplaySettings) -> None:
|
|
self._settings = settings
|
|
|
|
|
|
class _Progress:
|
|
"""The progress surface as the controller uses it, recording every transition."""
|
|
|
|
def __init__(self) -> None:
|
|
self.cancel_requested = _Signal()
|
|
self.states: list[str] = []
|
|
self.updates: list[dict] = []
|
|
self.cancels = 0
|
|
|
|
def start(self) -> None:
|
|
self.states.append("running")
|
|
|
|
def progress(self, **kwargs) -> None:
|
|
self.updates.append(dict(kwargs))
|
|
|
|
def reused(self, **kwargs) -> None:
|
|
self.states.append("reused")
|
|
|
|
def complete(self, **kwargs) -> None:
|
|
self.states.append("complete")
|
|
|
|
def cancelled(self, **kwargs) -> None:
|
|
self.states.append("cancelled")
|
|
|
|
def failed(self, **kwargs) -> None:
|
|
self.states.append("failed")
|
|
|
|
def clear(self) -> None:
|
|
self.states.append("idle")
|
|
|
|
|
|
class _Host:
|
|
def __init__(self, service_factory=None) -> None:
|
|
self.prefs: dict = {}
|
|
self.view = "replay"
|
|
self._running = False
|
|
self._preview_open = False
|
|
self._session_active = False
|
|
self.notes: list[str] = []
|
|
self.frames: list = []
|
|
self.analysis_packets: list = []
|
|
self.units: list[str] = []
|
|
self.presets: list[str] = []
|
|
self.service_factory = service_factory
|
|
self.services_built = 0
|
|
|
|
def _in_replay_view(self) -> bool:
|
|
return self.view == "replay"
|
|
|
|
def _current_source_kind(self) -> str:
|
|
return "replay"
|
|
|
|
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:
|
|
pass
|
|
|
|
def _show_image_frame(self, frame) -> None:
|
|
self.frames.append(frame)
|
|
|
|
def _show_analysis_frame(self, packet) -> None:
|
|
self.analysis_packets.append(packet)
|
|
|
|
def _drop_render_marker(self) -> None:
|
|
pass
|
|
|
|
def _notify(self, text: str, ms: int = 0) -> None:
|
|
self.notes.append(str(text))
|
|
|
|
def _start_run(self) -> None:
|
|
pass
|
|
|
|
def _stop_run(self, *, reason: str = "user_stop", wait: bool = False) -> None:
|
|
pass
|
|
|
|
def _update_run_enabled(self) -> None:
|
|
pass
|
|
|
|
def _refresh_status_rail(self) -> None:
|
|
pass
|
|
|
|
def _analysis_service(self, recording: Path, cancelled=None, *, on_progress=None):
|
|
self.services_built += 1
|
|
if self.service_factory is None:
|
|
raise AssertionError("the test did not provide an analysis service")
|
|
return self.service_factory(recording, cancelled, on_progress)
|
|
|
|
|
|
class _RecordingReader:
|
|
"""The recording reader as the engine and the controller use it."""
|
|
|
|
def __init__(self, directory: Path) -> None:
|
|
self.directory = Path(directory)
|
|
self.closed = False
|
|
|
|
def session(self) -> dict:
|
|
return {"session_id": 2, "device": {"sensor_type": "finger"}}
|
|
|
|
def segments(self) -> list[Path]:
|
|
return [self.directory / "raw_0001.mp4"]
|
|
|
|
def info(self):
|
|
return type("Info", (), {"duration_s": 1.92, "newtons": None, "decimated": False})()
|
|
|
|
def video_frame_at(self, t_rel_s: float):
|
|
from orisys_app_shared.record import RecordingFrame
|
|
|
|
index = int(t_rel_s * 30)
|
|
frame = RecordingFrame(
|
|
frame_id=FIRST_FRAME_ID + index,
|
|
t_rel_s=float(t_rel_s),
|
|
in_video=True,
|
|
dropped=False,
|
|
segment=1,
|
|
)
|
|
return np.zeros((4, 4, 3), dtype=np.uint8), frame
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def _controller(engine=None, host=None, **kwargs):
|
|
from orisys_studio.replay_controller import ReplayController
|
|
|
|
engine = engine if engine is not None else _Engine()
|
|
host = host if host is not None else _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()
|
|
progress = kwargs.pop("progress", None) or _Progress()
|
|
controller = ReplayController(
|
|
engine=engine,
|
|
host=host,
|
|
toolbar=toolbar,
|
|
charts=charts,
|
|
image_panel=panel,
|
|
mesh_panel=mesh,
|
|
analysis_progress=progress,
|
|
)
|
|
actions = {
|
|
"open_video": _Action(),
|
|
"playback": _Action(),
|
|
"latest": _Action(),
|
|
"play": _Action(),
|
|
"recompute": _Action(),
|
|
}
|
|
controller.bind_actions(**actions)
|
|
return controller, host, engine, (toolbar, charts, panel, mesh, progress, actions)
|
|
|
|
|
|
class _Engine:
|
|
"""The engine as the controller uses it, with the timeline's mode explicit."""
|
|
|
|
def __init__(self, *, duration: float = 1.92, position: float = 0.0) -> None:
|
|
self._duration = duration
|
|
self._position = position
|
|
self._mode = MODE_RECORDING
|
|
self._session: AnalysisSession | None = None
|
|
self._reader = _RecordingReader(RECORDING)
|
|
self._has_video = True
|
|
self.positions: list[float] = []
|
|
self.redraws = 0
|
|
|
|
def has_video(self) -> bool:
|
|
return self._has_video
|
|
|
|
def is_loaded(self) -> bool:
|
|
return True
|
|
|
|
def is_video_only(self) -> bool:
|
|
return False
|
|
|
|
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 directory(self):
|
|
return self._reader.directory
|
|
|
|
def video_path(self):
|
|
return None
|
|
|
|
def video_frame_bgr(self):
|
|
return np.zeros((4, 4, 3), dtype=np.uint8)
|
|
|
|
def video_message(self) -> str:
|
|
return ""
|
|
|
|
def set_unit(self, unit: str) -> None:
|
|
pass
|
|
|
|
def mode(self) -> str:
|
|
return self._mode
|
|
|
|
def analysis(self):
|
|
return self._session
|
|
|
|
def use_analysis(self, session) -> None:
|
|
self._session = session
|
|
self._mode = MODE_ANALYSIS
|
|
|
|
def use_recording(self) -> None:
|
|
self._session = None
|
|
self._mode = MODE_RECORDING
|
|
|
|
def analysed_frame(self):
|
|
if self._session is None:
|
|
return None
|
|
return self._session.frame_at(self._position)
|
|
|
|
def redraw(self) -> None:
|
|
self.redraws += 1
|
|
|
|
def open_recording(self, directory) -> bool:
|
|
self._reader = _RecordingReader(Path(directory))
|
|
self._session = None
|
|
self._mode = MODE_RECORDING
|
|
return True
|
|
|
|
def open_video(self, path) -> bool:
|
|
return False
|
|
|
|
|
|
# ------------------------------------------------------------------ the timeline
|
|
def test_a_stored_result_is_served_with_the_recording_s_own_frame_ids(session):
|
|
"""Frame identity is the recording's counter (20…52), never an index (0…32)."""
|
|
assert session.frame_count == 33
|
|
assert int(session.frame_ids[0]) == FIRST_FRAME_ID
|
|
assert int(session.frame_ids[-1]) == LAST_FRAME_ID
|
|
assert not session.has(0), "an index was accepted as a frame id"
|
|
assert session.has(FIRST_FRAME_ID)
|
|
|
|
|
|
def test_a_frame_id_the_recording_does_not_hold_is_refused(session):
|
|
"""A *position* passed where an *id* belongs gets an error, not a picture.
|
|
|
|
This is the trap the plan names: the result's ids run 20…52, so a caller that
|
|
passes an index reads a frame that exists and looks right. The reader refuses an id
|
|
it does not hold, and the session does not repair the request into one it can serve.
|
|
"""
|
|
with pytest.raises(AnalysisNotFoundError):
|
|
session.frame(int(session.frame_ids[0]) + 1000)
|
|
with pytest.raises(AnalysisNotFoundError):
|
|
session.frame(0)
|
|
|
|
|
|
def test_position_is_signal_time_and_resolves_to_a_stored_frame_id(session):
|
|
"""The cursor's time base is ``t_rel_s`` from the recording's own ``frames.csv``."""
|
|
span = session.duration_s()
|
|
assert span == pytest.approx(1.920, abs=1e-3), (
|
|
"the result's span must be signal time; 1.100 s would be the container's"
|
|
)
|
|
middle = int(session.frame_ids[len(session.frame_ids) // 2])
|
|
moment = session.time_of(middle)
|
|
assert moment is not None
|
|
assert session.frame_id_at(moment) == middle
|
|
assert session.frame_id_at(0.0) == FIRST_FRAME_ID
|
|
assert session.time_of(0) is None, "an absent id answered with a time"
|
|
|
|
|
|
def test_the_recovered_field_reselects_to_the_stored_samples(session):
|
|
"""The 2D canvas and the 3D pane draw the same samples, frame for frame.
|
|
|
|
The canvas takes a flow field and re-runs the shared selector; the pane takes the
|
|
stored selection. This asserts they agree on every stored frame — the locations, the
|
|
drawn offsets and the activity verdict — so "one timeline, two figures" is a
|
|
measurement rather than a hope. Magnitude is compared as float32-derived truth: it
|
|
differs from the stored value by at most one ulp, which does not move a sample.
|
|
"""
|
|
policy = session.display_policy
|
|
checked = 0
|
|
for frame_id in session.frame_ids:
|
|
frame = session.frame(int(frame_id))
|
|
reselected = select_flow_samples(frame.flow, frame.valid_mask, policy)
|
|
agreement = _same_samples(frame.samples, reselected)
|
|
assert all(agreement.values()), f"frame {frame_id}: {agreement}"
|
|
assert np.array_equal(
|
|
np.asarray(frame.samples.active), np.asarray(reselected.active)
|
|
)
|
|
checked += 1
|
|
assert checked == 33
|
|
|
|
|
|
def test_a_seek_equals_playing_forward_to_the_same_frame(session):
|
|
"""Seek to *N* and play to *N* must produce the same picture.
|
|
|
|
Tolerance: none on the frame's identity (the frame id and every drawn array are
|
|
compared exactly). The cursor is a float in seconds, so the *time* handed to a seek
|
|
is the stored ``t_rel_s`` read back from the result: the comparison is exact because
|
|
the seek is given the frame's own time, and the frame id it resolves to is then the
|
|
frame that was played to. Positions between stored frames resolve to the frame at or
|
|
before them, which is what playing forward passes through as well.
|
|
"""
|
|
ids = [int(value) for value in session.frame_ids]
|
|
# Play forward from the first frame, remembering what each frame looked like.
|
|
played: dict[int, object] = {}
|
|
for frame_id in ids:
|
|
played[frame_id] = session.frame(frame_id)
|
|
# Now seek directly to a sample of positions, in a fresh session so nothing is
|
|
# answered from the sequential walk's own cache.
|
|
fresh = AnalysisSession(
|
|
AnalysisReader(_result_directory(Path(session.directory).parent)),
|
|
policy=session.policy,
|
|
display_policy=session.display_policy,
|
|
)
|
|
try:
|
|
for frame_id in (ids[0], ids[1], ids[len(ids) // 2], ids[-2], ids[-1]):
|
|
moment = fresh.time_of(frame_id)
|
|
assert moment is not None
|
|
seeked = fresh.frame_at(moment)
|
|
assert seeked is not None
|
|
assert seeked.frame_id == frame_id
|
|
assert all(_same_samples(played[frame_id], seeked).values())
|
|
assert np.array_equal(
|
|
np.asarray(played[frame_id].flow),
|
|
np.asarray(seeked.flow),
|
|
equal_nan=True,
|
|
)
|
|
finally:
|
|
fresh.reader.close()
|
|
|
|
|
|
def test_the_3d_pane_is_fed_the_same_frame_the_canvas_draws(session):
|
|
"""One cursor, two figures: the packet's id, flow and mask are the frame's own."""
|
|
frame = session.frame_at(session.time_of(LAST_FRAME_ID - 2))
|
|
assert frame is not None
|
|
packet = session.packet(frame)
|
|
assert int(packet.frame_id) == int(frame.frame_id)
|
|
# ``equal_nan``: every cell that is not a recorded candidate is NaN by construction,
|
|
# in both the frame and the packet, and that is the contract rather than a defect.
|
|
assert np.array_equal(
|
|
np.asarray(packet.flow), np.asarray(frame.flow), equal_nan=True
|
|
)
|
|
assert np.array_equal(np.asarray(packet.valid_mask), np.asarray(frame.valid_mask))
|
|
assert packet.flow_valid is True and packet.flow_frame_id == int(frame.frame_id)
|
|
assert packet.forces_valid is False, "a result stores no force, so none may be claimed"
|
|
assert float(packet.mono_t) == pytest.approx(frame.t_rel_s)
|
|
|
|
|
|
# --------------------------------------------------------------- display settings
|
|
def test_a_display_change_is_a_read_and_writes_no_result(pass_root, session):
|
|
"""The result key is blind to display settings, so a change must not write.
|
|
|
|
The assertion is the strongest one available: the set of result directories is
|
|
compared before and after the threshold, the colour range and the idle dots are
|
|
changed, and a fresh service (the object that would write) is asked whether a
|
|
compatible result exists. Nothing may appear.
|
|
"""
|
|
before = {entry.directory.name for entry in scan_results(pass_root)}
|
|
recorded = int(session.reader.frame(int(session.frame_ids[0])).active_count)
|
|
|
|
for settings in (
|
|
DisplaySettings(flow_on_px=0.2),
|
|
DisplaySettings(flow_on_px=1e9),
|
|
# The colour range and the exaggeration are the same class of setting: display
|
|
# only, so they must not reach the cache either. ``color_scale`` needs a
|
|
# ``FieldDescriptor`` and a LUT id, so the field that is constructible on its own
|
|
# stands in for it here; the threshold above is the one that also changes what is
|
|
# drawn.
|
|
DisplaySettings(display_gain=4.0),
|
|
DisplaySettings(flow_grid_spacing=10),
|
|
):
|
|
assert session.serve_policy(policy_for_settings(settings)) is True
|
|
|
|
after = {entry.directory.name for entry in scan_results(pass_root)}
|
|
assert after == before, "a display change wrote a new analysis result"
|
|
# The same cache, asked the same question, still answers with the stored result:
|
|
# nothing about the display entered the key.
|
|
service = _service(pass_root)
|
|
assert service.stored_result() is not None
|
|
# And the change really did change the picture: a threshold of 1e9 activates nothing.
|
|
session.serve_policy(policy_for_settings(DisplaySettings(flow_on_px=1e9)))
|
|
assert session.frame(int(session.frame_ids[0])).active_count == 0
|
|
session.serve_policy(policy_for_settings(DisplaySettings(flow_on_px=0.4)))
|
|
assert session.frame(int(session.frame_ids[0])).active_count == recorded
|
|
|
|
|
|
def test_a_display_change_that_moves_the_grid_is_refused_not_served(session):
|
|
"""The stored candidate locations are the pass's; a new grid cannot be served."""
|
|
assert session.serve_policy(policy_for_settings(DisplaySettings(max_arrows=50))) is False
|
|
assert session.threshold() == pytest.approx(0.4), "a refused policy changed the session"
|
|
# The threshold is the one setting that is served from the stored magnitude.
|
|
assert session.serve_policy(policy_for_settings(DisplaySettings(flow_on_px=0.3))) is True
|
|
|
|
|
|
def _flush(qt_app=None) -> None:
|
|
"""Deliver the worker's queued signals: they arrive through the GUI event loop."""
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
QApplication.processEvents()
|
|
|
|
|
|
def test_the_controller_redraws_on_a_display_change_and_runs_no_pass(pass_root):
|
|
"""Through the controller: the surface shows the redraw and no service is built.
|
|
|
|
Its own reader and session, so neither the threshold this test sets nor the frame
|
|
cache is shared with another test.
|
|
"""
|
|
reader = AnalysisReader(_result_directory(pass_root))
|
|
session = AnalysisSession(
|
|
reader, display_policy=policy_for_settings(DisplaySettings())
|
|
)
|
|
try:
|
|
controller, host, engine, parts = _controller()
|
|
controller.results.open_session(session)
|
|
engine.use_analysis(session)
|
|
engine.redraws = 0
|
|
built = host.services_built
|
|
controller.on_display_settings(DisplaySettings(flow_on_px=0.5))
|
|
assert engine.redraws == 1, "the display change did not redraw"
|
|
assert host.services_built == built, "a display change built an analysis service"
|
|
# 0.5, not the 0.5 that arrived: ``flow_off_px`` is 0.3 and ``normalised()`` keeps
|
|
# on >= off, so a threshold below the off band is raised to it rather than served
|
|
# as an inverted band. The point here is that the session *adopted* the value.
|
|
assert session.threshold() == pytest.approx(0.5)
|
|
# The frame is rebuilt from the same stored record, not from a new one.
|
|
assert session.frame(int(session.frame_ids[0])).frame_id == FIRST_FRAME_ID
|
|
finally:
|
|
session.reader.close()
|
|
|
|
|
|
# ------------------------------------------------------------ analysis on open
|
|
def test_opening_a_recording_with_a_stored_result_runs_no_pass(pass_root):
|
|
"""Analysis on open reuses a compatible result: no progress, no pass."""
|
|
_require_inputs()
|
|
stored = _result_directory(pass_root)
|
|
calls: list[str] = []
|
|
|
|
def factory(recording: Path, cancelled, on_progress):
|
|
calls.append(str(recording))
|
|
return _service(pass_root)
|
|
|
|
host = _Host(service_factory=factory)
|
|
controller, host, engine, parts = _controller(host=host)
|
|
assert controller.load_recording(RECORDING) is True
|
|
assert calls == [str(RECORDING)], f"the service was built {len(calls)} times"
|
|
# Adoption is "the recording has a readable result", not "the cursor moved onto it":
|
|
# the engine keeps the recording's own timeline (Part B §9), so the session is held
|
|
# by the library and answered through ``available_result``.
|
|
assert controller.available_result() is not None, "the stored result was not adopted"
|
|
assert engine.mode() == MODE_RECORDING, "adopting a result moved the cursor's timeline"
|
|
assert parts[2].kinds[-1] == "replay", "the 2D panel was not told the cursor is a replay"
|
|
assert parts[3].kinds[-1] == "analysis", "the 3D pane was not told it is a result"
|
|
assert parts[4].states[-1] == "reused", "the surface did not report the reuse"
|
|
assert stored.is_dir()
|
|
|
|
|
|
def test_opening_a_recording_without_a_result_starts_the_pass(tmp_path, qt_app):
|
|
"""No compatible result: the pass runs behind a progress surface with a cancel."""
|
|
_require_inputs()
|
|
empty = tmp_path / "cache"
|
|
empty.mkdir()
|
|
|
|
def factory(recording: Path, cancelled, on_progress):
|
|
return _service(empty, cancelled=cancelled, on_progress=on_progress)
|
|
|
|
host = _Host(service_factory=factory)
|
|
controller, host, engine, parts = _controller(host=host)
|
|
assert controller.load_recording(RECORDING) is True
|
|
assert parts[4].states[:1] == ["running"], "the progress surface did not start"
|
|
assert controller.analysis_running(), "no pass was started"
|
|
assert controller.wait_for_analysis(120)
|
|
_flush()
|
|
assert parts[4].states[-1] == "complete", parts[4].states
|
|
assert controller.available_result() is not None, "a complete result was not adopted"
|
|
assert engine.mode() == MODE_RECORDING, "a finished pass moved the cursor's timeline"
|
|
|
|
|
|
def test_a_cancelled_pass_is_reported_and_never_served(tmp_path, qt_app):
|
|
"""Cancel mid-pass: the surface says so and playback is not offered from it."""
|
|
_require_inputs()
|
|
empty = tmp_path / "cache"
|
|
empty.mkdir()
|
|
|
|
def factory(recording: Path, cancelled, on_progress):
|
|
# Cancel on the first frame, which is what the operator's button does.
|
|
def cancel_after_one(progress):
|
|
if progress.frames_written >= 1:
|
|
cancelled.set()
|
|
if on_progress is not None:
|
|
on_progress(progress)
|
|
|
|
return _service(empty, cancelled=cancelled, on_progress=cancel_after_one)
|
|
|
|
host = _Host(service_factory=factory)
|
|
controller, host, engine, parts = _controller(host=host)
|
|
controller.load_recording(RECORDING)
|
|
assert controller.wait_for_analysis(120)
|
|
_flush()
|
|
assert parts[4].states[-1] == "cancelled", parts[4].states
|
|
assert not controller.analysis_active(), "an incomplete result was served"
|
|
assert engine.mode() == MODE_RECORDING
|
|
assert controller.analysis_error() == "cancelled"
|
|
assert not any(
|
|
t("playback.analysis_cancelled") in note for note in host.notes
|
|
), "a terminal analysis state was written to the transient status line"
|
|
# The partial result is on disk and refuses to be read as complete.
|
|
directories = [entry.directory for entry in scan_results(empty)]
|
|
assert directories, "the cancelled pass wrote nothing at all"
|
|
assert all(
|
|
not entry.complete for entry in scan_results(empty)
|
|
), "a cancelled result claims to be complete"
|
|
with pytest.raises(AnalysisIncompleteError):
|
|
AnalysisReader(directories[0])
|
|
assert AnalysisReader(directories[0], allow_incomplete=True).complete is False
|
|
|
|
|
|
def test_an_incomplete_result_on_disk_is_not_adopted(tmp_path):
|
|
"""The refusal is the reader's, and the session never dresses it as playable."""
|
|
_require_inputs()
|
|
empty = tmp_path / "cache"
|
|
empty.mkdir()
|
|
cancelled = threading.Event()
|
|
|
|
def cancel_after_one(progress):
|
|
if progress.frames_written >= 1:
|
|
cancelled.set()
|
|
|
|
service = _service(empty, cancelled=cancelled, on_progress=cancel_after_one)
|
|
outcome = service.run()
|
|
assert outcome.complete is False
|
|
assert outcome.frames_written >= 1, "the pass wrote nothing, so nothing was refused"
|
|
with pytest.raises(AnalysisIncompleteError):
|
|
AnalysisReader(outcome.directory)
|
|
reader = AnalysisReader(outcome.directory, allow_incomplete=True)
|
|
try:
|
|
with pytest.raises(ResultPolicyError):
|
|
AnalysisSession(reader)
|
|
finally:
|
|
reader.close()
|
|
# And the service's own "is one stored?" answer is not a playable one: the directory
|
|
# exists, so it answers by refusing — which is what makes the controller run a pass
|
|
# instead of serving a partial result.
|
|
with pytest.raises(AnalysisIncompleteError):
|
|
service.stored_result()
|
|
|
|
|
|
def test_the_library_never_hands_out_an_incomplete_session(tmp_path):
|
|
"""``ResultLibrary.open`` refuses a partial result even when asked to inspect one."""
|
|
_require_inputs()
|
|
empty = tmp_path / "cache"
|
|
empty.mkdir()
|
|
cancelled = threading.Event()
|
|
|
|
def cancel_after_one(progress):
|
|
if progress.frames_written >= 1:
|
|
cancelled.set()
|
|
|
|
outcome = _service(empty, cancelled=cancelled, on_progress=cancel_after_one).run()
|
|
library = ResultLibrary()
|
|
with pytest.raises(ResultPolicyError):
|
|
library.open(outcome.directory, allow_incomplete=True)
|
|
assert library.session is None
|
|
|
|
|
|
def test_the_worker_reports_progress_and_a_cancelled_outcome():
|
|
"""The thread boundary: a pass reports progress, and a cancel is a cancelled report."""
|
|
seen: list = []
|
|
|
|
def factory(cancelled, on_progress):
|
|
def cancel_after(progress):
|
|
on_progress(progress)
|
|
if progress.frames_written >= 2:
|
|
cancelled.set()
|
|
|
|
from orisys_app_shared.analysis.service import Progress
|
|
|
|
class _Fake:
|
|
def run(self):
|
|
for index in range(5):
|
|
cancel_after(
|
|
Progress(
|
|
frames_written=index + 1,
|
|
frames_expected=5,
|
|
segment_index=1,
|
|
segment_count=1,
|
|
frame_id=FIRST_FRAME_ID + index,
|
|
elapsed_s=0.1 * index,
|
|
)
|
|
)
|
|
from orisys_app_shared.analysis.service import AnalysisOutcome
|
|
from orisys_app_shared.analysis.model import ResultKey
|
|
|
|
return AnalysisOutcome(
|
|
key=ResultKey(
|
|
recording="r", algorithm="a", sensor_config="s", calibration="", reference_frame=None
|
|
),
|
|
directory=Path("."),
|
|
complete=False,
|
|
reason="cancelled",
|
|
frames_written=2,
|
|
frames_expected=5,
|
|
elapsed_s=0.3,
|
|
)
|
|
|
|
return _Fake()
|
|
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
worker = AnalysisWorker(factory)
|
|
worker.progressed.connect(seen.append)
|
|
worker.finished.connect(lambda report: seen.append(("report", report)))
|
|
assert worker.start() is True
|
|
assert worker.wait(30)
|
|
QApplication.processEvents()
|
|
reports = [item for item in seen if isinstance(item, tuple)]
|
|
assert reports, "the worker never reported"
|
|
report = reports[-1][1]
|
|
assert report.cancelled is True and report.complete is False
|
|
assert report.frames_written == 2
|
|
assert report_from(None, cancelled=True).cancelled is True
|
|
|
|
|
|
# ------------------------------------------------------------------ the depth view
|
|
def _analysis_frame_for(frame, *, source=SOURCE_ANALYSIS):
|
|
"""The frame the controller hands the window's single 2D sink for a stored frame."""
|
|
return ImageFrame(
|
|
frame.base_bgr,
|
|
flow=frame.flow,
|
|
depth=frame.depth,
|
|
valid_mask=frame.valid_mask,
|
|
source=source,
|
|
frame_id=int(frame.frame_id),
|
|
t_rel_s=float(frame.t_rel_s),
|
|
flow_valid=True,
|
|
flow_frame_id=int(frame.frame_id),
|
|
depth_valid=bool(frame.has_depth),
|
|
depth_frame_id=int(frame.frame_id) if frame.has_depth else None,
|
|
result_available=True,
|
|
)
|
|
|
|
|
|
def _drawn(frame, mode: str):
|
|
image = _analysis_frame_for(frame)
|
|
return render_image(
|
|
image, plan_image(image), mode=mode, theme="dark", draw_style="box"
|
|
)
|
|
|
|
|
|
def test_the_depth_view_is_drawn_from_the_cache_by_the_backend_live_uses(session):
|
|
"""The depth mode renders the stored map through the same call live renders through.
|
|
|
|
Nothing here is a second draw path: ``render_image`` in ``depth`` mode is the same
|
|
function a live frame reaches, over the map the reader handed back. The check that it
|
|
is really that map is a direct comparison with the shared converter, and the check
|
|
that it is not some other view is that the arrows picture differs from it.
|
|
"""
|
|
frame = session.frame(FIRST_FRAME_ID + 1)
|
|
assert frame.has_depth, "the pass recorded no depth channel for this frame"
|
|
assert frame.depth.shape == (300, 280)
|
|
assert frame.depth.dtype == np.float32
|
|
|
|
drawn = _drawn(frame, DEPTH)
|
|
assert drawn is not None
|
|
expected = depth_map_to_bgr(
|
|
frame.depth, flow_colormap("dark"), value_range=None, invalid_mask=frame.valid_mask
|
|
)
|
|
assert np.array_equal(drawn, expected), "the depth mode did not draw the stored map"
|
|
arrows = _drawn(frame, ARROW_PURE)
|
|
assert arrows is not None
|
|
assert not np.array_equal(drawn, arrows), "the depth view drew the arrow view"
|
|
|
|
|
|
def test_a_seek_equals_playing_forward_in_the_depth_view(session):
|
|
"""The seek test, in the depth mode: same frame, same map, same bytes on screen.
|
|
|
|
Playing forward records each frame's rendered depth picture; a fresh session is then
|
|
seeked to five stored timestamps and must produce those pictures byte for byte. The
|
|
tolerance is zero on the frame id, on ``depth``, on the statistics that travel with
|
|
it and on the rendered image — because a seek is a file read of the same record.
|
|
"""
|
|
ids = [int(value) for value in session.frame_ids]
|
|
played: dict[int, tuple] = {}
|
|
for frame_id in ids:
|
|
frame = session.frame(frame_id)
|
|
played[frame_id] = (
|
|
frame.depth,
|
|
frame.depth_nonzero,
|
|
frame.depth_min,
|
|
frame.depth_max,
|
|
_drawn(frame, DEPTH),
|
|
)
|
|
fresh = AnalysisSession(
|
|
AnalysisReader(_result_directory(Path(session.directory).parent)),
|
|
policy=session.policy,
|
|
display_policy=session.display_policy,
|
|
)
|
|
try:
|
|
nonempty = 0
|
|
for frame_id in (ids[0], ids[1], ids[len(ids) // 2], ids[-2], ids[-1]):
|
|
moment = fresh.time_of(frame_id)
|
|
assert moment is not None
|
|
seeked = fresh.frame_at(moment)
|
|
assert seeked is not None and seeked.frame_id == frame_id
|
|
depth, nonzero, low, high, picture = played[frame_id]
|
|
nonempty += 1 if nonzero else 0
|
|
assert seeked.depth_nonzero == nonzero
|
|
assert (seeked.depth_min, seeked.depth_max) == (low, high)
|
|
assert np.array_equal(np.asarray(seeked.depth), np.asarray(depth))
|
|
picture_now = _drawn(seeked, DEPTH)
|
|
if picture is None or picture_now is None:
|
|
assert picture is None and picture_now is None
|
|
else:
|
|
assert np.array_equal(picture_now, picture)
|
|
assert nonempty <= 5
|
|
finally:
|
|
fresh.reader.close()
|
|
|
|
|
|
def test_a_mode_change_writes_no_analysis(pass_root, session):
|
|
"""Switching the 2D mode is a read: the result directory set cannot move.
|
|
|
|
The depth map is in the result, so drawing it is a file read like the arrows are.
|
|
The assertion is the set of result directories before and after every mode the
|
|
selector offers is selected, drawn and switched away from.
|
|
"""
|
|
before = {entry.directory.name for entry in scan_results(pass_root)}
|
|
for mode in (DEPTH, ARROW_PURE, ARROW_STITCHED, "recorded"):
|
|
frame = session.frame(FIRST_FRAME_ID + 1)
|
|
if mode != "recorded":
|
|
assert _drawn(frame, mode) is not None
|
|
assert {entry.directory.name for entry in scan_results(pass_root)} == before
|
|
|
|
|
|
def test_the_depth_statistics_say_which_frames_are_empty(session):
|
|
"""An empty depth view is explainable: the frame says so, and so does the result."""
|
|
frames = list(session.frames())
|
|
emptied = [item for item in frames if item.depth_is_empty]
|
|
carrying = [item for item in frames if item.depth_nonzero]
|
|
assert len(emptied) + len(carrying) == len(frames)
|
|
assert all(item.has_depth for item in frames), "a frame lost its depth channel"
|
|
assert all(item.depth_pixels == 300 * 280 for item in frames)
|
|
# Every observed frame on this footage falls in one of the two groups, and the empty
|
|
# ones are exactly zeros rather than NaNs or a shape that failed to decode.
|
|
for item in emptied:
|
|
assert not bool(np.any(np.asarray(item.depth)))
|
|
assert (item.depth_min, item.depth_max) == (0.0, 0.0)
|
|
for item in carrying:
|
|
assert float(np.asarray(item.depth).max()) == item.depth_max > 0.0
|
|
|
|
|
|
# ------------------------------------------------------- the operator's one line
|
|
def _note_for(image: ImageFrame) -> str:
|
|
plan = plan_image(image)
|
|
return "" if not plan.note_key else t(plan.note_key, fields="flow、depth")
|
|
|
|
|
|
def test_the_nothing_to_draw_line_appears_only_when_no_result_exists():
|
|
"""The recording's note is an absence statement, and it must be true when it is made.
|
|
|
|
Two recording frames that differ in exactly one fact: whether a stored analysis
|
|
result exists. Without one, the pane really can draw nothing derived, so the line
|
|
appears and offers the recorded view alone. With one, the arrows and the depth map
|
|
are two file reads away, so the line is gone — and it is gone because the modes are
|
|
on offer, not because a message was suppressed.
|
|
"""
|
|
raw = ImageFrame(np.zeros((8, 8, 3), np.uint8), source=SOURCE_REPLAY)
|
|
plan = plan_image(raw)
|
|
assert plan.available_modes == ("recorded",)
|
|
assert plan.note_key == RAW_NOTE_KEY
|
|
assert plan.resolve(DEPTH) == "recorded", "a view that cannot be drawn was selected"
|
|
assert plan.has_depth() is False and plan.has_flow() is False
|
|
|
|
backed = ImageFrame(
|
|
np.zeros((8, 8, 3), np.uint8), source=SOURCE_REPLAY, result_available=True
|
|
)
|
|
plan = plan_image(backed)
|
|
assert plan.note_key == "", "the pane still says there is nothing to draw"
|
|
assert plan.available_modes == ("recorded", *DERIVED_MODES)
|
|
assert plan.default_mode == "recorded", "the raw picture stopped being the default"
|
|
for mode in DERIVED_MODES:
|
|
assert plan.accepts(mode), f"{mode} is not on offer with a result in hand"
|
|
|
|
|
|
def test_the_nothing_to_draw_line_names_the_action_that_produces_a_result():
|
|
"""Both locales: the line says what to do, not merely that there is nothing."""
|
|
from orisys_app_shared.i18n import set_locale
|
|
|
|
seen = {}
|
|
try:
|
|
for locale in ("zh_CN", "en_US"):
|
|
set_locale(locale)
|
|
text = t(RAW_NOTE_KEY)
|
|
seen[locale] = text
|
|
# The action that produces a result is named: the pass runs on open, and the
|
|
# toolbar action re-runs it.
|
|
assert "由视频重算" in text or "Recompute from the video" in text, text
|
|
assert (
|
|
"分析" in text or "analysis result" in text
|
|
), f"{locale} does not say a result is what is missing: {text}"
|
|
finally:
|
|
set_locale("zh_CN")
|
|
assert seen["zh_CN"] != seen["en_US"]
|
|
|
|
|
|
def test_the_empty_depth_note_is_not_the_missing_depth_note():
|
|
"""Both locales keep the two apart: a flat map is a reading, an absent one is not."""
|
|
from orisys_app_shared.i18n import set_locale
|
|
|
|
try:
|
|
for locale in ("zh_CN", "en_US"):
|
|
set_locale(locale)
|
|
empty = t("panels.depth_empty", pixels=84000)
|
|
absent = t("panels.depth_unavailable")
|
|
assert "84000" in empty or "84,000" in empty, empty
|
|
assert empty != absent
|
|
finally:
|
|
set_locale("zh_CN")
|
|
|
|
|
|
def test_a_derived_mode_is_served_from_the_result_even_on_the_recording_timeline(
|
|
pass_root,
|
|
):
|
|
"""Through the controller: the depth mode draws the stored frame, not the video.
|
|
|
|
The timeline is put on the recording (which is the state the operator is in while a
|
|
pass runs, and the state a chosen source leaves them in), and the mode selector is
|
|
set to depth. What reaches the window's single 2D sink must be the result's own frame
|
|
— with its depth map — and not the decoded video frame, which carries none.
|
|
"""
|
|
_require_inputs()
|
|
reader = AnalysisReader(_result_directory(pass_root))
|
|
session = AnalysisSession(
|
|
reader,
|
|
display_policy=policy_for_settings(DisplaySettings()),
|
|
# The recording the result describes. ``available_result`` matches it against the
|
|
# recording that is open, which is how a previous recording's result cannot
|
|
# answer for this one.
|
|
recording=RECORDING,
|
|
)
|
|
try:
|
|
panel = _ImagePanel()
|
|
controller, host, engine, _ = _controller(image_panel=panel)
|
|
controller.results.open_session(session)
|
|
engine.use_analysis(session)
|
|
engine._mode = MODE_RECORDING # the timeline reads the video; the view does not
|
|
engine.set_position(session.time_of(FIRST_FRAME_ID + 1))
|
|
host.frames.clear()
|
|
|
|
panel.set_view_mode(DEPTH)
|
|
controller.render_frame()
|
|
assert host.frames, "the depth mode drew nothing"
|
|
drawn = host.frames[-1]
|
|
assert drawn.source == SOURCE_REPLAY, "the recording timeline was not kept"
|
|
assert drawn.result_available is True
|
|
stored = session.frame(FIRST_FRAME_ID + 1)
|
|
assert drawn.frame_id == stored.frame_id
|
|
assert np.array_equal(np.asarray(drawn.depth), np.asarray(stored.depth))
|
|
assert drawn.depth_valid is True
|
|
assert plan_image(drawn).note_key == "", "the pane despaired with a result in hand"
|
|
|
|
# And the raw view still shows the recording's own picture.
|
|
panel.set_view_mode("recorded")
|
|
host.frames.clear()
|
|
controller.render_frame()
|
|
assert host.frames[-1].depth is None
|
|
assert host.frames[-1].source == SOURCE_REPLAY
|
|
|
|
# Now through the engine's own way back to the video: it forgets the session it
|
|
# was borrowing, and the *library* still holds the result for this recording. The
|
|
# derived mode must keep drawing from it — a result is a property of the
|
|
# recording, not of which timeline the cursor is on.
|
|
engine.use_recording()
|
|
assert engine.analysis() is None
|
|
panel.set_view_mode(DEPTH)
|
|
host.frames.clear()
|
|
controller.render_frame()
|
|
assert host.frames[-1].depth is not None, "the depth view followed the borrow"
|
|
assert np.array_equal(
|
|
np.asarray(host.frames[-1].depth), np.asarray(stored.depth)
|
|
)
|
|
assert plan_image(host.frames[-1]).note_key == ""
|
|
finally:
|
|
reader.close()
|