316 lines
9.8 KiB
Python
316 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from k1link.compute.degradation_recovery import (
|
|
DegradationKind,
|
|
DegradationRecoveryError,
|
|
DegradationScenario,
|
|
transform_track_geometry_frame,
|
|
)
|
|
from k1link.compute.e35_degradation_replay import (
|
|
E35DegradationReplayError,
|
|
read_e35_degradation_profile,
|
|
)
|
|
from k1link.compute.track_geometry import (
|
|
PointSlab,
|
|
TrackGeometry,
|
|
TrackGeometryCurrentness,
|
|
TrackGeometryEvidenceState,
|
|
TrackGeometryFrame,
|
|
TrackGeometryMetricBasis,
|
|
TrackGeometryOwnerKind,
|
|
TrackGeometrySourceBinding,
|
|
)
|
|
|
|
|
|
def _binding() -> TrackGeometrySourceBinding:
|
|
return TrackGeometrySourceBinding(
|
|
source_pack_id="e10-lidar-pack-" + "a" * 64,
|
|
source_session_id="source-session",
|
|
representation_profile_id="representation-profile/v1",
|
|
e31_qualification_id="e31-source-qualification-" + "b" * 64,
|
|
calibration_sha256="c" * 64,
|
|
coordinate_frame="map",
|
|
time_basis="source-time",
|
|
selected_offset_ms=0,
|
|
)
|
|
|
|
|
|
def _frame(frame_index: int = 10) -> TrackGeometryFrame:
|
|
geometries = (
|
|
TrackGeometry(
|
|
owner_key="track:7",
|
|
owner_kind=TrackGeometryOwnerKind.CAMERA_TRACK,
|
|
evidence_state=TrackGeometryEvidenceState.AGREE,
|
|
currentness=TrackGeometryCurrentness.CURRENT,
|
|
metric_basis=TrackGeometryMetricBasis.CURRENT_POINTS,
|
|
reason_codes=("accepted-hit-backed-support",),
|
|
semantic_track_id=7,
|
|
semantic_label="car",
|
|
bbox_xyxy=(1.0, 1.0, 2.0, 2.0),
|
|
range_m=3.0,
|
|
),
|
|
TrackGeometry(
|
|
owner_key="geometry:8",
|
|
owner_kind=TrackGeometryOwnerKind.GEOMETRY_CLUSTER,
|
|
evidence_state=TrackGeometryEvidenceState.GEOMETRY_ONLY,
|
|
currentness=TrackGeometryCurrentness.CURRENT,
|
|
metric_basis=TrackGeometryMetricBasis.CURRENT_POINTS,
|
|
reason_codes=("accepted-hit-backed-support",),
|
|
range_m=4.0,
|
|
),
|
|
TrackGeometry(
|
|
owner_key="track:9",
|
|
owner_kind=TrackGeometryOwnerKind.CAMERA_TRACK,
|
|
evidence_state=TrackGeometryEvidenceState.CAMERA_ONLY,
|
|
currentness=TrackGeometryCurrentness.CURRENT,
|
|
metric_basis=TrackGeometryMetricBasis.UNAVAILABLE,
|
|
reason_codes=("camera-without-current-points",),
|
|
semantic_track_id=9,
|
|
semantic_label="person",
|
|
bbox_xyxy=(2.0, 2.0, 3.0, 4.0),
|
|
),
|
|
)
|
|
return TrackGeometryFrame(
|
|
binding=_binding(),
|
|
frame_index=frame_index,
|
|
source_frame_index=frame_index,
|
|
session_seconds=frame_index / 10.0,
|
|
source_available=True,
|
|
point_slab=PointSlab(
|
|
frame_index=frame_index,
|
|
source_frame_index=frame_index,
|
|
source_point_count=4,
|
|
coordinate_frame="map",
|
|
owner_keys=("track:7", "geometry:8"),
|
|
source_indices=np.asarray([0, 1, 2, 3], dtype="<i8"),
|
|
points_xyz_m=np.asarray(
|
|
[
|
|
[1.0, 0.0, 0.0],
|
|
[1.1, 0.0, 0.0],
|
|
[2.0, 0.0, 0.0],
|
|
[2.1, 0.0, 0.0],
|
|
],
|
|
dtype="<f4",
|
|
),
|
|
owner_indices=np.asarray([0, 0, 1, 1], dtype="<u4"),
|
|
),
|
|
geometries=geometries,
|
|
)
|
|
|
|
|
|
def _scenario(
|
|
kind: DegradationKind,
|
|
parameters: dict[str, object],
|
|
*,
|
|
frame_start: int = 10,
|
|
frame_end: int = 20,
|
|
) -> DegradationScenario:
|
|
return DegradationScenario(
|
|
scenario_id=kind.value,
|
|
kind=kind,
|
|
frame_start=frame_start,
|
|
frame_end=frame_end,
|
|
parameters=parameters,
|
|
)
|
|
|
|
|
|
def test_camera_loss_removes_semantics_but_retains_geometry_points() -> None:
|
|
transformed = transform_track_geometry_frame(
|
|
_frame(),
|
|
_scenario(
|
|
DegradationKind.CAMERA_LOSS,
|
|
{"drop_camera_observations": True},
|
|
),
|
|
)
|
|
|
|
assert transformed.frame.point_slab.row_count == 4
|
|
assert all(
|
|
geometry.owner_kind is TrackGeometryOwnerKind.GEOMETRY_CLUSTER
|
|
for geometry in transformed.frame.geometries
|
|
)
|
|
assert all(
|
|
geometry.semantic_label is None
|
|
for geometry in transformed.frame.geometries
|
|
)
|
|
assert transformed.transformation["channels"]["camera"] == "unavailable"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("kind", "parameters", "expected_source_available"),
|
|
[
|
|
(
|
|
DegradationKind.LIDAR_LOSS,
|
|
{"drop_lidar_points": True},
|
|
False,
|
|
),
|
|
(
|
|
DegradationKind.POSE_STALENESS,
|
|
{"pose_age_seconds": 0.5},
|
|
True,
|
|
),
|
|
],
|
|
)
|
|
def test_metric_loss_keeps_camera_nonmetric_and_withholds_map_rows(
|
|
kind: DegradationKind,
|
|
parameters: dict[str, object],
|
|
expected_source_available: bool,
|
|
) -> None:
|
|
transformed = transform_track_geometry_frame(
|
|
_frame(),
|
|
_scenario(kind, parameters),
|
|
)
|
|
|
|
assert transformed.frame.point_slab.row_count == 0
|
|
assert transformed.frame.source_available is expected_source_available
|
|
assert all(
|
|
geometry.owner_kind is TrackGeometryOwnerKind.CAMERA_TRACK
|
|
for geometry in transformed.frame.geometries
|
|
)
|
|
assert all(
|
|
geometry.metric_basis is not TrackGeometryMetricBasis.CURRENT_POINTS
|
|
for geometry in transformed.frame.geometries
|
|
)
|
|
assert all(
|
|
geometry.evidence_state is not TrackGeometryEvidenceState.AGREE
|
|
for geometry in transformed.frame.geometries
|
|
)
|
|
|
|
|
|
def test_delayed_frame_is_discarded_and_cannot_retain_claims() -> None:
|
|
transformed = transform_track_geometry_frame(
|
|
_frame(),
|
|
_scenario(
|
|
DegradationKind.DELAYED_FRAMES,
|
|
{"delay_seconds": 1.0, "late_result_policy": "discard"},
|
|
),
|
|
)
|
|
|
|
assert not transformed.frame.source_available
|
|
assert transformed.frame.point_slab.row_count == 0
|
|
assert transformed.frame.geometries == ()
|
|
assert transformed.transformation["action"] == "late-frame-discarded"
|
|
|
|
|
|
def test_bounded_drop_is_deterministic_and_journals_pass_frames() -> None:
|
|
scenario = _scenario(
|
|
DegradationKind.BOUNDED_DROP,
|
|
{"drop_every_nth_frame": 3},
|
|
)
|
|
dropped = transform_track_geometry_frame(_frame(10), scenario)
|
|
passed_input = _frame(11)
|
|
passed = transform_track_geometry_frame(passed_input, scenario)
|
|
|
|
assert dropped.frame.geometries == ()
|
|
assert dropped.transformation["action"] == "input-frame-dropped"
|
|
assert passed.frame is passed_input
|
|
assert passed.frame.geometries
|
|
assert passed.transformation["action"] == "bounded-drop-pass"
|
|
|
|
|
|
def test_timing_offset_splits_camera_and_geometry_without_agree() -> None:
|
|
transformed = transform_track_geometry_frame(
|
|
_frame(),
|
|
_scenario(
|
|
DegradationKind.TIMING_OFFSET,
|
|
{"camera_lidar_offset_ms": 250},
|
|
),
|
|
)
|
|
|
|
camera = [
|
|
geometry
|
|
for geometry in transformed.frame.geometries
|
|
if geometry.owner_kind is TrackGeometryOwnerKind.CAMERA_TRACK
|
|
]
|
|
geometry = [
|
|
item
|
|
for item in transformed.frame.geometries
|
|
if item.owner_kind is TrackGeometryOwnerKind.GEOMETRY_CLUSTER
|
|
]
|
|
assert transformed.frame.point_slab.row_count == 4
|
|
assert len(camera) == 2
|
|
assert len(geometry) == 2
|
|
assert all(
|
|
item.evidence_state is not TrackGeometryEvidenceState.AGREE
|
|
for item in transformed.frame.geometries
|
|
)
|
|
assert transformed.frame.point_slab.owner_keys == (
|
|
"e35-timing-offset:track:7",
|
|
"geometry:8",
|
|
)
|
|
|
|
|
|
def test_outside_window_returns_exact_frame_with_inactive_journal() -> None:
|
|
frame = _frame(9)
|
|
transformed = transform_track_geometry_frame(
|
|
frame,
|
|
_scenario(
|
|
DegradationKind.CAMERA_LOSS,
|
|
{"drop_camera_observations": True},
|
|
),
|
|
)
|
|
|
|
assert transformed.frame is frame
|
|
assert transformed.transformation["active"] is False
|
|
assert transformed.transformation["original_frame_sha256"] == (
|
|
transformed.transformation["transformed_frame_sha256"]
|
|
)
|
|
|
|
|
|
def test_scenario_requires_exact_parameters_and_unsafe_offset_fails() -> None:
|
|
with pytest.raises(DegradationRecoveryError):
|
|
_scenario(DegradationKind.CAMERA_LOSS, {})
|
|
with pytest.raises(DegradationRecoveryError):
|
|
_scenario(
|
|
DegradationKind.TIMING_OFFSET,
|
|
{"camera_lidar_offset_ms": 100},
|
|
)
|
|
|
|
|
|
def test_source_frame_and_input_arrays_are_not_mutated() -> None:
|
|
frame = _frame()
|
|
before = copy.deepcopy(frame.to_dict())
|
|
before_points = frame.point_slab.points_xyz_m.copy()
|
|
|
|
transform_track_geometry_frame(
|
|
frame,
|
|
_scenario(
|
|
DegradationKind.CAMERA_LOSS,
|
|
{"drop_camera_observations": True},
|
|
),
|
|
)
|
|
|
|
assert frame.to_dict() == before
|
|
np.testing.assert_array_equal(frame.point_slab.points_xyz_m, before_points)
|
|
|
|
|
|
def test_frozen_profile_contains_every_degradation_once() -> None:
|
|
profile, scenarios = read_e35_degradation_profile(
|
|
Path("experiments/perception/e35_deterministic_degradation_profile.json")
|
|
)
|
|
|
|
assert profile["profile_id"] == "e35-six-channel-degradation-recovery/v1"
|
|
assert {scenario.kind for scenario in scenarios} == set(DegradationKind)
|
|
assert sum(scenario.frame_count for scenario in scenarios) == 360
|
|
|
|
|
|
def test_profile_rejects_authority_and_policy_relaxation(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source = Path(
|
|
"experiments/perception/e35_deterministic_degradation_profile.json"
|
|
)
|
|
profile = json.loads(source.read_text(encoding="utf-8"))
|
|
profile["policy"]["late_results_may_reenter"] = True
|
|
path = tmp_path / "relaxed.json"
|
|
path.write_text(json.dumps(profile), encoding="utf-8")
|
|
|
|
with pytest.raises(E35DegradationReplayError):
|
|
read_e35_degradation_profile(path)
|