"""Replay of a recording: the reader, and the depth replay policy. Depth is not stored with a recording (team decision): replay recomputes it from ``raw_*.mp4`` and the recorded config. These tests cover the reader contract and that the policy's risks are reported instead of discovered later. """ from __future__ import annotations import csv import json import shutil import sys import types from pathlib import Path import numpy as np import pytest from orisys_app_shared.acquisition.sdk_worker import SdkPacket from orisys_app_shared.record import ( ExperimentRecorder, RecordConfig, RecordingReader, list_recordings, ) SCRATCH = Path(__file__).resolve().parent / "_scratch_playback" def _packet(index: int, *, valid: bool = True) -> SdkPacket: nan = float("nan") frame = np.full((24, 32, 3), (index * 5) % 255, np.uint8) return SdkPacket( raw_frame=frame, processed_frame=frame, timestamp=1000.0 + index / 30.0, fps=30.0, process_ms=1.0, normal_force=2.0 if valid else nan, shear_x=0.5 if valid else nan, shear_y=-0.5 if valid else nan, normal_force_raw=(9.0 + index * 0.1) if valid else nan, shear_x_raw=0.4 if valid else nan, shear_y_raw=-0.4 if valid else nan, force_unit="N", force_calibrated=True, session_id=1, frame_id=index, valid=valid, invalid_reason="" if valid else "read failed", raw_available=True, ) @pytest.fixture def recording(): shutil.rmtree(SCRATCH, ignore_errors=True) SCRATCH.mkdir(parents=True, exist_ok=True) recorder = ExperimentRecorder(RecordConfig(video=True, fps=30.0, segment_seconds=0.4)) directory = recorder.start( root=SCRATCH, session_id=3, calibration_id="/cfg/sensor.json", metadata={ "device": { "sensor_type": "box", "config_ref": str(SCRATCH / "sensor.json"), "config_sha256": "recorded-hash", }, "versions": {"sdk": "0.5.1"}, "depth": {"stored": False, "replay": "recompute"}, }, ) (SCRATCH / "sensor.json").write_text('{"camera": {}}\n', encoding="utf-8") for index in range(60): recorder.on_packet(_packet(index)) recorder.on_packet(_packet(90, valid=False)) recorder.stop() yield directory shutil.rmtree(SCRATCH, ignore_errors=True) def test_reader_reports_the_recording_shape(recording): reader = RecordingReader(recording) info = reader.info() assert len(info.frames) == 61 assert info.times.size == 61 assert info.duration_s == pytest.approx(3.0, abs=0.05) assert int(info.valid.sum()) == 60, "the failed readout should be marked invalid" def test_reader_keeps_unavailable_values_empty(recording): reader = RecordingReader(recording) info = reader.info() assert np.isnan(info.raw[-1]).all(), "an empty CSV cell must not read back as 0" assert np.isfinite(info.raw[0]).all() def test_reader_maps_time_to_the_video_segment(recording): reader = RecordingReader(recording) frame = reader.frame_at(1.0) assert frame is not None assert frame.frame_id == 30 assert frame.t_rel_s == pytest.approx(1.0, abs=0.02) assert frame.in_video assert frame.segment >= 1 assert reader.segment_path(frame.segment) is not None # Before the start and past the end clamp instead of raising. assert reader.frame_at(-5.0).frame_id == 0 assert reader.frame_at(500.0).frame_id == 90 def test_reader_decodes_the_video_frame(recording): import cv2 if not hasattr(cv2, "VideoCapture"): # pragma: no cover pytest.skip("no video backend") reader = RecordingReader(recording) image, frame = reader.video_frame_at(1.0) assert frame is not None and frame.in_video assert image is not None, "the frame index and the video disagreed" assert image.shape[:2] == (24, 32) def test_reader_window_and_streaming_agree(recording): reader = RecordingReader(recording) times, values = reader.forces_window(0.5, 1.0) assert len(times) == 16 assert values.shape == (16, 3) streamed = sum(len(batch_times) for batch_times, _ in reader.iter_forces(batch=25)) assert streamed == 61, "streaming must see every row" def test_an_empty_recording_directory_is_readable(): """A folder with no CSVs must open (empty), not raise an unpack error.""" shutil.rmtree(SCRATCH, ignore_errors=True) SCRATCH.mkdir(parents=True, exist_ok=True) try: reader = RecordingReader(SCRATCH) info = reader.info() assert info.frames == [] assert info.times.size == 0 assert info.duration_s == 0.0 assert reader.frame_at(1.0) is None assert reader.replay_warnings(sdk_version="0.5.1") finally: shutil.rmtree(SCRATCH, ignore_errors=True) def test_a_new_recording_is_reported_as_time_ordered(recording): info = RecordingReader(recording).info() assert info.frames_out_of_order, "the writer no longer produces ordered rows" assert info.forces_out_of_order codes = RecordingReader(recording).replay_warnings(sdk_version="0.5.1") assert "out_of_order" not in codes def test_an_old_out_of_order_recording_is_re_sorted_and_reported(): """Rows written by the pre-fix writer are not in time order. ``frame_at`` binary-searches these times, so an unsorted file would answer with the wrong frame; the reader sorts and says so instead. """ shutil.rmtree(SCRATCH, ignore_errors=True) SCRATCH.mkdir(parents=True, exist_ok=True) (SCRATCH / "forces.csv").write_text( "session_id,frame_id,t_rel_s,FN_raw,FX_raw,FY_raw,FN_N,FX_N,FY_N,valid,calibration_id\n" "1,0,0.0,1.0,0.1,0.1,2.0,0.1,0.1,1,h\n" "1,2,0.066,3.0,0.1,0.1,6.0,0.1,0.1,1,h\n" # written before frame 1 "1,1,0.033,2.0,0.1,0.1,4.0,0.1,0.1,1,h\n", encoding="utf-8", ) (SCRATCH / "frames.csv").write_text( "frame_id,t_rel_s,acquired_at,segment,segment_frame,in_video,dropped\n" "0,0.0,2026-01-01T00:00:00.000Z,1,1,1,0\n" "2,0.066,2026-01-01T00:00:00.066Z,1,3,1,0\n" "1,0.033,2026-01-01T00:00:00.033Z,1,2,1,0\n", encoding="utf-8", ) (SCRATCH / "session.json").write_text( json.dumps( { "format_version": "1.1", "device": {"sensor_type": "box", "config_ref": "/cfg/sensor.json"}, "versions": {"sdk": "0.5.1"}, "depth": {"stored": False}, } ), encoding="utf-8", ) try: reader = RecordingReader(SCRATCH) info = reader.info() assert not info.frames_out_of_order assert not info.forces_out_of_order assert [frame.frame_id for frame in info.frames] == [0, 1, 2] assert info.times.tolist() == pytest.approx([0.0, 0.033, 0.066]) # The binary search is only valid because the rows were sorted. assert reader.frame_at(0.05).frame_id == 1 assert reader.frame_at(0.07).frame_id == 2 assert "out_of_order" in reader.replay_warnings(sdk_version="0.5.1") finally: shutil.rmtree(SCRATCH, ignore_errors=True) def test_depth_policy_says_recompute_and_warns_about_drift(recording): reader = RecordingReader(recording) assert reader.depth_is_stored() is False assert reader.depth_policy().get("replay") == "recompute" assert reader.recorded_config_ref() == str(SCRATCH / "sensor.json") # The config file on disk no longer matches what was recorded. codes = reader.replay_warnings(sdk_version="0.5.1") assert "config_changed" in codes # Same SDK: no version warning. assert "sdk_changed" not in codes assert "sdk_changed" in reader.replay_warnings(sdk_version="9.9.9") def test_list_recordings_finds_directories_newest_first(recording): second = SCRATCH / "zzz_later" second.mkdir() (second / "forces.csv").write_text("", encoding="utf-8") found = list_recordings(SCRATCH) assert found[0].name == "zzz_later", "recordings should be newest first" assert recording in found if __name__ == "__main__": # pragma: no cover - manual runner import shutil as _shutil failures = 0 for name, function in sorted(globals().items()): if not name.startswith("test_") or not callable(function): continue try: if name == "test_list_recordings_finds_directories_newest_first": _shutil.rmtree(SCRATCH, ignore_errors=True) SCRATCH.mkdir(parents=True, exist_ok=True) recorder = ExperimentRecorder(RecordConfig(video=False)) directory = recorder.start( root=SCRATCH, session_id=1, metadata={"depth": {"stored": False}} ) recorder.on_packet(_packet(0)) recorder.stop() try: function(directory) finally: _shutil.rmtree(SCRATCH, ignore_errors=True) print(f" ok {name}") continue _shutil.rmtree(SCRATCH, ignore_errors=True) SCRATCH.mkdir(parents=True, exist_ok=True) recorder = ExperimentRecorder(RecordConfig(video=True, fps=30.0, segment_seconds=0.4)) directory = recorder.start( root=SCRATCH, session_id=3, calibration_id="/cfg/sensor.json", metadata={ "device": { "sensor_type": "box", "config_ref": str(SCRATCH / "sensor.json"), "config_sha256": "recorded-hash", }, "versions": {"sdk": "0.5.1"}, "depth": {"stored": False, "replay": "recompute"}, }, ) (SCRATCH / "sensor.json").write_text('{"camera": {}}\n', encoding="utf-8") for index in range(60): recorder.on_packet(_packet(index)) recorder.on_packet(_packet(90, valid=False)) recorder.stop() try: function(directory) finally: _shutil.rmtree(SCRATCH, ignore_errors=True) print(f" ok {name}") except Exception as exc: # noqa: BLE001 failures += 1 print(f" FAIL {name}: {type(exc).__name__}: {exc}") print(f"failures: {failures}") raise SystemExit(1 if failures else 0)