evaluate_depth_dataset.py

"""Evaluate a JSONL manifest of paired Simam3D depth arrays."""

from __future__ import annotations

import argparse
import json
from pathlib import Path

import numpy as np

from evaluate_depth import evaluate_depth_pair


def evaluate_manifest(manifest_path: str | Path) -> dict[str, object]:
    """Evaluate every manifest row and aggregate successful metric values."""
    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_path = manifest_path.parent / str(record["prediction"])
            ground_truth_path = manifest_path.parent / str(record["ground_truth"])
            mask_path = record.get("mask")
            mask = np.load(manifest_path.parent / str(mask_path)) if mask_path else None
            report = evaluate_depth_pair(
                np.load(prediction_path),
                np.load(ground_truth_path),
                mask,
                bool(record.get("inverse_depth", False)),
            )
            rows.append({"id": record.get("id", f"line-{line_number}"), **report})
        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 evaluation rows")
    metric_names = ("abs_rel", "sq_rel", "rmse", "rmse_log", "delta1", "delta2", "delta3")
    aggregate = {
        name: {
            "mean": float(np.mean([row["median_scaled"][name] for row in rows])),
            "median": float(np.median([row["median_scaled"][name] for row in rows])),
        }
        for name in metric_names
    }
    return {
        "project": "Simam3D",
        "manifest": str(manifest_path),
        "successful_rows": len(rows),
        "failed_rows": failures,
        "aggregate_median_scaled": aggregate,
        "rows": rows,
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("manifest", type=Path, help="JSONL depth-pair manifest")
    args = parser.parse_args(argv)
    print(json.dumps(evaluate_manifest(args.manifest), indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())