Files
NODEDC_MISSION_CORE/tests/test_kitti_pointpillars_benchmark.py

259 lines
7.2 KiB
Python

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,
)