analyze_results.py

"""Analyze scored benchmark results from tasks_and_rubrics.tsv.

Prints five analysis sections to the terminal and writes a summary
to results_summary.txt.

Usage:
    python3 analyze_results.py [path/to/results.tsv]
"""

from __future__ import annotations

import csv
import math
import sys
from collections import defaultdict
from pathlib import Path

TSV_PATH = Path(__file__).parent / "tasks_and_rubrics.tsv"
SUMMARY_PATH = Path(__file__).parent / "results_summary.txt"

SCORE_DIMS = [
    "trajectory_score",
    "energy_score",
    "consistency_score",
    "causal_score",
    "surface_score",
    "permanence_score",
    "total_score",
]

_BACKGROUND_KEYWORDS = [
    "trees in the background", "brick wall", "clear blue sky",
    "mountains in the distance", "cluttered workshop", "crowd watching",
]
_WEATHER_KEYWORDS = [
    "sunny day", "overcast skies", "light drizzle",
    "golden-hour sunset", "at dawn",
]
_CAMERA_KEYWORDS = [
    "steady on a tripod", "slowly zooms in",
    "pans to follow", "handheld with slight shake",
]


def _load_rows(path: Path) -> list[dict[str, str]]:
    with path.open(encoding="utf-8") as f:
        return list(csv.DictReader(f, delimiter="\t"))


def _f(v: str | None) -> float:
    if v is None:
        return 0.0
    try:
        return float(v)
    except (ValueError, TypeError):
        return 0.0


def _get(row: dict, key: str) -> str:
    return row.get(key, "")


def _mean(vals: list[float]) -> float:
    return sum(vals) / len(vals) if vals else 0.0


def _std(vals: list[float]) -> float:
    if len(vals) < 2:
        return 0.0
    m = _mean(vals)
    return math.sqrt(sum((v - m) ** 2 for v in vals) / (len(vals) - 1))


def _fmt(m: float, s: float) -> str:
    return f"{m:.3f}±{s:.3f}"


def _table(headers: list[str], rows: list[list[str]], col_widths: list[int] | None = None) -> str:
    if col_widths is None:
        col_widths = [
            max(len(h), *(len(r[i]) for r in rows)) + 2
            for i, h in enumerate(headers)
        ]
    lines: list[str] = []
    hdr = "".join(h.ljust(w) for h, w in zip(headers, col_widths))
    lines.append(hdr)
    lines.append("─" * len(hdr))
    for row in rows:
        lines.append("".join(c.ljust(w) for c, w in zip(row, col_widths)))
    return "\n".join(lines)


def _detect_distractors(prompt: str) -> dict[str, bool]:
    p = prompt.lower()
    return {
        "background": any(kw in p for kw in _BACKGROUND_KEYWORDS),
        "weather":    any(kw in p for kw in _WEATHER_KEYWORDS),
        "camera":     any(kw in p for kw in _CAMERA_KEYWORDS),
    }


# ═════════════════════════════════════════════════════════════════════════
#  Analysis sections
# ═════════════════════════════════════════════════════════════════════════

def model_comparison(rows: list[dict]) -> str:
    by_model: dict[str, list[dict]] = defaultdict(list)
    for r in rows:
        by_model[r["model"]].append(r)

    headers = ["Model", "N"] + [d.replace("_score", "") for d in SCORE_DIMS]
    tbl_rows: list[list[str]] = []
    for model in sorted(by_model):
        entries = by_model[model]
        cols: list[str] = [model, str(len(entries))]
        for dim in SCORE_DIMS:
            vals = [_f(_get(e, dim)) for e in entries]
            cols.append(_fmt(_mean(vals), _std(vals)))
        tbl_rows.append(cols)

    return "1) MODEL COMPARISON\n\n" + _table(headers, tbl_rows)


