From ee15160862b52b1164e5fb28843a95aa1be4be60 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Thu, 23 Jul 2026 11:42:19 +0300 Subject: [PATCH] fix(viewer): make recorded layers deterministic --- .../src/components/RerunViewport.tsx | 25 ++++++- .../src/workspaces/Workspaces.tsx | 24 +++++-- .../test/observationSources.test.mjs | 33 +++++++++ .../rerunViewportProgressivePlayback.test.mjs | 45 +++++++++++++ src/k1link/viewer/recorded.py | 67 ++++++++++++------- tests/test_rrd_export.py | 62 ++++++++++++++++- 6 files changed, 222 insertions(+), 34 deletions(-) diff --git a/apps/control-station/src/components/RerunViewport.tsx b/apps/control-station/src/components/RerunViewport.tsx index 583a03d..e6b104c 100644 --- a/apps/control-station/src/components/RerunViewport.tsx +++ b/apps/control-station/src/components/RerunViewport.tsx @@ -49,6 +49,7 @@ export interface RerunViewportProps { presentationGate?: RecordedAdmissionPhase; expectedTimelineStartSeconds?: number; expectedTimelineEndSeconds?: number; + initialPlaybackStartSeconds?: number; onStatusChange?: (status: RerunViewportStatus, message?: string) => void; onSelectionChange?: (selection: RerunSelection | null) => void; onPlaybackChange?: (state: RerunPlaybackState | null) => void; @@ -295,12 +296,21 @@ export function createRecordedAutoplayGate(): { rangeNs: { min: number; max: number } | null, seekToStart: (startNs: number) => void, startPlaying: () => void, + preferredStartNs?: number, ) => boolean; attempted: () => boolean; } { let consumed = false; return { - attempt(viewerStarted, fullyBuffered, presentationReady, rangeNs, seekToStart, startPlaying) { + attempt( + viewerStarted, + fullyBuffered, + presentationReady, + rangeNs, + seekToStart, + startPlaying, + preferredStartNs, + ) { if ( consumed || !viewerStarted || @@ -309,8 +319,11 @@ export function createRecordedAutoplayGate(): { !isUsableRecordedPlaybackRange(rangeNs) ) return false; consumed = true; + const startNs = Number.isFinite(preferredStartNs) + ? Math.min(Math.max(preferredStartNs as number, rangeNs.min), rangeNs.max) + : rangeNs.min; return attemptRecordedAutoplay( - () => seekToStart(rangeNs.min), + () => seekToStart(startNs), startPlaying, ); }, @@ -515,7 +528,8 @@ export async function fetchRecordedBlueprintRrd( custom_color: settings.customColor, active_view: activeView, view_reset_generation: viewResetGeneration, - unified_perception: perceptionLayers.enabled, + unified_perception: + perceptionLayers.detections2d || perceptionLayers.segmentation, show_detections_2d: perceptionLayers.detections2d, show_segmentation: perceptionLayers.segmentation, show_cuboids_3d: perceptionLayers.cuboids3d, @@ -616,6 +630,7 @@ export function RerunViewport({ presentationGate = "ready", expectedTimelineStartSeconds, expectedTimelineEndSeconds, + initialPlaybackStartSeconds, onStatusChange, onSelectionChange, onPlaybackChange, @@ -982,6 +997,9 @@ export function RerunViewport({ () => { viewer.set_playing(event.recording_id, true); }, + initialPlaybackStartSeconds === undefined + ? undefined + : initialPlaybackStartSeconds * 1_000_000_000, ); if (autoplayStarted) { playing = true; @@ -1153,6 +1171,7 @@ export function RerunViewport({ expectedTimelineEndSeconds, expectedTimelineStartSeconds, followLive, + initialPlaybackStartSeconds, onPlaybackChange, onPlaybackControllerChange, onSelectionChange, diff --git a/apps/control-station/src/workspaces/Workspaces.tsx b/apps/control-station/src/workspaces/Workspaces.tsx index 4645751..78454f3 100644 --- a/apps/control-station/src/workspaces/Workspaces.tsx +++ b/apps/control-station/src/workspaces/Workspaces.tsx @@ -336,10 +336,13 @@ function SpatialWorkspace({ (source) => source.capabilities.overlay && source.modality !== "point-cloud", ); const recordedPerceptionSupported = recordedSource && perceptionAvailability !== "unavailable"; - // Keep one stable original-video + world composition for every admitted - // perception recording. Layer buttons then only change entity visibility; - // they never reparent the 3D view or invalidate the operator's camera. - const unifiedPerception = recordedPerceptionSupported; + const recordedPerceptionEnabled = + showDetections2d || showSegmentation || showCuboids3d; + // The native recorded camera remains the authoritative original. Only 2D + // image-space overlays need Rerun's paired camera/world composition; 3D + // cuboids are added directly to the stable spatial view. + const unifiedPerception = recordedPerceptionSupported && + (showDetections2d || showSegmentation); const livePerceptionAvailable = !recordedSource && streamActive; const detections2dActive = recordedSource ? showDetections2d @@ -354,6 +357,16 @@ function SpatialWorkspace({ observationLayout.visibleSourceIds.has(source.id) && !source.id.startsWith("recorded.perception."), ); + const initialRecordedPlaybackStartSeconds = recordedSource + ? mediaSources.reduce((earliest, source) => { + if ( + source.id.startsWith("recorded.perception.") || + source.delivery?.kind !== "recorded-fmp4-manifest" + ) return earliest; + const start = source.delivery.timelineStartSeconds; + return earliest === undefined ? start : Math.min(earliest, start); + }, undefined) + : undefined; const presentedMediaSourceCount = unifiedPerception ? 0 : visibleMediaSources.length; const pointCloudFocused = Boolean( pointCloudSource && observationLayout.focusedSourceId === pointCloudSource.id, @@ -574,10 +587,11 @@ function SpatialWorkspace({ expectedTimelineEndSeconds={recordedSource ? state?.observationTimeline?.range?.endSeconds : undefined} + initialPlaybackStartSeconds={initialRecordedPlaybackStartSeconds} sceneSettings={sceneSettings} recordedViewResetGeneration={recordedViewResetGeneration} recordedPerceptionLayers={{ - enabled: unifiedPerception, + enabled: recordedPerceptionSupported && recordedPerceptionEnabled, detections2d: showDetections2d, segmentation: showSegmentation, cuboids3d: showCuboids3d, diff --git a/apps/control-station/test/observationSources.test.mjs b/apps/control-station/test/observationSources.test.mjs index b5939f8..a01165d 100644 --- a/apps/control-station/test/observationSources.test.mjs +++ b/apps/control-station/test/observationSources.test.mjs @@ -309,6 +309,39 @@ test("recorded blueprint fetch is bounded, strict and sends only display setting show_cuboids_3d: true, }); + await fetchRecordedBlueprintRrd( + "http://127.0.0.1:5174/api/v1/observation-sessions/session-1/blueprint.rrd", + { + accumulationSeconds: 0, + showGrid: true, + showPoints: true, + showTrajectory: true, + pointSize: 2.5, + palette: "turbo", + customColor: "#ffffff", + }, + { applicationId: "nodedc_mission_core_recorded", recordingId: "recording-001" }, + { + origin: "http://127.0.0.1:5174", + blueprintSessionId: "b".repeat(32), + perceptionLayers: { + enabled: true, + detections2d: false, + segmentation: false, + cuboids3d: true, + }, + fetcher: async (input, init) => { + calls.push({ input: String(input), init, body: JSON.parse(String(init.body)) }); + return new Response(payload, { + status: 200, + headers: { "Content-Type": "application/vnd.rerun.rrd" }, + }); + }, + }, + ); + assert.equal(calls[1].body.unified_perception, false); + assert.equal(calls[1].body.show_cuboids_3d, true); + await assert.rejects( fetchRecordedBlueprintRrd( "https://outside.invalid/api/v1/observation-sessions/session-1/blueprint.rrd", diff --git a/apps/control-station/test/rerunViewportProgressivePlayback.test.mjs b/apps/control-station/test/rerunViewportProgressivePlayback.test.mjs index 78feef2..a754faf 100644 --- a/apps/control-station/test/rerunViewportProgressivePlayback.test.mjs +++ b/apps/control-station/test/rerunViewportProgressivePlayback.test.mjs @@ -128,6 +128,51 @@ test("recorded autoplay waits for the full range and then runs exactly once", () assert.equal(gate.attempted(), true); }); +test("recorded autoplay starts at the first presentable camera frame without shrinking the range", () => { + const gate = createRecordedAutoplayGate(); + const seeks = []; + const range = { min: 0, max: 535_717_620_042 }; + + assert.equal(gate.attempt( + true, + true, + true, + range, + (value) => seeks.push(value), + () => {}, + 35_421_857_292, + ), true); + assert.deepEqual(seeks, [35_421_857_292]); + assert.deepEqual(range, { min: 0, max: 535_717_620_042 }); +}); + +test("recorded autoplay clamps an invalid presentation start to the admitted archive", () => { + const before = createRecordedAutoplayGate(); + const after = createRecordedAutoplayGate(); + const seeks = []; + const range = { min: 5, max: 10 }; + + assert.equal(before.attempt( + true, + true, + true, + range, + (value) => seeks.push(value), + () => {}, + -50, + ), true); + assert.equal(after.attempt( + true, + true, + true, + range, + (value) => seeks.push(value), + () => {}, + 50, + ), true); + assert.deepEqual(seeks, [5, 10]); +}); + test("a failed vendor autoplay attempt is consumed instead of rewinding later", () => { const gate = createRecordedAutoplayGate(); let seeks = 0; diff --git a/src/k1link/viewer/recorded.py b/src/k1link/viewer/recorded.py index 1471b0e..ea5a320 100644 --- a/src/k1link/viewer/recorded.py +++ b/src/k1link/viewer/recorded.py @@ -55,48 +55,58 @@ class _RecordedBlueprintStream: def __init__( self, application_id: str, + recording_id: str, *, view_reset_generation: Literal[0, 1], ) -> None: self.view_reset_generation = view_reset_generation self._lock = Lock() - self._pending: list[bytes] = [] - self._initial_payload: bytes | None = None - self._latest_payload = b"" + self._sequence = 0 + self._closed = False native = bindings.new_blueprint( application_id=application_id, make_default=False, make_thread_default=False, default_enabled=True, ) - self._recording = rr.RecordingStream._from_native(native) - bindings.set_callback_sink_blueprint( - lambda chunk: self._pending.append(bytes(chunk)), - True, - False, - native, + self._blueprint_recording = rr.RecordingStream._from_native(native) + self._blueprint_memory = self._blueprint_recording.memory_recording() + self._transport_recording = rr.RecordingStream( + application_id, + recording_id=recording_id, + send_properties=False, ) - self._recording.set_time("blueprint", sequence=0) + self._transport = rr.binary_stream(self._transport_recording) def render(self, blueprint: rrb.Blueprint) -> bytes: with self._lock: - if self._initial_payload is not None: - self._pending.clear() - blueprint._log_to_stream(self._recording) - self._recording.flush(timeout_sec=5.0) - delta = b"".join(self._pending) - if not delta: + if self._closed: + raise RecordedBlueprintError("stable blueprint stream is closed") + self._blueprint_recording.set_time( + "blueprint", + sequence=self._sequence, + ) + self._sequence += 1 + blueprint._log_to_stream(self._blueprint_recording) + self._blueprint_recording.flush(timeout_sec=5.0) + bindings.send_blueprint( + self._blueprint_memory.storage, + True, + False, + self._transport_recording.to_native(), + ) + payload = self._transport.read(flush=True, flush_timeout_sec=5.0) + if not payload: raise RecordedBlueprintError("stable blueprint stream produced no data") - if self._initial_payload is None: - self._initial_payload = delta - self._latest_payload = b"" - else: - self._latest_payload = delta - return self._initial_payload + self._latest_payload + return payload def close(self) -> None: - with suppress(Exception): - self._recording.disconnect() + with self._lock: + self._closed = True + with suppress(Exception): + self._blueprint_recording.disconnect() + with suppress(Exception): + self._transport_recording.disconnect() _MAX_RECORDED_BLUEPRINT_STREAMS = 32 @@ -123,6 +133,7 @@ def _stable_recorded_blueprint_stream( if stream is None: stream = _RecordedBlueprintStream( application_id, + recording_id, view_reset_generation=view_reset_generation, ) _recorded_blueprint_streams[key] = stream @@ -197,7 +208,13 @@ def recorded_blueprint( # object-history trail when the operator widens the cloud window. "/world/points": point_overrides, "/world/trajectory": trajectory_overrides, - "/world/perception": rrb.EntityBehavior(visible=False), + "/world/perception": rrb.EntityBehavior(visible=show_cuboids_3d), + "/world/perception/lidar": rrb.EntityBehavior(visible=False), + "/world/perception/support": rrb.EntityBehavior(visible=False), + "/world/perception/semantic_points": rrb.EntityBehavior(visible=False), + "/world/perception/boxes3d": rrb.EntityBehavior( + visible=show_cuboids_3d, + ), }, # Log an explicit empty range set so a previous accumulated blueprint # for this stable view id is cleared instead of surviving in Rerun. diff --git a/tests/test_rrd_export.py b/tests/test_rrd_export.py index bad8607..4cc1886 100644 --- a/tests/test_rrd_export.py +++ b/tests/test_rrd_export.py @@ -12,6 +12,7 @@ from pathlib import Path import pytest import k1link.device_plugins.xgrids_k1.rrd_export as export_module +import k1link.viewer.recorded as viewer_recorded_module from k1link.device_plugins.xgrids_k1.mqtt.capture import ( CAPTURE_CLOCK_FILENAME, FRAME_HEADER, @@ -299,6 +300,39 @@ def test_zero_accumulation_uses_latest_frame_instead_of_empty_time_range() -> No assert point_components == {"Points3D:radii"} +def test_cuboids_are_overlaid_on_the_stable_spatial_view_without_unifying_video() -> None: + baseline = viewer_recorded_blueprint( + RerunSceneSettings(accumulation_seconds=12.0), + include_initial_playback_state=False, + unified_perception=False, + show_cuboids_3d=False, + ) + blueprint = viewer_recorded_blueprint( + RerunSceneSettings(accumulation_seconds=12.0), + include_initial_playback_state=False, + unified_perception=False, + show_cuboids_3d=True, + ) + spatial_view = blueprint.root_container.contents[0] + + assert spatial_view.id == baseline.root_container.contents[0].id + assert spatial_view.visualizer_overrides[ + "/world/perception" + ].visible.as_arrow_array().to_pylist() == [True] + assert spatial_view.visualizer_overrides[ + "/world/perception/lidar" + ].visible.as_arrow_array().to_pylist() == [False] + assert spatial_view.visualizer_overrides[ + "/world/perception/support" + ].visible.as_arrow_array().to_pylist() == [False] + assert spatial_view.visualizer_overrides[ + "/world/perception/semantic_points" + ].visible.as_arrow_array().to_pylist() == [False] + assert spatial_view.visualizer_overrides[ + "/world/perception/boxes3d" + ].visible.as_arrow_array().to_pylist() == [True] + + def test_dynamic_blueprint_reuses_scene_ids_without_playback_mutation() -> None: first = _recorded_blueprint( RerunSceneSettings(show_points=False, show_trajectory=True), @@ -416,7 +450,26 @@ def test_viewer_blueprint_reset_is_bounded_and_recreates_render_views() -> None: assert perception_behavior.visible.as_arrow_array().to_pylist() == [False] -def test_recorded_blueprint_layer_updates_preserve_store_until_explicit_reset() -> None: +def test_recorded_blueprint_layer_updates_preserve_store_until_explicit_reset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + activations: list[tuple[object, bool, bool]] = [] + send_blueprint = viewer_recorded_module.bindings.send_blueprint + + def capture_activation( + storage: object, + make_active: bool, + make_default: bool, + recording: object, + ) -> None: + activations.append((storage, make_active, make_default)) + send_blueprint(storage, make_active, make_default, recording) + + monkeypatch.setattr( + viewer_recorded_module.bindings, + "send_blueprint", + capture_activation, + ) session_id = "b" * 32 initial = viewer_recorded_blueprint_rrd( RerunSceneSettings(accumulation_seconds=12.0), @@ -452,6 +505,13 @@ def test_recorded_blueprint_layer_updates_preserve_store_until_explicit_reset() assert layer_store_ids == initial_store_ids assert len(reset_store_ids) == 1 assert reset_store_ids.isdisjoint(initial_store_ids) + assert [activation[1:] for activation in activations] == [ + (True, False), + (True, False), + (True, False), + ] + assert activations[0][0] is activations[1][0] + assert activations[2][0] is not activations[0][0] assert len(initial) < 350_000 assert len(layers_enabled) < 350_000 assert len(reset) < 350_000