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=" 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)