"""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, RESULT_FAILED, RESULT_IMPOSSIBLE, RESULT_NONE, RESULT_RUNNING, SOURCE_ANALYSIS, SOURCE_LIVE, SOURCE_REPLAY, ImageFrame, plan_image, render_image, ) from orisys_app_shared.analysis import service as analysis_service_mod 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, chain_spec_for, ) from orisys_app_shared.i18n import t from orisys_app_shared.record.playback import RecordingReader 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, ) @pytest.fixture(scope="module") def box_pass(tmp_path_factory): """One box-chain pass over the evidence recording, for live/replay paint checks.""" _require_inputs() root = tmp_path_factory.mktemp("replay-analysis-box") chain = chain_spec_for(RecordingReader(RECORDING).session()) service = AnalysisService(RECORDING, chain=chain, cache=ResultCache(root)) outcome = service.run() assert outcome.complete, outcome.reason return service, outcome @pytest.fixture(scope="module") def box_session(box_pass): service, outcome = box_pass reader = AnalysisReader(outcome.directory) session = AnalysisSession( reader, policy=service.policy, display_policy=service.policy, session_id=2, recording=RECORDING, ) yield session reader.close() 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") def _wire_replay(controller, host): controller.view_requested.connect(lambda preset: ( host.presets.append(preset), setattr(host, "view", preset), controller.set_active_view(preset), )) controller.acquire_requested.connect(host.start_acquire) controller.acquire_stop_requested.connect( lambda reason: host.stop_acquire(reason=reason) ) controller.set_active_view(host.view) return controller class _Host: def __init__(self, service_factory=None) -> None: self.prefs: dict = {} self.view = "replay" self._running_flag = False self._preview_open_flag = False self._session_active_flag = 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 @property def running(self) -> bool: return self._running_flag @running.setter def running(self, value: bool) -> None: self._running_flag = bool(value) @property def preview_open(self) -> bool: return self._preview_open_flag @preview_open.setter def preview_open(self, value: bool) -> None: self._preview_open_flag = bool(value) @property def session_active(self) -> bool: return self._session_active_flag @session_active.setter def session_active(self, value: bool) -> None: self._session_active_flag = bool(value) @property def _running(self) -> bool: return self._running_flag @_running.setter def _running(self, value: bool) -> None: self._running_flag = bool(value) @property def _preview_open(self) -> bool: return self._preview_open_flag @_preview_open.setter def _preview_open(self, value: bool) -> None: self._preview_open_flag = bool(value) @property def _session_active(self) -> bool: return self._session_active_flag @_session_active.setter def _session_active(self, value: bool) -> None: self._session_active_flag = bool(value) def current_source_kind(self) -> str: return "replay" def set_display_unit(self, unit: str) -> None: self.units.append(unit) def set_source_mode(self, mode: str) -> None: pass def show_frame(self, frame) -> None: self.frames.append(frame) def show_analysis_packet(self, packet) -> None: self.deliver_measurement(packet, advance="flush", paint_2d=False) def deliver_measurement( self, packet, *, advance: str = "tick", arrow_policy=None, image_source: str | None = None, paint_2d: bool = True, ) -> None: if paint_2d: from orisys_app_shared.acquisition.image_backend import ImageFrame self.frames.append( ImageFrame( getattr(packet, "processed_frame", None), flow=getattr(packet, "flow", None), depth=getattr(packet, "depth", None), valid_mask=getattr(packet, "valid_mask", None), source=image_source or "replay", frame_id=int(getattr(packet, "frame_id", 0) or 0), arrow_policy=arrow_policy, result_available=True, ) ) 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_acquire(self) -> None: pass def stop_acquire(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 persist_prefs(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) def bare_video_workspace(self, video, **kwargs): return None def recording_analysis_overrides(self, recording): return False 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 = _wire_replay( ReplayController( engine=engine, host=host, toolbar=toolbar, charts=charts, image_panel=panel, mesh_panel=mesh, analysis_progress=progress, ), host, ) 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): # ``None`` for a bare video, exactly as ``ReplayEngine.directory`` answers: the # controller's "is there anything here a pass could read?" is this call. return None if self._reader is None else 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_box_live_and_replay_arrow_canvases_do_not_diverge(box_pass, box_session, monkeypatch): """Live dense field equals the replay canvas, frame for frame. The operator's live view paints the chain's dense flow, and the replay paints a field recovered from the stored samples. Both go through :func:`render_image` with ``draw_style="box"`` and the shared painter. """ service, outcome = box_pass assert outcome.complete, outcome.reason original = analysis_service_mod.select_flow_samples dense_flows: list[np.ndarray] = [] def _capture(flow, mask, policy, **kwargs): dense_flows.append(np.asarray(flow, dtype=np.float64).copy()) return original(flow, mask, policy, **kwargs) monkeypatch.setattr(analysis_service_mod, "select_flow_samples", _capture) lives = list(service.recompute()) assert len(lives) == len(dense_flows) == outcome.frames_written canvas_bgr = (0x16, 0x13, 0x12) policy = box_session.paint_policy() assert int(policy.flow_grid_spacing) == 20 painted_any = 0 for live, dense in zip(lives, dense_flows, strict=True): frame_id = int(live.row.frame_id) analysed = box_session.frame(frame_id) height, width = int(dense.shape[0]), int(dense.shape[1]) base = np.empty((height, width, 3), dtype=np.uint8) base[:] = canvas_bgr live_canvas = render_image( ImageFrame( base, flow=dense, source=SOURCE_LIVE, flow_valid=True, valid_mask=live.mask, ), mode=ARROW_PURE, draw_style="box", arrow_policy=policy, canvas_bgr=canvas_bgr, ) replay_canvas = render_image( ImageFrame( base, flow=analysed.flow, source=SOURCE_ANALYSIS, flow_valid=True, result_available=True, valid_mask=analysed.valid_mask, ), mode=ARROW_PURE, draw_style="box", arrow_policy=policy, canvas_bgr=canvas_bgr, ) assert live_canvas is not None and replay_canvas is not None assert np.array_equal(np.asarray(live_canvas), np.asarray(replay_canvas)), ( f"frame {frame_id}: live and replay arrow canvases diverge" ) painted_any = max( painted_any, int(np.count_nonzero(np.any(np.asarray(live_canvas) != canvas_bgr, axis=2))), ) assert painted_any > 0, "every frame painted nothing; the discrepancy was vacuous" 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, depth 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 bool(packet.depth_valid) is bool(frame.has_depth) if frame.has_depth: assert np.array_equal(np.asarray(packet.depth), np.asarray(frame.depth)) assert packet.depth_frame_id == int(frame.frame_id) else: assert packet.depth is None and packet.depth_frame_id is None 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) def test_paint_policy_matches_the_2d_painter_threshold(session, box_session): """3D overlays use ``paint_policy``: finger keeps the display floor, box keeps 2 px.""" from orisys_app_shared.acquisition.arrows import sdk_box_policy from orisys_app_shared.scene3d.visual_policy import VisualPolicy finger = session.paint_policy() assert finger.flow_on_px == pytest.approx(session.display_policy.flow_on_px) assert int(finger.flow_grid_spacing) == int( session.stored_policy().get("flow_grid_spacing", session.policy.flow_grid_spacing) ) # Keep the stored grid; only the activity threshold moves. adopted = session.display_policy assert session.serve_policy( VisualPolicy( flow_on_px=0.9, flow_off_px=0.6, flow_grid_spacing=adopted.flow_grid_spacing, flow_arrow_scale=adopted.flow_arrow_scale, flow_edge_margin=adopted.flow_edge_margin, flow_length_margin=adopted.flow_length_margin, max_arrows=adopted.max_arrows, ) ) assert session.paint_policy().flow_on_px == pytest.approx(0.9) painter = sdk_box_policy() box = box_session.paint_policy() assert box.flow_on_px == pytest.approx(painter.flow_on_px) assert box.flow_off_px == pytest.approx(painter.flow_off_px) assert int(box.flow_grid_spacing) == int(painter.flow_grid_spacing) # --------------------------------------------------------------- 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 assert host.notes == [], "a threshold change must not lecture about the grid" finally: session.reader.close() def test_a_refused_grid_change_redraws_without_a_status_warning(): """Moving max_arrows cannot be served; the frame is redrawn and nothing is said. No evidence recording: the session is a stub whose ``serve_policy`` refuses, which is exactly the path that used to push ``playback.analysis_grid_mismatch``. """ class _Reader: def close(self) -> None: pass class _Session: def __init__(self) -> None: self.calls: list = [] self.reader = _Reader() def serve_policy(self, policy) -> bool: self.calls.append(policy) return False def close(self) -> None: pass controller, host, engine, parts = _controller() session = _Session() controller.results.open_session(session) # type: ignore[arg-type] engine.use_analysis(session) engine.redraws = 0 controller.on_display_settings(DisplaySettings(max_arrows=50)) assert engine.redraws == 1, "the refused policy still redraws the stored frame" assert len(session.calls) == 1 assert host.notes == [], "a grid mismatch must not reach the status bar" controller.results.close() # ------------------------------------------------------------ analysis on open def test_a_running_session_does_not_hide_a_stored_result(pass_root): """A result on disk is served while a session is open: reading it opens no chain. The defect this pins: ``on_analysis_requested`` asked "is a session or a preview open?" *before* it asked "is a result already stored?". Serving a result is a file read — no SDK chain, no device, no working directory — so the one-chain-at-a-time rule never applied to it, but the guard refused the whole call anyway. The operator who recorded something and opened it while the acquisition session was still live (a paused session keeps the device, so it counts) was shown "this recording has no analysis result" over a recording whose arrows were sitting in the cache, and had to stop the session and reopen the recording to see them. """ _require_inputs() calls: list[str] = [] def factory(recording: Path, cancelled, on_progress): calls.append(str(recording)) return _service(pass_root) for live in ("running", "preview", "winding_down"): calls.clear() host = _Host(service_factory=factory) if live == "running": host._running = True elif live == "preview": host._preview_open = True else: host._session_active = True panel = _ImagePanel() controller, host, engine, parts = _controller(host=host, image_panel=panel) assert controller.load_recording(RECORDING) is True assert controller.available_result() is not None, ( f"a cached result was not served while a session was {live}" ) assert not controller.analysis_running(), "a pass ran over a stored result" assert calls == [str(RECORDING)], "the stored result was not even looked for" assert parts[4].states[-1] == "reused", "the surface did not report the reuse" assert controller.analysis_error() == "" # What reaches the 2D sink is the cached measurement, not "there is nothing". host.frames.clear() panel.set_view_mode(DEPTH) controller.render_frame() drawn = host.frames[-1] assert drawn.result_available is True assert drawn.depth is not None, "the depth view was not served from the cache" assert plan_image(drawn).note_key == "", "the panel still says it has no result" def test_a_pass_is_refused_while_a_session_is_open_and_says_so(tmp_path, qt_app): """With nothing cached, the refusal is *state*: the panel states it, with the reason. The pass really cannot run next to a session (one SDK chain and one working directory at a time), so the fix is not to run it anyway — it is to stop the refusal from being a silent no-op whose only trace is a status line that expires. Before this, the four states below were one line, and three of them were wrong. """ _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) host._running = True panel = _ImagePanel() controller, host, engine, parts = _controller(host=host, image_panel=panel) assert controller.load_recording(RECORDING) is True assert not controller.analysis_running(), "a pass was started next to a session" assert parts[4].states == [], "the progress surface claimed a pass" state, reason = controller.result_state() assert state == RESULT_FAILED, f"the refusal was reported as {state!r}" assert reason == t("analysis.blocked") and reason, "the refusal lost its reason" assert controller.analysis_error() == reason host.frames.clear() controller.render_frame() drawn = host.frames[-1] assert drawn.result_available is False plan = plan_image(drawn) assert plan.note_key == "panels.replay_analysis_failed" assert plan.note_reason == reason, "the line the panel draws lost the reason" assert plan.note_key != RAW_NOTE_KEY, ( "the panel promised a pass that was refused before it started" ) assert plan.available_modes == ("recorded",) def test_the_four_no_result_states_are_four_different_lines(tmp_path, qt_app): """No result yet, running, failed (with its reason), impossible — all four reachable. Each is a different fact and a different next action, so each is a different line: a pass is coming, a pass is going, a pass is not coming and here is why, or no pass can ever serve this frame. ``none`` is what the panel draws for a recording whose analysis has not been asked for yet — the redraw the engine's own ``recording_changed`` makes inside ``load_recording``, before ``on_analysis_requested`` runs. ``failed`` is reached by a pass that really fails (a chain whose configuration file is absent), because the reason on the line has to be the pass's own, not one a test invented. """ _require_inputs() empty = tmp_path / "cache" empty.mkdir() def working(recording: Path, cancelled, on_progress): return _service(empty, cancelled=cancelled, on_progress=on_progress) def broken(recording: Path, cancelled, on_progress): # A chain the SDK cannot open: the configuration file does not exist. This is the # shape of every "the pass ran and failed" the operator can meet — a moved # install, a config edited away — and the reason it produces is the pass's own. return AnalysisService( RECORDING, chain=ChainSpec(sensor_type="box", config_ref=tmp_path / "absent-sensor.json"), cache=ResultCache(empty), cancelled=cancelled, on_progress=on_progress, ) # none: the recording is open, and nothing has asked for a pass yet. host = _Host(service_factory=working) panel = _ImagePanel() controller, host, engine, parts = _controller(host=host, image_panel=panel) assert engine.open_recording(RECORDING) is True assert controller.result_state() == (RESULT_NONE, "") host.frames.clear() controller.render_frame() assert plan_image(host.frames[-1]).note_key == RAW_NOTE_KEY # running: the pass is in flight and the panel says so instead of promising one. controller.on_analysis_requested() assert controller.analysis_running(), "no pass was started" assert controller.result_state() == (RESULT_RUNNING, "") host.frames.clear() controller.render_frame() assert plan_image(host.frames[-1]).note_key == "panels.replay_analysis_running" assert controller.wait_for_analysis(120) _flush() # failed: a pass that ended without a result carries its own reason to the panel. failing = _Host(service_factory=broken) failing_panel = _ImagePanel() failed_ctl, failing, _, failed_parts = _controller( host=failing, image_panel=failing_panel ) assert failed_ctl.load_recording(RECORDING) is True assert failed_ctl.wait_for_analysis(120), "the failing pass never finished" _flush() state, reason = failed_ctl.result_state() assert reason, "a failed pass left no reason to show" assert failed_parts[4].states[-1] == "failed", failed_parts[4].states if state != RESULT_FAILED: # a box chain that somehow opened: then it has a result pytest.fail(f"the failed pass is reported as {state!r}") failing.frames.clear() failed_ctl.render_frame() plan = plan_image(failing.frames[-1]) assert plan.note_key == "panels.replay_analysis_failed" assert plan.note_reason == reason, "the line the panel draws lost the reason" # impossible: a bare video has no session metadata, so no pass can serve it at all. bare = tmp_path / "clip.mp4" bare.write_bytes(b"\0" * 8) engine._reader = None controller._toolbar.set_video_path(str(bare)) assert controller.recording_directory() is None assert controller.result_state() == (RESULT_IMPOSSIBLE, "") host.frames.clear() controller.render_frame() assert plan_image(host.frames[-1]).note_key == "panels.replay_analysis_unavailable" 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 "Analyze 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_depth_unavailable_is_a_short_absent_note(): """Both locales say the frame has no depth, without explaining zero vs absent.""" from orisys_app_shared.i18n import set_locale try: for locale in ("zh_CN", "en_US"): set_locale(locale) absent = t("panels.depth_unavailable") assert absent assert "\n" not in absent, absent scale = t("panels.depth_scale", low="0.000", high="1.000") assert scale != 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() def test_a_stitch_grid_is_not_painted_onto_the_camera_frame(): """Box results have no mask. The canvas is the stitch image, not 1080p. Candidates and the depth plane live on ``resolution_stitch``. The raw camera frame is the old stand-in, and it leaves the arrows in one corner. """ from types import SimpleNamespace class _Reader: complete = True key = "stitch" directory = Path(".") frame_ids = np.array([1], dtype=np.int64) times = np.array([0.0]) keyframes: dict = {} def mask(self): return None def frame(self, frame_id): return SimpleNamespace(depth=np.zeros((365, 365), dtype=np.float32)) def grid(self, grid_id): raise AssertionError("a depth plane already names the canvas") session = AnalysisSession(_Reader()) assert session.domain_hw() == (365, 365) rows = np.array([15, 355], dtype=np.int32) cols = np.array([15, 300], dtype=np.int32) class _GridReader(_Reader): keyframes = {0: SimpleNamespace(id=0, kind="grid")} def frame(self, frame_id): return SimpleNamespace(depth=None) def grid(self, grid_id): return rows, cols assert AnalysisSession(_GridReader()).domain_hw() == (356, 301)