evaluate_pointcloud_dataset.py

"""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())