89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from k1link.compute.live_perception import TELEMETRY_SCHEMA, WORLD_STATE_SCHEMA
|
|
from k1link.compute.live_replay_qualification import (
|
|
_validate_telemetry_rows,
|
|
_validate_world_rows,
|
|
)
|
|
from k1link.sessions import SessionIntegrityError
|
|
|
|
|
|
def _world_row() -> dict[str, object]:
|
|
return {
|
|
"schema_version": WORLD_STATE_SCHEMA,
|
|
"frame_index": 3,
|
|
"source_frame_index": 1003,
|
|
"session_seconds": 12.3,
|
|
"coordinate_frames": {
|
|
"world": "k1-map",
|
|
"sensor_relative": "k1-lidar",
|
|
"vehicle_body": "unavailable-no-rig-to-vehicle-transform",
|
|
},
|
|
"fusion_state": "fused",
|
|
"objects": [
|
|
{
|
|
"track_id": 7,
|
|
"range_m": 4.0,
|
|
"position_map_m": [1.0, 2.0, 3.0],
|
|
"position_lidar_m": [4.0, 0.0, 0.0],
|
|
"size_m": [4.0, 2.0, 1.5],
|
|
"velocity_map_mps": None,
|
|
"velocity_status": "unavailable-insufficient-history",
|
|
"geometry": "point-supported-visible-surface-envelope",
|
|
}
|
|
],
|
|
"object_count": 1,
|
|
"clearance": {"state": "observed"},
|
|
"delivery": {"health": "healthy", "result_age_ms": 12.0},
|
|
}
|
|
|
|
|
|
def test_world_and_telemetry_rows_preserve_exact_frame_binding(tmp_path: Path) -> None:
|
|
world = tmp_path / "world-state.jsonl"
|
|
telemetry = tmp_path / "telemetry.jsonl"
|
|
world.write_text(json.dumps(_world_row()) + "\n", encoding="utf-8")
|
|
telemetry.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": TELEMETRY_SCHEMA,
|
|
"frame_index": 3,
|
|
"end_to_end_latency_ms": 12.0,
|
|
"health": "healthy",
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
indices = _validate_world_rows(
|
|
world,
|
|
tmp_path,
|
|
expected_count=1,
|
|
selection_start=3,
|
|
selection_count=1,
|
|
)
|
|
_validate_telemetry_rows(telemetry, tmp_path, expected_indices=indices)
|
|
|
|
assert indices == (3,)
|
|
|
|
|
|
def test_world_row_rejects_invented_object_count(tmp_path: Path) -> None:
|
|
world = tmp_path / "world-state.jsonl"
|
|
value = _world_row()
|
|
value["object_count"] = 2
|
|
world.write_text(json.dumps(value) + "\n", encoding="utf-8")
|
|
|
|
with pytest.raises(SessionIntegrityError, match="world-state row"):
|
|
_validate_world_rows(
|
|
world,
|
|
tmp_path,
|
|
expected_count=1,
|
|
selection_start=3,
|
|
selection_count=1,
|
|
)
|