evaluation/eval_camera/eval.py

"""Evaluate camera-motion adherence of generated videos against a camera reference clip.

Both videos are reduced to a 3D camera trajectory (``traj.py``), canonicalized to frame 0 and
scale-normalized, then compared (``metric.py``):

    rot_err    mean geodesic angle between the two rotation sequences, degrees   (lower better)
    trans_err  mean L2 between scale-normalized camera centres                   (lower better)
    cam_mc     mean Frobenius over [R | C], rotation and translation together    (lower better)

plus ``rpe_rot`` / ``rpe_trans`` (windowed), ``traj_cos`` (scale-free direction agreement) and
``ate`` (Sim(3)-aligned) as diagnostics, and the legacy 2D ``zoom`` / ``trans`` for continuity
with pre-2026-08 runs.

Two controls, both emitted per sample:

  <metric>_chance   the static camera -- identity rotation, zero translation -- scored against
                    the same reference. Analytic, exact and free. This is the chance level, and
                    reading raw errors without it is a mistake this metric invites: the four v1
                    references differ enormously in motion magnitude, so a raw error mostly
                    reports which camera the sample drew. <metric>_rel puts them on one scale;
                    < 1 means the generation beat a locked-off camera.

  --shuffle-refs    scores each generation against a DIFFERENT sample's reference. With four
                    camera clips a derangement always draws one of the other three, so this is
                    a direct test of whether the metric can tell the references apart at all.
                    If shuffled ~= matched, the number is not measuring camera adherence.

Known limitations:
  * VGGT is a static-scene model; its accuracy degrades where a moving subject dominates the
    frame. ``ref_motion_scale`` / ``gen_motion_scale`` are emitted so a trajectory the
    estimator saw as motionless is visible rather than silently scored.
  * Camera motion is all this sees -- not scene coherence, temporal flicker, or quality.

Usage:
    python eval_camera/eval.py \\
        --generated-dir /path/to/generated/videos \\
        --dataset-dir   /path/to/dataset/camera \\
        --output-dir    /path/to/output \\
        --backend vggt
"""

from __future__ import annotations

import argparse
import csv
import json
from pathlib import Path

import numpy as np
from tqdm import tqdm

from metric import SCORED_METRICS, eval_metrics, legacy_2d_score, score
from traj import BACKENDS, MAX_FRAMES_DEFAULT, extract_trajectory, limit_threads

# Columns written to results.csv, in order: every metric with its control and ratio.
_NO_RATIO = {"traj_cos"}


def _columns(include_legacy: bool) -> list[str]:
    cols: list[str] = []
    for metric in eval_metrics:
        if metric in ("zoom", "trans"):
            continue
        cols += [metric, f"{metric}_chance"]
        if metric not in _NO_RATIO:
            cols.append(f"{metric}_rel")
    if include_legacy:
        for metric in ("zoom", "trans"):
            cols += [metric, f"{metric}_chance", f"{metric}_rel"]
    return cols


