from dataclasses import replace from types import SimpleNamespace import numpy as np import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from rerun.experimental import RrdReader from test_map_reference_version import fixture as map_fixture from test_rrd_export import _point_payload, _pose_payload, _write_capture from test_session_recording import _command from k1link.device_plugins.xgrids_k1.rrd_export import export_k1mqtt_to_rrd from k1link.missions.default_sources import DefaultPlanningSources from k1link.missions.drafts import MissionDrafts from k1link.reconstruction.recorded_geometry import RecordedMapGeometry from k1link.reconstruction.session_versions import SessionMapVersions from k1link.sessions.map_recording import MapReplaySessionStore from k1link.sessions.models import ReplayMapVersion, SessionIntegrityError from k1link.sessions.preparation import _source_identity from k1link.sessions.recording import ( RecordingMaterializationError, _validate_source, _validated_artifact_digests, ) from k1link.web.mission_planner_api import build_mission_planner_router @pytest.fixture def fixture(tmp_path): return map_fixture.__wrapped__(tmp_path) def sources(f): f.original.store = SimpleNamespace(data_dir=f.root) f.original.root = f.root / "planning" versions = SessionMapVersions(f.root / "data") return DefaultPlanningSources(f.original, versions), versions def test_default_is_explicit_durable_and_preserves_original_and_old_draft(fixture): f = fixture planning, versions = sources(f) version = f.publish() assert planning.get("source-a")["generation"] == "a" * 64 drafts = MissionDrafts(f.root / "drafts", planning) args = dict( session_id="source-a", generation="a" * 64, id=None, revision=0, name="Before correction", start_index=0, end_index=2, direction="forward", ) old = drafts.save(SimpleNamespace(**args)) versions.activate(version, f.original) current = planning.get("source-a") assert current["generation"] == version.generation assert current["label"] == f.source["label"] # ordinary catalog name np.testing.assert_array_equal([p["position"] for p in current["poses"]], f.corrected) new = drafts.save(SimpleNamespace(**{**args, "generation": current["generation"]})) assert new["zone"]["generation"] != old["zone"]["generation"] assert drafts.get(old["id"]) == old assert drafts.check(old["id"], 1)["vehicle_control"] is False assert SessionMapVersions(f.root / "data").selected("source-a").generation == version.generation assert planning.bound("source-a", "a" * 64)["poses"] == f.source["poses"] points, metadata = planning.submap("source-a", current["generation"], 0, 2) np.testing.assert_array_equal(points, f.corrected) assert metadata["generation"] == current["generation"] app = FastAPI() app.include_router(build_mission_planner_router(drafts)) with TestClient(app) as client: url = "/api/v1/mission-planner/sources/source-a" assert client.get(url).json()["generation"] == version.generation assert client.get(url, params={"generation": "a" * 64}).json()["generation"] == "a" * 64 assert client.get(url, params={"generation": "c" * 64}).status_code == 409 def test_missing_or_corrupt_default_never_silently_becomes_original(fixture): f = fixture planning, versions = sources(f) version = versions.activate(f.publish(), f.original) file = version.directory / "points.f32" original = file.read_bytes() file.write_bytes(b"x" * len(original)) with pytest.raises(ValueError, match="identity changed"): planning.get("source-a") file.write_bytes(original) version.directory.rename(version.directory.with_name("removed-version")) with pytest.raises(ValueError, match="not substituted"): planning.get("source-a") def test_replay_binding_keeps_native_root_and_raw_identity(fixture): f = fixture command = _command(f.root / "capture") f.source.update( session_id=command.session_id, source_digests=_validated_artifact_digests(_validate_source(command)), ) # The fixture parent is source-a-specific; the replay binding itself verifies # real native digests independently of planning-source fixtures. version = f.publish() pinned = replace(command, map_version=ReplayMapVersion(version.directory, version.generation)) source = _validate_source(pinned) assert pinned.session_root == command.session_root assert source.primary.path == command.primary_artifact.path assert len(source.artifacts) == len(command.artifacts) + 3 assert _source_identity(pinned) != _source_identity(command) assert ( _validated_artifact_digests(source)[command.primary_artifact_id] == f.source["source_digests"][command.primary_artifact_id] ) wrong = replace(pinned, session_id="some-other-session") with pytest.raises(RecordingMaterializationError): _validate_source(wrong) facade = MapReplaySessionStore( SimpleNamespace(prepare_replay=lambda *_: command), SimpleNamespace(resolve_replay=lambda _: wrong), ) with pytest.raises(SessionIntegrityError): facade.prepare_replay(command.session_id) def test_projector_owns_full_frames_pose_orientation_and_clock(fixture): f = fixture version = f.publish() geometry = RecordedMapGeometry( { "map-version-" + name: version.directory / name for name in ("manifest.json", "points.f32", "trajectory.npz") } ) for i in range(3): np.testing.assert_array_equal(geometry.points(i, i, 2), f.points[i * 2 : i * 2 + 2]) np.testing.assert_array_equal(geometry.pose(i, i)[0], f.corrected[i]) geometry.complete(3, 3) with pytest.raises(ValueError, match="clock"): geometry.pose(0, 0.5) with pytest.raises(ValueError, match="ownership"): geometry.points(0, 0, 3) with pytest.raises(ValueError, match="all native"): geometry.complete(2, 3) assert geometry.recording_id("original") != "original" def test_overview_default_follows_map_but_original_comparison_stays_available(fixture): from k1link.sessions.overview import SessionOverviewService f = fixture _, versions = sources(f) version = versions.activate(f.publish(), f.original) service = SessionOverviewService(SimpleNamespace(data_dir=f.root), {}, map_versions=versions) pair = SimpleNamespace( document={ "map_generation": version.generation, "source_digests": f.source["source_digests"], } ) try: assert service.default_representation("source-a", pair) == "corrected" assert service.default_representation("source-a", pair, version.generation) == "corrected" assert service.default_representation("source-a", pair, "a" * 64) == "original" with pytest.raises(ValueError, match="unavailable"): service.default_representation("source-a", None) finally: service.close() def test_corrected_rrd_retains_clock_and_replaces_cloud_pose_and_route(tmp_path): frames = [("unrelated", b"{}", 2_000_000_000)] for i in range(6): frames += [ ("RealtimePointcloud", _point_payload(i), 2_100_000_000 + i * 200_000_000), ("RealtimePath", _pose_payload(i), 2_200_000_000 + i * 200_000_000), ] raw = _write_capture(tmp_path, frames, capture_clock_offsets_ns=(0, 4_000_000_000)) class Geometry: def recording_id(self, original): return "corrected-" + original def points(self, index, time, count): assert count == 1 and time == pytest.approx(0.1 + index * 0.2) return np.array([[100 + index, 200, 300]], dtype=np.float32) def pose(self, index, time): assert time == pytest.approx(0.2 + index * 0.2) return (100.0 + index, 200.0, 300.0), (0.0, 0.0, 1.0, 0.0) def complete(self, clouds, poses): assert (clouds, poses) == (6, 6) original = export_k1mqtt_to_rrd(raw, tmp_path / "original.rrd") corrected = export_k1mqtt_to_rrd(raw, tmp_path / "corrected.rrd", map_geometry=Geometry()) assert corrected["source_sha256"] == original["source_sha256"] assert corrected["timeline_start_ns"] == original["timeline_start_ns"] == 0 assert corrected["timeline_end_ns"] == original["timeline_end_ns"] == 4_000_000_000 assert corrected["recording_id"] != original["recording_id"] clouds, poses, route = [], [], [] for chunk in RrdReader(tmp_path / "corrected.rrd").stream(): batch = chunk.to_record_batch() if chunk.entity_path == "/world/points": clouds.extend(batch.column("Points3D:positions").to_pylist()) if ( chunk.entity_path == "/world/sensor_pose" and "Transform3D:translation" in batch.schema.names ): poses.extend(batch.column("Transform3D:translation").to_pylist()) if chunk.entity_path == "/world/trajectory" and "LineStrips3D:strips" in batch.schema.names: route.extend(batch.column("LineStrips3D:strips").to_pylist()) np.testing.assert_array_equal( np.array(clouds).reshape(-1, 3), [[100 + index, 200, 300] for index in range(6)] ) assert poses and np.array(poses).reshape(-1, 3)[-1, 0] == 105 # Existing operator cadence is preserved; every published route vertex is corrected. np.testing.assert_array_equal( np.array(route[-1]).reshape(-1, 3), [[100, 200, 300], [103, 200, 300]] )