feat(perception): add PointPillars visual audit
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,664 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Build a bounded visual-audit derivative of a sealed L3 transfer run."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import tempfile
|
||||
import zipfile
|
||||
from contextlib import suppress
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Final
|
||||
|
||||
import numpy as np
|
||||
|
||||
from k1link.compute.kitti_pointpillars_benchmark import (
|
||||
CROSS_DOMAIN_EVALUATION_RANGE,
|
||||
KITTI_BENCHMARK_CLASSES,
|
||||
KITTI_IOU_THRESHOLDS,
|
||||
MODEL_TO_KITTI_CLASS,
|
||||
KittiLidarTruth,
|
||||
read_kitti_validation_truth,
|
||||
)
|
||||
from k1link.compute.pointpillars_postprocess import (
|
||||
PointPillarsBox,
|
||||
oriented_3d_iou,
|
||||
)
|
||||
from k1link.datasets.kitti_3d_admission import (
|
||||
KITTI_3D_RELEASE_ROOT,
|
||||
KITTI_CALIB_ARCHIVE,
|
||||
KITTI_LABEL_ARCHIVE,
|
||||
KITTI_VELODYNE_ARCHIVE,
|
||||
read_kitti_3d_admission,
|
||||
read_kitti_standard_splits,
|
||||
)
|
||||
|
||||
SOURCE_MANIFEST_SCHEMA: Final = (
|
||||
"missioncore.l3-pointpillars-kitti-transfer-result/v1"
|
||||
)
|
||||
SOURCE_FRAME_SCHEMA: Final = (
|
||||
"missioncore.l3-pointpillars-kitti-transfer-frame/v1"
|
||||
)
|
||||
VISUAL_AUDIT_SCHEMA: Final = "missioncore.l3-pointpillars-visual-audit/v1"
|
||||
VISUAL_CATALOG_SCHEMA: Final = (
|
||||
"missioncore.l3-pointpillars-visual-audit-catalog/v1"
|
||||
)
|
||||
VISUAL_FRAME_SCHEMA: Final = "missioncore.l3-pointpillars-visual-frame/v1"
|
||||
EXPECTED_SOURCE_RUN_ID: Final = (
|
||||
"l3-pointpillars-kitti-"
|
||||
"1a6b499e194a363644854dc324bd1b565c100b809f145c1324a25328e7ae0910"
|
||||
)
|
||||
EXPECTED_FRAME_RESULTS_IDENTITY: Final = (
|
||||
"30b1933d508a09025a7d3c3c460fc2d06128e4bbe96a753bec7ba8545fda3e9c"
|
||||
)
|
||||
MAX_SELECTED_FRAMES: Final = 18
|
||||
MAX_SAMPLED_POINTS: Final = 12_000
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--source-run", type=Path, required=True)
|
||||
parser.add_argument("--dataset-root", type=Path, required=True)
|
||||
parser.add_argument("--output-root", type=Path, required=True)
|
||||
args = parser.parse_args()
|
||||
result = build_visual_audit(
|
||||
source_run=args.source_run,
|
||||
dataset_root=args.dataset_root,
|
||||
output_root=args.output_root,
|
||||
)
|
||||
print(json.dumps(result, sort_keys=True), flush=True)
|
||||
return 0
|
||||
|
||||
|
||||
def build_visual_audit(
|
||||
*,
|
||||
source_run: Path,
|
||||
dataset_root: Path,
|
||||
output_root: Path,
|
||||
) -> dict[str, object]:
|
||||
run_root = source_run.expanduser().resolve(strict=True)
|
||||
if not run_root.is_dir() or run_root.is_symlink():
|
||||
raise RuntimeError("sealed L3 source run is unavailable")
|
||||
source_manifest_path = run_root / "manifest.json"
|
||||
source_report_path = run_root / "report.json"
|
||||
source_manifest = _read_json(source_manifest_path)
|
||||
source_report = _read_json(source_report_path)
|
||||
_validate_source_manifest(
|
||||
source_manifest,
|
||||
source_report,
|
||||
source_manifest_path=source_manifest_path,
|
||||
source_report_path=source_report_path,
|
||||
run_root=run_root,
|
||||
)
|
||||
|
||||
dataset = read_kitti_3d_admission(dataset_root.expanduser().absolute())
|
||||
validation_ids = read_kitti_standard_splits(
|
||||
dataset_root.expanduser().absolute()
|
||||
)["validation"]
|
||||
if (
|
||||
dataset["release_identity_sha256"]
|
||||
!= source_manifest["identity"]["dataset_release_identity_sha256"]
|
||||
or len(validation_ids) != source_manifest["frame_result_count"]
|
||||
):
|
||||
raise RuntimeError("sealed run and admitted KITTI release diverge")
|
||||
|
||||
archive_root = (
|
||||
dataset_root.expanduser().absolute()
|
||||
/ KITTI_3D_RELEASE_ROOT
|
||||
/ "archives"
|
||||
)
|
||||
truths = read_kitti_validation_truth(
|
||||
labels_archive=archive_root / KITTI_LABEL_ARCHIVE,
|
||||
calibrations_archive=archive_root / KITTI_CALIB_ARCHIVE,
|
||||
validation_frame_ids=validation_ids,
|
||||
)
|
||||
predictions = _read_predictions(run_root / "frames", validation_ids)
|
||||
matched_predictions, matched_truth = _global_matches(predictions, truths)
|
||||
summaries = _frame_summaries(
|
||||
predictions,
|
||||
truths,
|
||||
matched_predictions,
|
||||
matched_truth,
|
||||
)
|
||||
selected_ids = _select_frames(summaries)
|
||||
|
||||
identity = {
|
||||
"source_run_id": source_manifest["run_id"],
|
||||
"source_manifest_sha256": _sha256(source_manifest_path),
|
||||
"source_frame_results_identity_sha256": source_manifest[
|
||||
"frame_results_identity_sha256"
|
||||
],
|
||||
"dataset_source_id": dataset["source_id"],
|
||||
"dataset_release_identity_sha256": dataset[
|
||||
"release_identity_sha256"
|
||||
],
|
||||
"matching": {
|
||||
"metric": "oriented-3d-iou",
|
||||
"ordering": "global-score-descending",
|
||||
"class_iou_thresholds": KITTI_IOU_THRESHOLDS,
|
||||
"shared_evaluation_range": list(CROSS_DOMAIN_EVALUATION_RANGE),
|
||||
},
|
||||
"selection": {
|
||||
"policy": "tp-first-then-fp-fn-class-coverage/v1",
|
||||
"maximum_frames": MAX_SELECTED_FRAMES,
|
||||
"selected_frame_ids": list(selected_ids),
|
||||
},
|
||||
"point_sampling": {
|
||||
"policy": "shared-range-even-index/v1",
|
||||
"maximum_points_per_frame": MAX_SAMPLED_POINTS,
|
||||
"fields": ["x_m", "y_m", "z_m", "intensity"],
|
||||
},
|
||||
"producer_sha256": _sha256(Path(__file__).resolve(strict=True)),
|
||||
"authority": {
|
||||
"read_only": True,
|
||||
"commands_enabled": False,
|
||||
"navigation_or_safety_accepted": False,
|
||||
},
|
||||
}
|
||||
identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
|
||||
result_id = f"l3-pointpillars-visual-audit-{identity_sha256}"
|
||||
result_root = output_root.expanduser().absolute() / result_id
|
||||
if result_root.exists():
|
||||
raise RuntimeError("visual-audit derivative already exists")
|
||||
frames_root = result_root / "frames"
|
||||
frames_root.mkdir(mode=0o700, parents=True)
|
||||
|
||||
descriptors: list[dict[str, object]] = []
|
||||
points_path = archive_root / KITTI_VELODYNE_ARCHIVE
|
||||
try:
|
||||
with zipfile.ZipFile(points_path.resolve(strict=True)) as points_zip:
|
||||
for frame_id in selected_ids:
|
||||
raw = points_zip.read(f"training/velodyne/{frame_id}.bin")
|
||||
source_frame = predictions[frame_id]["payload"]
|
||||
if hashlib.sha256(raw).hexdigest() != source_frame["point_sha256"]:
|
||||
raise RuntimeError(
|
||||
f"KITTI points changed for selected frame {frame_id}"
|
||||
)
|
||||
detail = _frame_detail(
|
||||
frame_id=frame_id,
|
||||
point_bytes=raw,
|
||||
prediction=predictions[frame_id],
|
||||
truths=truths[frame_id],
|
||||
matched_predictions=matched_predictions,
|
||||
matched_truth=matched_truth,
|
||||
summary=summaries[frame_id],
|
||||
)
|
||||
path = frames_root / f"{frame_id}.json"
|
||||
_write_once(path, detail)
|
||||
descriptors.append(
|
||||
{
|
||||
**summaries[frame_id],
|
||||
"detail_path": f"frames/{frame_id}.json",
|
||||
"detail_sha256": _sha256(path),
|
||||
"detail_byte_length": path.stat().st_size,
|
||||
}
|
||||
)
|
||||
except (OSError, KeyError, zipfile.BadZipFile) as exc:
|
||||
raise RuntimeError("selected KITTI points could not be read") from exc
|
||||
|
||||
catalog = {
|
||||
"schema_version": VISUAL_CATALOG_SCHEMA,
|
||||
"result_id": result_id,
|
||||
"source_run_id": source_manifest["run_id"],
|
||||
"frame_count": len(descriptors),
|
||||
"frames": descriptors,
|
||||
}
|
||||
catalog_path = result_root / "catalog.json"
|
||||
_write_once(catalog_path, catalog)
|
||||
manifest = {
|
||||
"schema_version": VISUAL_AUDIT_SCHEMA,
|
||||
"result_id": result_id,
|
||||
"identity_sha256": identity_sha256,
|
||||
"identity": identity,
|
||||
"created_at_utc": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
|
||||
"status": "operator-visual-review-required",
|
||||
"source_metrics": source_report["metrics"],
|
||||
"catalog": _artifact(catalog_path, "visual-frame-catalog"),
|
||||
"authority": identity["authority"],
|
||||
}
|
||||
_write_once(result_root / "manifest.json", manifest)
|
||||
return {
|
||||
"result_id": result_id,
|
||||
"selected_frame_count": len(descriptors),
|
||||
"status": manifest["status"],
|
||||
}
|
||||
|
||||
|
||||
def _validate_source_manifest(
|
||||
manifest: dict[str, Any],
|
||||
report: dict[str, Any],
|
||||
*,
|
||||
source_manifest_path: Path,
|
||||
source_report_path: Path,
|
||||
run_root: Path,
|
||||
) -> None:
|
||||
artifacts = manifest.get("artifacts")
|
||||
if (
|
||||
manifest.get("schema_version") != SOURCE_MANIFEST_SCHEMA
|
||||
or manifest.get("run_id") != EXPECTED_SOURCE_RUN_ID
|
||||
or run_root.name != EXPECTED_SOURCE_RUN_ID
|
||||
or manifest.get("frame_result_count") != 3769
|
||||
or manifest.get("frame_results_identity_sha256")
|
||||
!= EXPECTED_FRAME_RESULTS_IDENTITY
|
||||
or manifest.get("status")
|
||||
!= "public-cross-domain-transfer-probe-measured"
|
||||
or report.get("run_id") != manifest.get("run_id")
|
||||
or report.get("status") != manifest.get("status")
|
||||
or not isinstance(artifacts, list)
|
||||
):
|
||||
raise RuntimeError("sealed L3 source manifest is invalid")
|
||||
observed = {
|
||||
descriptor.get("role"): descriptor
|
||||
for descriptor in artifacts
|
||||
if isinstance(descriptor, dict)
|
||||
}
|
||||
for role, path in (
|
||||
("run-identity", run_root / "identity.json"),
|
||||
("benchmark-report", source_report_path),
|
||||
):
|
||||
descriptor = observed.get(role)
|
||||
if (
|
||||
not isinstance(descriptor, dict)
|
||||
or descriptor.get("sha256") != _sha256(path)
|
||||
or descriptor.get("byte_length") != path.stat().st_size
|
||||
):
|
||||
raise RuntimeError(f"sealed L3 {role} changed")
|
||||
if _frame_results_identity(run_root / "frames") != EXPECTED_FRAME_RESULTS_IDENTITY:
|
||||
raise RuntimeError("sealed L3 frame set changed")
|
||||
if _sha256(source_manifest_path) != _sha256(run_root / "manifest.json"):
|
||||
raise RuntimeError("sealed L3 manifest path changed")
|
||||
|
||||
|
||||
def _read_predictions(
|
||||
frames_root: Path,
|
||||
validation_ids: tuple[str, ...],
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
expected = {f"{frame_id}.json" for frame_id in validation_ids}
|
||||
actual = {path.name for path in frames_root.glob("*.json")}
|
||||
if actual != expected:
|
||||
raise RuntimeError("sealed L3 frame set does not match KITTI validation")
|
||||
predictions: dict[str, dict[str, Any]] = {}
|
||||
for frame_id in validation_ids:
|
||||
payload = _read_json(frames_root / f"{frame_id}.json")
|
||||
raw_boxes = payload.get("boxes")
|
||||
if (
|
||||
payload.get("schema_version") != SOURCE_FRAME_SCHEMA
|
||||
or payload.get("frame_id") != frame_id
|
||||
or not isinstance(payload.get("point_sha256"), str)
|
||||
or not isinstance(raw_boxes, list)
|
||||
):
|
||||
raise RuntimeError(f"sealed L3 frame {frame_id} is invalid")
|
||||
try:
|
||||
boxes = tuple(PointPillarsBox(**box) for box in raw_boxes)
|
||||
inference_ms = float(payload["inference_ms"])
|
||||
except (KeyError, TypeError, ValueError) as exc:
|
||||
raise RuntimeError(
|
||||
f"sealed L3 frame {frame_id} is invalid"
|
||||
) from exc
|
||||
if (
|
||||
not math.isfinite(inference_ms)
|
||||
or inference_ms <= 0
|
||||
or any(not _valid_box(box) for box in boxes)
|
||||
):
|
||||
raise RuntimeError(f"sealed L3 frame {frame_id} is invalid")
|
||||
predictions[frame_id] = {
|
||||
"payload": payload,
|
||||
"boxes": boxes,
|
||||
"inference_ms": inference_ms,
|
||||
}
|
||||
return predictions
|
||||
|
||||
|
||||
def _global_matches(
|
||||
predictions: dict[str, dict[str, Any]],
|
||||
truths: dict[str, tuple[KittiLidarTruth, ...]],
|
||||
) -> tuple[dict[tuple[str, int], tuple[int, float]], set[tuple[str, int]]]:
|
||||
matched_predictions: dict[tuple[str, int], tuple[int, float]] = {}
|
||||
matched_truth: set[tuple[str, int]] = set()
|
||||
for class_name in KITTI_BENCHMARK_CLASSES:
|
||||
ordered = sorted(
|
||||
(
|
||||
(box.score, frame_id, index, box)
|
||||
for frame_id, frame in predictions.items()
|
||||
for index, box in enumerate(frame["boxes"])
|
||||
if _inside_shared_range(box)
|
||||
and MODEL_TO_KITTI_CLASS[box.model_class] == class_name
|
||||
),
|
||||
key=lambda item: (-item[0], item[1], item[2]),
|
||||
)
|
||||
for _, frame_id, index, box in ordered:
|
||||
best_index = -1
|
||||
best_iou = -1.0
|
||||
for truth_index, truth in enumerate(truths[frame_id]):
|
||||
if (
|
||||
truth.benchmark_class != class_name
|
||||
or (frame_id, truth_index) in matched_truth
|
||||
):
|
||||
continue
|
||||
overlap = oriented_3d_iou(box, _truth_box(truth))
|
||||
if overlap > best_iou:
|
||||
best_index = truth_index
|
||||
best_iou = overlap
|
||||
if (
|
||||
best_index >= 0
|
||||
and best_iou >= KITTI_IOU_THRESHOLDS[class_name]
|
||||
):
|
||||
matched_truth.add((frame_id, best_index))
|
||||
matched_predictions[(frame_id, index)] = (
|
||||
best_index,
|
||||
best_iou,
|
||||
)
|
||||
return matched_predictions, matched_truth
|
||||
|
||||
|
||||
def _frame_summaries(
|
||||
predictions: dict[str, dict[str, Any]],
|
||||
truths: dict[str, tuple[KittiLidarTruth, ...]],
|
||||
matched_predictions: dict[tuple[str, int], tuple[int, float]],
|
||||
matched_truth: set[tuple[str, int]],
|
||||
) -> dict[str, dict[str, object]]:
|
||||
result: dict[str, dict[str, object]] = {}
|
||||
for frame_id, frame in predictions.items():
|
||||
evaluated = [
|
||||
(index, box)
|
||||
for index, box in enumerate(frame["boxes"])
|
||||
if _inside_shared_range(box)
|
||||
]
|
||||
true_positive_count = sum(
|
||||
(frame_id, index) in matched_predictions
|
||||
for index, _ in evaluated
|
||||
)
|
||||
false_negative_count = sum(
|
||||
(frame_id, index) not in matched_truth
|
||||
for index in range(len(truths[frame_id]))
|
||||
)
|
||||
result[frame_id] = {
|
||||
"frame_id": frame_id,
|
||||
"inference_ms": frame["inference_ms"],
|
||||
"prediction_count": len(frame["boxes"]),
|
||||
"evaluated_prediction_count": len(evaluated),
|
||||
"outside_shared_range_count": len(frame["boxes"]) - len(evaluated),
|
||||
"truth_count": len(truths[frame_id]),
|
||||
"true_positive_count": true_positive_count,
|
||||
"false_positive_count": len(evaluated) - true_positive_count,
|
||||
"false_negative_count": false_negative_count,
|
||||
"truth_classes": sorted(
|
||||
{truth.benchmark_class for truth in truths[frame_id]}
|
||||
),
|
||||
}
|
||||
return result
|
||||
|
||||
|
||||
def _select_frames(
|
||||
summaries: dict[str, dict[str, object]],
|
||||
) -> tuple[str, ...]:
|
||||
selected: list[str] = []
|
||||
|
||||
def add(frame_id: str) -> None:
|
||||
if frame_id not in selected and len(selected) < MAX_SELECTED_FRAMES:
|
||||
selected.append(frame_id)
|
||||
|
||||
for frame_id in sorted(
|
||||
summaries,
|
||||
key=lambda item: (
|
||||
-int(summaries[item]["true_positive_count"]),
|
||||
item,
|
||||
),
|
||||
):
|
||||
if int(summaries[frame_id]["true_positive_count"]) > 0:
|
||||
add(frame_id)
|
||||
for metric in ("false_positive_count", "false_negative_count"):
|
||||
for frame_id in sorted(
|
||||
summaries,
|
||||
key=lambda item: (-int(summaries[item][metric]), item),
|
||||
)[:6]:
|
||||
add(frame_id)
|
||||
for class_name in KITTI_BENCHMARK_CLASSES:
|
||||
candidates = [
|
||||
frame_id
|
||||
for frame_id, summary in summaries.items()
|
||||
if class_name in summary["truth_classes"]
|
||||
]
|
||||
if candidates:
|
||||
add(
|
||||
max(
|
||||
candidates,
|
||||
key=lambda item: (
|
||||
int(summaries[item]["false_negative_count"]),
|
||||
int(summaries[item]["false_positive_count"]),
|
||||
item,
|
||||
),
|
||||
)
|
||||
)
|
||||
for frame_id in sorted(
|
||||
summaries,
|
||||
key=lambda item: (
|
||||
-int(summaries[item]["false_positive_count"])
|
||||
- int(summaries[item]["false_negative_count"]),
|
||||
item,
|
||||
),
|
||||
):
|
||||
add(frame_id)
|
||||
if not selected:
|
||||
raise RuntimeError("visual-audit selection is empty")
|
||||
return tuple(selected)
|
||||
|
||||
|
||||
def _frame_detail(
|
||||
*,
|
||||
frame_id: str,
|
||||
point_bytes: bytes,
|
||||
prediction: dict[str, Any],
|
||||
truths: tuple[KittiLidarTruth, ...],
|
||||
matched_predictions: dict[tuple[str, int], tuple[int, float]],
|
||||
matched_truth: set[tuple[str, int]],
|
||||
summary: dict[str, object],
|
||||
) -> dict[str, object]:
|
||||
points = np.frombuffer(point_bytes, dtype="<f4")
|
||||
if points.size % 4:
|
||||
raise RuntimeError(f"KITTI point frame {frame_id} is malformed")
|
||||
points = points.reshape(-1, 4)
|
||||
bounds = CROSS_DOMAIN_EVALUATION_RANGE
|
||||
mask = (
|
||||
(points[:, 0] >= bounds[0])
|
||||
& (points[:, 0] <= bounds[3])
|
||||
& (points[:, 1] >= bounds[1])
|
||||
& (points[:, 1] <= bounds[4])
|
||||
& (points[:, 2] >= bounds[2])
|
||||
& (points[:, 2] <= bounds[5])
|
||||
)
|
||||
bounded = points[mask]
|
||||
if len(bounded) > MAX_SAMPLED_POINTS:
|
||||
indices = np.linspace(
|
||||
0,
|
||||
len(bounded) - 1,
|
||||
MAX_SAMPLED_POINTS,
|
||||
dtype=np.int64,
|
||||
)
|
||||
sampled = bounded[indices]
|
||||
else:
|
||||
sampled = bounded
|
||||
flat_points = np.round(sampled, decimals=4).reshape(-1).tolist()
|
||||
|
||||
prediction_boxes: list[dict[str, object]] = []
|
||||
for index, box in enumerate(prediction["boxes"]):
|
||||
if not _inside_shared_range(box):
|
||||
continue
|
||||
match = matched_predictions.get((frame_id, index))
|
||||
prediction_boxes.append(
|
||||
{
|
||||
**_box_payload(box, MODEL_TO_KITTI_CLASS[box.model_class]),
|
||||
"score": box.score,
|
||||
"status": "true-positive" if match else "false-positive",
|
||||
"matched_truth_index": match[0] if match else None,
|
||||
"matched_iou_3d": match[1] if match else None,
|
||||
}
|
||||
)
|
||||
truth_boxes = [
|
||||
{
|
||||
**_truth_payload(truth),
|
||||
"truth_index": index,
|
||||
"status": (
|
||||
"matched" if (frame_id, index) in matched_truth
|
||||
else "false-negative"
|
||||
),
|
||||
}
|
||||
for index, truth in enumerate(truths)
|
||||
]
|
||||
return {
|
||||
"schema_version": VISUAL_FRAME_SCHEMA,
|
||||
"frame_id": frame_id,
|
||||
"summary": summary,
|
||||
"points": {
|
||||
"layout": "flat-xyzi",
|
||||
"source_point_count": len(points),
|
||||
"shared_range_point_count": len(bounded),
|
||||
"sampled_point_count": len(sampled),
|
||||
"values": flat_points,
|
||||
},
|
||||
"truth_boxes": truth_boxes,
|
||||
"prediction_boxes": prediction_boxes,
|
||||
}
|
||||
|
||||
|
||||
def _valid_box(box: PointPillarsBox) -> bool:
|
||||
values = (
|
||||
box.x_m,
|
||||
box.y_m,
|
||||
box.z_m,
|
||||
box.length_m,
|
||||
box.width_m,
|
||||
box.height_m,
|
||||
box.yaw_rad,
|
||||
box.score,
|
||||
)
|
||||
return (
|
||||
box.model_class in MODEL_TO_KITTI_CLASS
|
||||
and all(math.isfinite(value) for value in values)
|
||||
and min(box.length_m, box.width_m, box.height_m) > 0
|
||||
and 0 <= box.score <= 1
|
||||
)
|
||||
|
||||
|
||||
def _inside_shared_range(box: PointPillarsBox) -> bool:
|
||||
bounds = CROSS_DOMAIN_EVALUATION_RANGE
|
||||
return (
|
||||
bounds[0] <= box.x_m <= bounds[3]
|
||||
and bounds[1] <= box.y_m <= bounds[4]
|
||||
and bounds[2] <= box.z_m <= bounds[5]
|
||||
)
|
||||
|
||||
|
||||
def _truth_box(truth: KittiLidarTruth) -> PointPillarsBox:
|
||||
return PointPillarsBox(
|
||||
x_m=truth.x_m,
|
||||
y_m=truth.y_m,
|
||||
z_m=truth.z_m,
|
||||
length_m=truth.length_m,
|
||||
width_m=truth.width_m,
|
||||
height_m=truth.height_m,
|
||||
yaw_rad=truth.yaw_rad,
|
||||
class_id=-1,
|
||||
model_class=truth.benchmark_class,
|
||||
score=1.0,
|
||||
)
|
||||
|
||||
|
||||
def _box_payload(
|
||||
box: PointPillarsBox,
|
||||
benchmark_class: str,
|
||||
) -> dict[str, object]:
|
||||
return {
|
||||
"benchmark_class": benchmark_class,
|
||||
"center_xyz_m": [box.x_m, box.y_m, box.z_m],
|
||||
"size_lwh_m": [box.length_m, box.width_m, box.height_m],
|
||||
"yaw_rad": box.yaw_rad,
|
||||
}
|
||||
|
||||
|
||||
def _truth_payload(truth: KittiLidarTruth) -> dict[str, object]:
|
||||
return {
|
||||
"benchmark_class": truth.benchmark_class,
|
||||
"center_xyz_m": [truth.x_m, truth.y_m, truth.z_m],
|
||||
"size_lwh_m": [truth.length_m, truth.width_m, truth.height_m],
|
||||
"yaw_rad": truth.yaw_rad,
|
||||
}
|
||||
|
||||
|
||||
def _frame_results_identity(frames_root: Path) -> str:
|
||||
descriptors = [
|
||||
{
|
||||
"name": path.name,
|
||||
"sha256": _sha256(path),
|
||||
"byte_length": path.stat().st_size,
|
||||
}
|
||||
for path in sorted(frames_root.glob("*.json"))
|
||||
]
|
||||
return hashlib.sha256(_canonical_json(descriptors)).hexdigest()
|
||||
|
||||
|
||||
def _artifact(path: Path, role: str) -> dict[str, object]:
|
||||
return {
|
||||
"path": path.name,
|
||||
"role": role,
|
||||
"media_type": "application/json",
|
||||
"sha256": _sha256(path),
|
||||
"byte_length": path.stat().st_size,
|
||||
}
|
||||
|
||||
|
||||
def _read_json(path: Path) -> dict[str, Any]:
|
||||
try:
|
||||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
raise RuntimeError(f"{path.name} is invalid") from exc
|
||||
if not isinstance(payload, dict):
|
||||
raise RuntimeError(f"{path.name} is not an object")
|
||||
return payload
|
||||
|
||||
|
||||
def _sha256(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as source:
|
||||
for chunk in iter(lambda: source.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _canonical_json(payload: Any) -> bytes:
|
||||
return json.dumps(
|
||||
payload,
|
||||
ensure_ascii=False,
|
||||
separators=(",", ":"),
|
||||
sort_keys=True,
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
def _write_once(path: Path, payload: dict[str, Any]) -> None:
|
||||
path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||||
if path.exists():
|
||||
raise RuntimeError(f"{path.name} already exists")
|
||||
descriptor, temporary = tempfile.mkstemp(
|
||||
prefix=f".{path.name}.",
|
||||
suffix=".tmp",
|
||||
dir=path.parent,
|
||||
)
|
||||
try:
|
||||
with os.fdopen(descriptor, "wb") as target:
|
||||
target.write(_canonical_json(payload))
|
||||
target.flush()
|
||||
os.fsync(target.fileno())
|
||||
os.chmod(temporary, 0o600)
|
||||
os.replace(temporary, path)
|
||||
finally:
|
||||
with suppress(FileNotFoundError):
|
||||
os.unlink(temporary)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user