def evaluate_camera(
    generated_dir: Path,
    dataset_dir: Path,
    output_dir: Path | None = None,
    shuffle_refs: bool = False,
    backend: str = "vggt",
    max_frames: int = MAX_FRAMES_DEFAULT,
    cache_dir: Path | None = None,
    device: str = "cuda",
    include_legacy: bool = True,
) -> None:
    generated_files = {f.stem: f for f in sorted(generated_dir.glob("*.mp4"))}
    reference_files = {f.stem: f for f in sorted(dataset_dir.glob("*.mp4"))}

    sample_ids = sorted(set(generated_files) & set(reference_files))
    if not sample_ids:
        print(f"No matching samples between {generated_dir} and {dataset_dir}")
        return

    pairing = dict(zip(sample_ids, sample_ids))
    if shuffle_refs:
        import sys

        sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
        from derange import deranged_pairing  # noqa: E402

        pairing = deranged_pairing(sample_ids)

    if cache_dir is None and output_dir is not None:
        # Default the cache beside the results so a re-score after a metric change is free,
        # and so the four reference trajectories are extracted once per run rather than
        # once per sample.
        cache_dir = output_dir / "traj_cache"

    print(f"Evaluating {len(sample_ids)} sample(s) with backend={backend}...")
    per_sample: dict[str, dict] = {}
    columns = _columns(include_legacy)

    for sid in tqdm(sample_ids):
        ref_id = pairing[sid]
        try:
            gen_traj = extract_trajectory(
                generated_files[sid], backend, max_frames, cache_dir, device
            )
            ref_traj = extract_trajectory(
                reference_files[ref_id], backend, max_frames, cache_dir, device
            )
        except (ValueError, RuntimeError) as exc:
            print(f"  {sid}: skipped ({exc})")
            continue

        row = score(gen_traj, ref_traj)
        if include_legacy:
            try:
                row.update(
                    legacy_2d_score(
                        generated_files[sid], reference_files[ref_id], cache_dir
                    )
                )
            except (ValueError, ImportError) as exc:
                print(f"  {sid}: legacy 2D metrics unavailable ({exc})")
        row["ref_id"] = ref_id
        per_sample[sid] = row

    if not per_sample:
        print("No samples scored.")
        return

    scored_ids = sorted(per_sample)
    mean = {}
    for key in columns:
        vals = [
            per_sample[s][key]
            for s in scored_ids
            if per_sample[s].get(key) is not None and np.isfinite(per_sample[s].get(key, np.nan))
        ]
        mean[key] = float(np.mean(vals)) if vals else float("nan")

    beats = {
        m: sum(1 for s in scored_ids if per_sample[s][m] < per_sample[s][f"{m}_chance"])
        for m in SCORED_METRICS
    }

    print(f"\n{'=' * 60}")
    print(f"Mean metrics over {len(scored_ids)} sample(s)"
          f"{' [SHUFFLED REFS]' if shuffle_refs else ''}:")
    for key in columns:
        print(f"  {key}: {mean[key]:.6f}")
    print("  beats static control:")
    for m in SCORED_METRICS:
        print(f"    {m}: {beats[m]}/{len(scored_ids)}")

    if output_dir is None:
        return

    output_dir.mkdir(parents=True, exist_ok=True)
    result = {
        "backend": backend,
        "max_frames": max_frames,
        "shuffled_refs": shuffle_refs,
        "mean": {k: (None if v != v else round(v, 6)) for k, v in mean.items()},
        "beats_control": {m: f"{beats[m]}/{len(scored_ids)}" for m in SCORED_METRICS},
        "per_sample": {
            sid: {
                k: (round(v, 6) if isinstance(v, float) and v == v else v)
                for k, v in per_sample[sid].items()
            }
            for sid in scored_ids
        },
    }
    (output_dir / "results.json").write_text(json.dumps(result, indent=2))

    with (output_dir / "results.csv").open("w", newline="") as handle:
        writer = csv.writer(handle)
        writer.writerow(["sample_id"] + columns)
        writer.writerow(["average"] + [f"{mean[k]:.6f}" for k in columns])
        for sid in scored_ids:
            writer.writerow(
                [sid]
                + [
                    ""
                    if per_sample[sid].get(k) is None or per_sample[sid].get(k) != per_sample[sid].get(k)
                    else f"{per_sample[sid][k]:.6f}"
                    for k in columns
                ]
            )
    print(f"Results saved to {output_dir}")


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Evaluate camera-trajectory adherence against a reference camera clip.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument(
        "--generated-dir",
        required=True,
        help="Directory containing generated .mp4 files (one per sample ID).",
    )
    parser.add_argument(
        "--dataset-dir",
        required=True,
        help=(
            "Directory of camera reference videos, i.e. <dataset>/camera. Expects "
            "<id>.mp4 whose stems match the generated video stems."
        ),
    )
    parser.add_argument(
        "--output-dir",
        default=None,
        help="Optional folder to write results.json and results.csv.",
    )
    parser.add_argument(
        "--shuffle-refs",
        action="store_true",
        help=(
            "Score each generation against a different sample's reference. The static control "
            "is always emitted as <metric>_chance regardless; this is the stronger test of "
            "whether the metric can distinguish the reference clips from one another."
        ),
    )
    parser.add_argument("--backend", default="vggt", choices=BACKENDS)
    parser.add_argument("--max-frames", type=int, default=MAX_FRAMES_DEFAULT)
    parser.add_argument(
        "--cache-dir",
        default=None,
        help="Trajectory cache. Defaults to <output-dir>/traj_cache.",
    )
    parser.add_argument("--device", default="cuda")
    parser.add_argument(
        "--cpu-threads",
        type=int,
        default=2,
        help="Cap on OpenCV/torch CPU threads. The GPU is the bottleneck; leaving these "
             "uncapped makes sharded workers fight for cores instead.",
    )
    parser.add_argument(
        "--no-legacy-2d",
        action="store_true",
        help="Skip the legacy zoom/trans diagnostics (saves a CPU pass over both videos).",
    )
    args = parser.parse_args()
    limit_threads(args.cpu_threads)

    generated_dir = Path(args.generated_dir)
    dataset_dir = Path(args.dataset_dir)
    if not generated_dir.exists():
        parser.error(f"Generated video directory not found: {generated_dir}")
    if not dataset_dir.exists():
        parser.error(f"Camera reference directory not found: {dataset_dir}")

    evaluate_camera(
        generated_dir=generated_dir,
        dataset_dir=dataset_dir,
        output_dir=Path(args.output_dir) if args.output_dir else None,
        shuffle_refs=args.shuffle_refs,
        backend=args.backend,
        max_frames=args.max_frames,
        cache_dir=Path(args.cache_dir) if args.cache_dir else None,
        device=args.device,
        include_legacy=not args.no_legacy_2d,
    )


if __name__ == "__main__":
    main()