feat(perception): qualify lossless lidar observations

This commit is contained in:
DCCONSTRUCTIONS
2026-07-30 11:37:57 +03:00
parent e56fa0c074
commit 7d1a70d8e0
13 changed files with 1874 additions and 119 deletions
+40 -1
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import json
import struct
from pathlib import Path
from typing import Any, cast
import lz4.block
import pytest
@@ -11,11 +12,13 @@ from fastapi.routing import APIRoute
from k1link.compute import (
K1_LIDAR_PACK_V2_PROFILE,
K1LocalSurfaceV1,
LidarPipelineStage,
LidarReadiness,
LidarReplayError,
LidarReplayPackV2,
assess_lidar_profile,
build_k1_local_surface,
build_lidar_replay_pack_v2,
verify_lidar_replay_equivalence,
)
@@ -163,7 +166,11 @@ def _capture(tmp_path: Path) -> Path:
def _endpoint(router: APIRouter, path: str) -> object:
for route in router.routes:
if isinstance(route, APIRoute) and route.path == path and "GET" in route.methods:
if (
isinstance(route, APIRoute)
and route.path == path
and "GET" in (route.methods or set())
):
return route.endpoint
raise AssertionError(f"GET {path} route is missing")
@@ -208,6 +215,38 @@ def test_lidar_replay_v2_retains_fields_and_passes_equivalence(tmp_path: Path) -
assert pack.equivalence["array_mismatches"] == 0
rerun = verify_lidar_replay_equivalence(capture, pack)
assert rerun["status"] == "passed"
surface_output = build_k1_local_surface(
pack,
tmp_path / "local-surfaces",
display_name="synthetic replay v2 local surface",
)
surface = K1LocalSurfaceV1(surface_output)
try:
assert surface.identity["source_pack_id"] == pack.pack_id
assert (
surface.identity["source_pack_identity_sha256"]
== pack.manifest["identity_sha256"]
)
assert (
surface.identity["source_schema_version"]
== "missioncore.lidar-replay-pack/v2"
)
assert (
surface.identity["source_representation"]
== "lossless-lidar-replay-v2"
)
assert surface.identity["frame_count"] == pack.point_frame_count
assert surface.identity["point_count"] == pack.point_count
surface_source = cast(dict[str, Any], surface.report["source"])
assert surface_source["intensity_available"] is True
detail = surface.frame_detail(pack, 0)
assert detail["source_pack_id"] == pack.pack_id
assert detail["source_frame_index"] == 1
detail_pose = cast(dict[str, Any], detail["pose"])
assert detail_pose["binding_age_ms"] == pytest.approx(10.0)
finally:
surface.close()
finally:
pack.close()