from __future__ import annotations import math import zipfile from pathlib import Path import pytest from k1link.compute.kitti_pointpillars_benchmark import ( KittiLidarTruth, KittiPointPillarsBenchmarkError, PointPillarsFramePrediction, evaluate_pointpillars_predictions, read_kitti_validation_truth, ) from k1link.compute.pointpillars_postprocess import PointPillarsBox def _calibration() -> str: return "\n".join( [ "R0_rect: 1 0 0 0 1 0 0 0 1", "Tr_velo_to_cam: 1 0 0 0 0 1 0 0 0 0 1 0", "", "", ] ) def _label(class_name: str, x_m: float) -> str: return f"{class_name} 0 0 0 0 0 10 10 1.5 2 4 {x_m} 0 0 0\n" def _truth( frame_id: str, class_name: str, *, x_m: float = 10.0, ) -> KittiLidarTruth: return KittiLidarTruth( frame_id=frame_id, benchmark_class=class_name, x_m=x_m, y_m=0.0, z_m=0.0, length_m=4.0, width_m=2.0, height_m=1.5, yaw_rad=0.0, ) def _prediction_box( model_class: str, *, x_m: float = 10.0, score: float = 0.9, ) -> PointPillarsBox: return PointPillarsBox( x_m=x_m, y_m=0.0, z_m=0.0, length_m=4.0, width_m=2.0, height_m=1.5, yaw_rad=0.0, class_id={"Vehicle": 0, "Pedestrian": 1, "Cyclist": 2}[model_class], model_class=model_class, score=score, ) def _perfect_fixture() -> tuple[ dict[str, tuple[KittiLidarTruth, ...]], tuple[PointPillarsFramePrediction, ...], ]: classes = ( ("000000", "Car", "Vehicle"), ("000001", "Pedestrian", "Pedestrian"), ("000002", "Cyclist", "Cyclist"), ) truth = { frame_id: (_truth(frame_id, benchmark_class),) for frame_id, benchmark_class, _ in classes } predictions = tuple( PointPillarsFramePrediction( frame_id=frame_id, boxes=(_prediction_box(model_class),), inference_ms=50.0 + index, ) for index, (frame_id, _, model_class) in enumerate(classes) ) return truth, predictions def test_reads_and_converts_kitti_camera_bottom_centers( tmp_path: Path, ) -> None: labels = tmp_path / "labels.zip" calibrations = tmp_path / "calib.zip" with zipfile.ZipFile(labels, "w") as archive: archive.writestr( "training/label_2/000000.txt", _label("Car", 10.0), ) archive.writestr( "training/label_2/000001.txt", _label("Pedestrian", 11.0), ) archive.writestr( "training/label_2/000002.txt", _label("Cyclist", 12.0), ) with zipfile.ZipFile(calibrations, "w") as archive: for frame_id in ("000000", "000001", "000002"): archive.writestr( f"training/calib/{frame_id}.txt", _calibration(), ) truth = read_kitti_validation_truth( labels_archive=labels, calibrations_archive=calibrations, validation_frame_ids=("000000", "000001", "000002"), ) car = truth["000000"][0] assert car.x_m == 10.0 assert car.z_m == pytest.approx(0.75) assert car.yaw_rad == pytest.approx(-math.pi / 2.0) assert car.length_m == 4.0 assert car.width_m == 2.0 def test_perfect_predictions_produce_complete_metrics() -> None: truth, predictions = _perfect_fixture() report = evaluate_pointpillars_predictions( truth_by_frame=truth, predictions=predictions, ) assert report["aggregates"]["bev_map40"] == pytest.approx(1.0) assert report["aggregates"]["3d_map40"] == pytest.approx(1.0) assert report["aggregates"]["false_occupied_rate"] == 0.0 assert report["aggregates"]["center_error_m"]["mean"] == 0.0 assert report["aggregates"]["range_error_m"]["mean"] == 0.0 assert report["aggregates"]["yaw_error_rad"]["mean"] == 0.0 assert report["aggregates"]["distance_bucket_recall"]["0-20m"]["recall"] == 1.0 assert report["metric_contract"]["evaluation_kind"] == ( "public-cross-domain-transfer-probe" ) assert report["claim_boundary"]["native_model_accuracy_evaluated"] is False assert report["claim_boundary"]["k1_transfer_evaluated"] is False def test_false_prediction_reduces_precision_and_counts_false_occupied() -> None: truth, predictions = _perfect_fixture() first = predictions[0] predictions = ( PointPillarsFramePrediction( frame_id=first.frame_id, boxes=( _prediction_box("Vehicle", x_m=40.0, score=0.95), *first.boxes, ), inference_ms=first.inference_ms, ), *predictions[1:], ) report = evaluate_pointpillars_predictions( truth_by_frame=truth, predictions=predictions, ) assert report["per_class"]["Car"]["true_positives"] == 1 assert report["per_class"]["Car"]["false_positives"] == 1 assert report["per_class"]["Car"]["precision"] == pytest.approx(0.5) assert report["aggregates"]["false_occupied_rate"] == pytest.approx(0.25) def test_predictions_outside_shared_cross_domain_range_are_not_false_positives() -> None: truth, predictions = _perfect_fixture() first = predictions[0] predictions = ( PointPillarsFramePrediction( frame_id=first.frame_id, boxes=( _prediction_box("Vehicle", x_m=-10.0, score=0.95), *first.boxes, ), inference_ms=first.inference_ms, ), *predictions[1:], ) report = evaluate_pointpillars_predictions( truth_by_frame=truth, predictions=predictions, ) assert report["per_class"]["Car"]["false_positives"] == 0 assert report["aggregates"]["prediction_volume"] == { "model_output_box_count": 4, "evaluated_box_count": 3, "outside_shared_range_count": 1, } def test_frame_set_must_equal_admitted_validation_split() -> None: truth, predictions = _perfect_fixture() with pytest.raises( KittiPointPillarsBenchmarkError, match="do not equal", ): evaluate_pointpillars_predictions( truth_by_frame=truth, predictions=predictions[:-1], ) def test_unknown_model_class_is_rejected() -> None: truth, predictions = _perfect_fixture() first = predictions[0] unknown = PointPillarsBox( **{ field: getattr(first.boxes[0], field) for field in ( "x_m", "y_m", "z_m", "length_m", "width_m", "height_m", "yaw_rad", "class_id", "score", ) }, model_class="Unknown", ) predictions = ( PointPillarsFramePrediction( frame_id=first.frame_id, boxes=(unknown,), inference_ms=first.inference_ms, ), *predictions[1:], ) with pytest.raises(KittiPointPillarsBenchmarkError, match="not admitted"): evaluate_pointpillars_predictions( truth_by_frame=truth, predictions=predictions, )