223 lines
9.5 KiB
Python
223 lines
9.5 KiB
Python
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]]
|
|
)
|