"""Evaluate a JSONL manifest of exported and reference point clouds.""" from __future__ import annotations import argparse import json from pathlib import Path from evaluate_pointcloud import load_points, pointcloud_metrics def evaluate_manifest(manifest_path: str | Path) -> dict[str, object]: """Evaluate each point-cloud pair and preserve failures for auditability.""" manifest_path = Path(manifest_path) rows: list[dict[str, object]] = [] failures: list[dict[str, str]] = [] for line_number, raw_line in enumerate(manifest_path.read_text(encoding="utf-8").splitlines(), 1): if not raw_line.strip() or raw_line.lstrip().startswith("#"): continue try: record = json.loads(raw_line) prediction = load_points(manifest_path.parent / str(record["prediction"])) reference = load_points(manifest_path.parent / str(record["reference"])) threshold = float(record.get("threshold", 0.05)) rows.append({ "id": record.get("id", f"line-{line_number}"), "metrics": pointcloud_metrics(prediction, reference, threshold), }) except (KeyError, OSError, ValueError, json.JSONDecodeError) as exc: failures.append({"line": str(line_number), "error": f"{type(exc).__name__}: {exc}"}) if not rows: raise ValueError("manifest produced no successful point-cloud rows") metric_names = ("chamfer_l1", "precision", "recall", "fscore") aggregate = { name: { "mean": sum(float(row["metrics"][name]) for row in rows) / len(rows), "median": sorted(float(row["metrics"][name]) for row in rows)[len(rows) // 2], } for name in metric_names } return { "project": "Simam3D", "manifest": str(manifest_path), "successful_rows": len(rows), "failed_rows": failures, "aggregate": aggregate, "rows": rows, } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("manifest", type=Path, help="JSONL point-cloud manifest") args = parser.parse_args(argv) print(json.dumps(evaluate_manifest(args.manifest), indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())