def seed_difficulty(rows: list[dict]) -> str:
    by_seed: dict[str, list[float]] = defaultdict(list)
    for r in rows:
        by_seed[r.get("seed_id", "")].append(_f(_get(r, "total_score")))

    ranked = sorted(by_seed.items(), key=lambda kv: _mean(kv[1]))

    headers = ["Seed", "N", "Mean Total", "Std", "Status"]
    tbl_rows: list[list[str]] = []
    for seed_id, vals in ranked:
        m = _mean(vals)
        s = _std(vals)
        if m > 0.75:
            status = "⚠ too_easy"
        elif m < 0.35:
            status = "⚠ too_hard"
        else:
            status = "ok"
        tbl_rows.append([seed_id, str(len(vals)), f"{m:.4f}", f"{s:.4f}", status])

    return "2) SEED DIFFICULTY RANKING (hardest → easiest)\n\n" + _table(headers, tbl_rows)


def distractor_sensitivity(rows: list[dict]) -> str:
    groups: dict[str, dict[str, list[float]]] = {
        dt: {"with": [], "without": []}
        for dt in ("background", "weather", "camera")
    }

    for r in rows:
        det = _detect_distractors(r.get("task_description", ""))
        total = _f(_get(r, "total_score"))
        for dt in ("background", "weather", "camera"):
            bucket = "with" if det[dt] else "without"
            groups[dt][bucket].append(total)

    headers = ["Distractor", "With (mean)", "Without (mean)", "Delta", "N_with", "N_without"]
    tbl_rows: list[list[str]] = []
    for dt in ("background", "weather", "camera"):
        w = groups[dt]["with"]
        wo = groups[dt]["without"]
        mw = _mean(w)
        mwo = _mean(wo)
        delta = mw - mwo
        tbl_rows.append([
            dt, f"{mw:.4f}", f"{mwo:.4f}", f"{delta:+.4f}",
            str(len(w)), str(len(wo)),
        ])

    return "3) DISTRACTOR SENSITIVITY ANALYSIS\n\n" + _table(headers, tbl_rows)


def winner_per_seed(rows: list[dict]) -> str:
    nested: dict[str, dict[str, list[float]]] = defaultdict(lambda: defaultdict(list))
    for r in rows:
        nested[r.get("seed_id", "")][r.get("model", "")].append(_f(_get(r, "trajectory_score")))

    headers = ["Seed", "Winner", "Winner Traj", "Runner-up", "Runner-up Traj"]
    tbl_rows: list[list[str]] = []
    for seed_id in sorted(nested):
        models = nested[seed_id]
        ranked = sorted(models.items(), key=lambda kv: _mean(kv[1]), reverse=True)
        if len(ranked) >= 2:
            w_name, w_vals = ranked[0]
            r_name, r_vals = ranked[1]
            tbl_rows.append([
                seed_id, w_name, f"{_mean(w_vals):.4f}",
                r_name, f"{_mean(r_vals):.4f}",
            ])
        elif ranked:
            w_name, w_vals = ranked[0]
            tbl_rows.append([seed_id, w_name, f"{_mean(w_vals):.4f}", "—", "—"])

    return "4) WINNER PER SEED (trajectory_score)\n\n" + _table(headers, tbl_rows)


