from __future__ import annotations import hashlib import json import zipfile from pathlib import Path from typing import Any import pytest from k1link.datasets import kitti_3d_admission as module def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _calibration(*, complete: bool = True) -> str: rows = [ "P0: " + " ".join(["1"] * 12), "P1: " + " ".join(["1"] * 12), "P2: " + " ".join(["1"] * 12), "P3: " + " ".join(["1"] * 12), "R0_rect: " + " ".join(["1"] * 9), "Tr_velo_to_cam: " + " ".join(["1"] * 12), ] if complete: rows.append("Tr_imu_to_velo: " + " ".join(["1"] * 12)) return "\n".join(rows) + "\n" def _label(class_name: str, *, valid: bool = True) -> str: dimensions = "1.5 1.6 3.8" if valid else "0 1.6 3.8" return ( f"{class_name} 0 0 0 0 0 10 10 {dimensions} 1 1 10 0\n" ) def _release( root: Path, monkeypatch: pytest.MonkeyPatch, *, labels: dict[str, str] | None = None, complete_calibration: bool = True, invalid_velodyne_frame: bool = False, train_rows: tuple[str, ...] = ("000000",), validation_rows: tuple[str, ...] = ("000001", "000002"), ) -> dict[str, Path]: archive_root = root / module.KITTI_3D_RELEASE_ROOT / "archives" split_root = root / module.KITTI_3D_RELEASE_ROOT / "splits" / ( f"openpcdet-{module.KITTI_STANDARD_SPLIT_COMMIT}" ) archive_root.mkdir(parents=True) split_root.mkdir(parents=True) paths = { module.KITTI_VELODYNE_ARCHIVE: archive_root / module.KITTI_VELODYNE_ARCHIVE, module.KITTI_LABEL_ARCHIVE: archive_root / module.KITTI_LABEL_ARCHIVE, module.KITTI_CALIB_ARCHIVE: archive_root / module.KITTI_CALIB_ARCHIVE, "train": split_root / "train.txt", "validation": split_root / "val.txt", } with zipfile.ZipFile(paths[module.KITTI_VELODYNE_ARCHIVE], "w") as archive: for frame_id in ("000000", "000001", "000002"): size = 15 if invalid_velodyne_frame and frame_id == "000001" else 16 archive.writestr(f"training/velodyne/{frame_id}.bin", b"\x00" * size) for frame_id in ("000000", "000001"): archive.writestr(f"testing/velodyne/{frame_id}.bin", b"\x00" * 16) label_payloads = labels or { "000000": _label("Car"), "000001": _label("Pedestrian") + _label("Car"), "000002": _label("Cyclist"), } with zipfile.ZipFile(paths[module.KITTI_LABEL_ARCHIVE], "w") as archive: for frame_id, payload in label_payloads.items(): archive.writestr(f"training/label_2/{frame_id}.txt", payload) with zipfile.ZipFile(paths[module.KITTI_CALIB_ARCHIVE], "w") as archive: for split, frames in { "training": ("000000", "000001", "000002"), "testing": ("000000", "000001"), }.items(): for frame_id in frames: archive.writestr( f"{split}/calib/{frame_id}.txt", _calibration(complete=complete_calibration), ) paths["train"].write_text("\n".join(train_rows) + "\n", encoding="ascii") paths["validation"].write_text( "\n".join(validation_rows) + "\n", encoding="ascii", ) monkeypatch.setattr(module, "KITTI_TRAINING_FRAMES", 3) monkeypatch.setattr(module, "KITTI_TEST_FRAMES", 2) monkeypatch.setattr( module, "KITTI_ARCHIVE_BYTES", { name: paths[name].stat().st_size for name in ( module.KITTI_VELODYNE_ARCHIVE, module.KITTI_LABEL_ARCHIVE, module.KITTI_CALIB_ARCHIVE, ) }, ) monkeypatch.setattr( module, "KITTI_SPLIT_COUNTS", {"train": len(train_rows), "validation": len(validation_rows)}, ) monkeypatch.setattr( module, "KITTI_SPLIT_SHA256", {"train": _sha256(paths["train"]), "validation": _sha256(paths["validation"])}, ) monkeypatch.setattr(module, "_is_worker_dataset_root", lambda _root: True) return paths def test_admits_archive_only_box_truth_with_path_free_state( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _release(tmp_path, monkeypatch) manifest = module.admit_kitti_3d_object_release(tmp_path) assert manifest["status"] == "archive-ready" assert manifest["storage"]["source_archives_extracted"] is False assert manifest["benchmark_contract"] == { "independent_ground_truth": True, "annotations": ["oriented-3d-boxes"], "point_fields": ["x", "y", "z", "intensity"], "eligible_split": "validation", "target_classes": ["Car", "Pedestrian", "Cyclist"], "official_test_submission_authorized": False, "retuning_on_validation_allowed": False, "k1_quality_claim_authorized": False, } assert manifest["alignment"]["validation_target_box_counts"] == { "Car": 1, "Cyclist": 1, "Pedestrian": 1, } serialized = json.dumps(manifest) assert str(tmp_path) not in serialized assert module.read_kitti_3d_admission(tmp_path) == manifest assert module.read_kitti_standard_splits(tmp_path) == { "train": ("000000",), "validation": ("000001", "000002"), } def test_rejects_tampered_standard_split( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: paths = _release(tmp_path, monkeypatch) paths["validation"].write_text("000002\n000001\n", encoding="ascii") with pytest.raises(module.Kitti3DAdmissionError, match="pinned OpenPCDet"): module.admit_kitti_3d_object_release(tmp_path) def test_rejects_overlapping_split( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _release( tmp_path, monkeypatch, train_rows=("000000", "000001"), validation_rows=("000001", "000002"), ) with pytest.raises(module.Kitti3DAdmissionError, match="overlapping or incomplete"): module.admit_kitti_3d_object_release(tmp_path) def test_rejects_non_xyzi_velodyne_frame( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _release(tmp_path, monkeypatch, invalid_velodyne_frame=True) with pytest.raises(module.Kitti3DAdmissionError, match="not packed XYZI"): module.admit_kitti_3d_object_release(tmp_path) def test_rejects_invalid_target_box( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _release( tmp_path, monkeypatch, labels={ "000000": _label("Car"), "000001": _label("Pedestrian", valid=False), "000002": _label("Cyclist"), }, ) with pytest.raises(module.Kitti3DAdmissionError, match="invalid dimensions"): module.admit_kitti_3d_object_release(tmp_path) def test_rejects_missing_calibration_transform( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _release(tmp_path, monkeypatch, complete_calibration=False) with pytest.raises(module.Kitti3DAdmissionError, match="required transforms"): module.admit_kitti_3d_object_release(tmp_path) def test_read_rejects_tampered_content_identity( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _release(tmp_path, monkeypatch) module.admit_kitti_3d_object_release(tmp_path) state = tmp_path / "state/kitti-3d-object-v2017.json" payload: dict[str, Any] = json.loads(state.read_text(encoding="utf-8")) payload["identity"]["license"]["spdx"] = "unknown" state.write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(module.Kitti3DAdmissionError, match="identity is invalid"): module.read_kitti_3d_admission(tmp_path)