rpx_benchmark.examples.summarize_metrics

  1"""Calculate Φ and JEDI from CSV, JSON or JSONL raw metric records.
  2
  3Records need scene/scene_id, phase, and explicitly selected metric columns.
  4Optional model and task columns separate independent runs. Three canonical
  5phases are required in each run; incomplete scenes are reported and dropped
  6by the existing complete-case analysis. Aggregate paper tables cannot recover
  7the paired observations needed for Φ.
  8"""
  9
 10from __future__ import annotations
 11
 12import argparse
 13import csv
 14import hashlib
 15import json
 16import math
 17from dataclasses import asdict
 18from pathlib import Path
 19from typing import Any
 20
 21from ..exceptions import MetricError
 22from ..metrics.specs import get_spec
 23from ..phi_jedi_summary import CANONICAL_PHASE_ORDER, summarize_phi_jedi
 24
 25
 26def read_records(path: Path) -> list[dict[str, Any]]:
 27    """Read a list of records; runner JSON may wrap them in ``per_sample``."""
 28    records: Any
 29    if path.suffix.lower() == ".csv":
 30        with path.open(newline="", encoding="utf-8") as handle:
 31            records = list(csv.DictReader(handle))
 32    elif path.suffix.lower() in {".jsonl", ".ndjson"}:
 33        records = [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
 34    else:
 35        records = json.loads(path.read_text())
 36        if isinstance(records, dict):
 37            wrapper = records
 38            records = wrapper.get("per_sample")
 39            if isinstance(records, list):
 40                # Runner JSON keeps model and task once, at the top level.
 41                inherited = {k: wrapper[k] for k in ("model", "task") if isinstance(wrapper.get(k), str)}
 42                records = [
 43                    {**inherited, **r} if isinstance(r, dict) else r for r in records
 44                ]
 45    if (
 46        not isinstance(records, list)
 47        or not records
 48        or not all(isinstance(r, dict) for r in records)
 49    ):
 50        raise MetricError("Input must contain a nonempty list of raw metric records")
 51    return records
 52
 53
 54def calculate(records: list[dict[str, Any]], metric_keys: list[str]) -> dict[str, Any]:
 55    """Validate raw values and summarize each (model, task) independently."""
 56    if not metric_keys or len(set(metric_keys)) != len(metric_keys):
 57        raise MetricError("Choose at least one metric, with no duplicates")
 58    specs = {key: asdict(get_spec(key)) for key in metric_keys}
 59    groups: dict[tuple[str, str], list[dict[str, Any]]] = {}
 60    aliases = {"0": "clutter", "1": "interaction", "2": "clean"}
 61    for index, record in enumerate(records, 1):
 62        row = dict(record)
 63        scene = row.get("scene") or row.get("scene_id")
 64        if scene is None or not str(scene).strip():
 65            raise MetricError(f"Row {index}: scene/scene_id is required")
 66        phase = aliases.get(str(row.get("phase")), str(row.get("phase")).lower())
 67        if phase not in CANONICAL_PHASE_ORDER:
 68            raise MetricError(f"Row {index}: phase must be clutter, interaction, clean or 0/1/2")
 69        row.update(scene=str(scene), phase=phase)
 70        for key in metric_keys:
 71            raw = row.get(key)
 72            if isinstance(raw, bool):
 73                raise MetricError(f"Row {index}: {key} cannot be a boolean")
 74            if raw is None:
 75                raise MetricError(f"Row {index}: {key} must be a finite number")
 76            try:
 77                value = float(raw)
 78            except (TypeError, ValueError) as exc:
 79                raise MetricError(f"Row {index}: {key} must be a finite number") from exc
 80            if not math.isfinite(value):
 81                raise MetricError(f"Row {index}: {key} must be a finite number")
 82            row[key] = value
 83        group = (str(row.get("model") or "custom"), str(row.get("task") or "unspecified"))
 84        groups.setdefault(group, []).append(row)
 85    results = []
 86    for (model, task), rows in groups.items():
 87        phases = {row["phase"] for row in rows}
 88        if phases != set(CANONICAL_PHASE_ORDER):
 89            raise MetricError(
 90                f"{model}/{task}: all three phases are required; found {sorted(phases)}"
 91            )
 92        summary = summarize_phi_jedi(rows, metric_keys=metric_keys).to_dict()
 93        jedi = summary["jedi"]
 94        j_min = min(jedi["per_phase"].values()) if jedi else None
 95        results.append(
 96            {"model": model, "task": task, "input_rows": len(rows), "j_min": j_min, **summary}
 97        )
 98    return {
 99        "protocol": "three_phase_complete_case",
100        "metric_specs": specs,
101        "phase_order": list(CANONICAL_PHASE_ORDER),
102        "results": results,
103    }
104
105
106def main() -> None:
107    parser = argparse.ArgumentParser(description=__doc__)
108    parser.add_argument("--input", type=Path, required=True)
109    parser.add_argument("--metrics", nargs="+", required=True)
110    parser.add_argument("--output", type=Path, required=True)
111    args = parser.parse_args()
112    try:
113        result = calculate(read_records(args.input), args.metrics)
114    except MetricError as exc:
115        parser.exit(2, f"rpx: cannot compute Φ/JEDI: {exc}\n")
116    result["input_sha256"] = hashlib.sha256(args.input.read_bytes()).hexdigest()
117    args.output.parent.mkdir(parents=True, exist_ok=True)
118    args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n")
119    for row in result["results"]:
120        phi = (row.get("phi") or {}).get("phi_conservative")
121        per_phase = (row.get("jedi") or {}).get("per_phase") or {}
122        phases = "  ".join(f"{name} {value:.2f}" for name, value in per_phase.items())
123        phi_text = f"{phi:.2f}" if isinstance(phi, (int, float)) else "n/a"
124        print(
125            f"{row['model']} / {row['task']}: Φ {phi_text}   Jmin {row['j_min']:.2f}"
126            f"   ({row['n_eff']} scenes; J per phase: {phases})"
127        )
128    print(f"Full report: {args.output}")
129
130
131if __name__ == "__main__":
132    main()