perf(lidar): avoid duplicate replay decode during build

This commit is contained in:
DCCONSTRUCTIONS
2026-08-31 18:35:12 +03:00
parent f64fbb1fa6
commit 73a7d28065
2 changed files with 56 additions and 7 deletions
+30
View File
@@ -10,6 +10,7 @@ import pytest
from fastapi import APIRouter
from fastapi.routing import APIRoute
import k1link.compute.lidar_replay as lidar_replay_module
from k1link.compute import (
K1_LIDAR_PACK_V2_PROFILE,
K1LocalSurfaceV1,
@@ -253,6 +254,35 @@ def test_lidar_replay_v2_retains_fields_and_passes_equivalence(tmp_path: Path) -
assert build_lidar_replay_pack_v2(capture, tmp_path / "packs") == output
def test_lidar_replay_build_decodes_source_once(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
capture = _capture(tmp_path)
original_capture_arrays = lidar_replay_module._capture_arrays
calls: list[Path] = []
def counting_capture_arrays(path: Path) -> dict[str, Any]:
calls.append(path)
return original_capture_arrays(path)
monkeypatch.setattr(
lidar_replay_module,
"_capture_arrays",
counting_capture_arrays,
)
output = build_lidar_replay_pack_v2(capture, tmp_path / "packs")
pack = LidarReplayPackV2(output)
try:
assert pack.equivalence["status"] == "passed"
assert pack.equivalence["array_mismatches"] == 0
finally:
pack.close()
assert calls == [capture.resolve(strict=True)]
def test_lidar_replay_v2_unblocks_detector_input_but_not_mapping() -> None:
assessments = {
item.stage: item for item in assess_lidar_profile(K1_LIDAR_PACK_V2_PROFILE)