def intrinsic_vs_superficial_faithfulness(rows: list[dict]) -> str:
    """Measure gap between superficial plausibility and intrinsic correctness.

    Superficial = permanence_score (object exists = looks convincing).
    Intrinsic   = mean(trajectory_score, energy_score) (follows physics).
    A large positive gap means the model LOOKS correct but ISN'T.
    """
    nested: dict[str, dict[str, list[dict]]] = defaultdict(lambda: defaultdict(list))
    for r in rows:
        nested[r.get("model", "")][r.get("seed_id", "")].append(r)

    headers = ["Model", "Seed", "Superficial", "Intrinsic", "Gap"]
    tbl_rows: list[list[str]] = []
    model_gaps: dict[str, list[float]] = defaultdict(list)

    for model in sorted(nested):
        for seed_id in sorted(nested[model]):
            entries = nested[model][seed_id]
            sup = _mean([_f(_get(e, "permanence_score")) for e in entries])
            traj = _mean([_f(_get(e, "trajectory_score")) for e in entries])
            energy = _mean([_f(_get(e, "energy_score")) for e in entries])
            intr = (traj + energy) / 2.0
            gap = sup - intr
            model_gaps[model].append(gap)
            tbl_rows.append([
                model, seed_id,
                f"{sup:.3f}", f"{intr:.3f}", f"{gap:+.3f}",
            ])

    tbl_rows.append(["", "", "", "", ""])
    for model in sorted(model_gaps):
        mg = _mean(model_gaps[model])
        tbl_rows.append([model, "MEAN GAP", "", "", f"{mg:+.3f}"])

    worst = max(
        ((m, s, sup, intr, gap) for m, seeds in nested.items()
         for s, entries in seeds.items()
         for sup, intr, gap in [(_mean([_f(_get(e, "permanence_score")) for e in entries]),
                                  (_mean([_f(_get(e, "trajectory_score")) for e in entries]) +
                                   _mean([_f(_get(e, "energy_score")) for e in entries])) / 2.0,
                                  _mean([_f(_get(e, "permanence_score")) for e in entries]) -
                                  (_mean([_f(_get(e, "trajectory_score")) for e in entries]) +
                                   _mean([_f(_get(e, "energy_score")) for e in entries])) / 2.0)]),
        key=lambda t: t[4],
    )
    note = (
        f"\n  Largest gap: {worst[0]}/{worst[1]} "
        f"(sup={worst[2]:.3f}, intr={worst[3]:.3f}, gap={worst[4]:+.3f})\n"
        f"  → This is the VBench-2.0 failure mode: visually plausible but physically wrong."
    )

    return (
        "6) INTRINSIC vs SUPERFICIAL FAITHFULNESS (VBench-2.0)\n\n"
        + _table(headers, tbl_rows) + "\n" + note
    )


def calibration_report(rows: list[dict]) -> str:
    buckets = {"too_easy": 0, "ok": 0, "too_hard": 0}
    for r in rows:
        t = _f(_get(r, "total_score"))
        if t > 0.75:
            buckets["too_easy"] += 1
        elif t < 0.35:
            buckets["too_hard"] += 1
        else:
            buckets["ok"] += 1

    total = len(rows)
    headers = ["Bucket", "Count", "Pct"]
    tbl_rows = []
    for b in ("too_hard", "ok", "too_easy"):
        pct = (buckets[b] / total * 100) if total else 0
        tbl_rows.append([b, str(buckets[b]), f"{pct:.1f}%"])
    tbl_rows.append(["TOTAL", str(total), "100.0%"])

    return "5) CALIBRATION REPORT\n\n" + _table(headers, tbl_rows)


# ═════════════════════════════════════════════════════════════════════════

def main() -> int:
    tsv = Path(sys.argv[1]) if len(sys.argv) > 1 else TSV_PATH
    if not tsv.exists():
        print(f"ERROR: {tsv} not found. Run the pipeline first.", file=sys.stderr)
        return 1

    rows = _load_rows(tsv)
    if not rows:
        print("ERROR: TSV has no data rows.", file=sys.stderr)
        return 1

    sections = [
        model_comparison(rows),
        seed_difficulty(rows),
        distractor_sensitivity(rows),
        winner_per_seed(rows),
        calibration_report(rows),
        intrinsic_vs_superficial_faithfulness(rows),  # VBench-2.0 analysis
    ]

    banner = f"philo-benchmark  ·  Analysis  ·  {len(rows)} rows from {tsv.name}"
    full_output = (
        "\n" + "═" * len(banner) + "\n"
        + banner + "\n"
        + "═" * len(banner) + "\n\n"
        + "\n\n".join(sections) + "\n"
    )

    print(full_output)

    SUMMARY_PATH.write_text(full_output, encoding="utf-8")
    print(f"\nSummary written to {SUMMARY_PATH}")
    return 0


if __name__ == "__main__":
    sys.exit(main())