scripts/export_paper_data.py

#!/usr/bin/env python3
"""Export a canonical paper_data.yaml for the autopaper pipeline.

Single source of truth for the PINO paper's facts/numbers. Sources (all read
verbatim, no recomputation beyond trivial deltas):
  - artifacts/representation_ablation/ablation_results.json  (frozen ablation)
  - artifacts/v10_line_drawn_metrics.json                     (substantivity GBM)
  - artifacts/frozen_eval_{morgan,openpom_256}.json           (A/B readout)
  - dataset curation stats (trainable corpus, repair chain)

Macro naming contract (autopaper build_paper.py): keys are CamelCase letters
only; the builder prefixes each with "D" (e.g. pomThreshRho -> \\DPomThreshRho)
and also emits a lowercase legacy alias. Digits in a key are spelled to words,
so keep digits OUT of key names (use pomDim -> \\DPomDim, value carries 256).
"""
from __future__ import annotations

import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "artifacts" / "paper_data.yaml"


def load(rel: str):
    p = ROOT / rel
    return json.loads(p.read_text()) if p.exists() else None


def r(x, n=3):
    return round(float(x), n) if x is not None else None


def main() -> int:
    abl = load("artifacts/representation_ablation/ablation_results.json") or {}
    v10 = load("artifacts/v10_line_drawn_metrics.json") or {}
    # Prefer the authoritative GPU (20-epoch) readout; fall back to the CPU
    # (6-epoch) run only if the GPU evals are absent. Prevents regenerating
    # paper_data.yaml with the superseded small-n CPU numbers.
    frozen_m = load("artifacts/frozen_eval_morgan_gpu.json") or load("artifacts/frozen_eval_morgan.json") or {}
    frozen_p = load("artifacts/frozen_eval_openpom_256_gpu.json") or load("artifacts/frozen_eval_openpom_256.json") or {}

    tasks = abl.get("tasks", {})
    thr = tasks.get("odor_threshold", {})
    sub = tasks.get("substantivity", {})
    desc = tasks.get("descriptor", {})
    subst = tasks.get("substitution", {})

    def rep(task, name):
        return (task.get("representations", {}) or {}).get(name, {})

    def delta(task, name):
        return (task.get("paired_deltas", {}) or {}).get(name, {})

    # substantivity GBM locked holdout
    sub_lock = (v10.get("part1", {}).get("substantivity", {}).get("locked_holdout", {}))
    sub_cv = (v10.get("part1", {}).get("substantivity", {}).get("cv", {}))

    # A/B frozen readout (verbatim)
    def tr_acc(d):
        return (d.get("substitution_triplets", {}) or {}).get("accuracy")

    def tr_n(d):
        return (d.get("substitution_triplets", {}) or {}).get("n_triplets")

    def pr_cos(d):
        return (d.get("prospective_formulas", {}) or {}).get("mean_family_profile_cosine")

    def pr_n(d):
        return (d.get("prospective_formulas", {}) or {}).get("n_formulas")

    dm = {
        # ---- representation dims ----
        "morganDim": 138,
        "pomDim": 256,
        "abMorganInputDim": 151,
        "abPomInputDim": 269,
        # ---- frozen ablation: odor threshold ----
        "threshN": thr.get("n"),
        "threshMorganRho": r(rep(thr, "morgan").get("spearman_mean")),
        "threshPomRho": r(rep(thr, "real_pom").get("spearman_mean")),
        "threshPomPhysRho": r(rep(thr, "pom_plus_physics").get("spearman_mean")),
        "threshPomVsMorganDelta": r(delta(thr, "real_pom_vs_morgan").get("spearman_mean_delta")),
        # ---- frozen ablation: substantivity ----
        "substN": sub.get("n"),
        "substMorganRho": r(rep(sub, "morgan").get("spearman_mean")),
        "substPomRho": r(rep(sub, "real_pom").get("spearman_mean")),
        "substPomPhysRho": r(rep(sub, "pom_plus_physics").get("spearman_mean")),
        "substPomVsMorganDelta": r(delta(sub, "real_pom_vs_morgan").get("spearman_mean_delta")),
        # ---- frozen ablation: descriptor (macro-F1) ----
        "descN": desc.get("n"),
        "descNClasses": desc.get("n_classes"),
        "descMorganMacroF": r(rep(desc, "morgan").get("macro_f1_mean")),
        "descPomMacroF": r(rep(desc, "real_pom").get("macro_f1_mean")),
        "descPomPhysMacroF": r(rep(desc, "pom_plus_physics").get("macro_f1_mean")),
        "descPomVsMorganDelta": r(delta(desc, "real_pom_vs_morgan").get("macro_f1_mean_delta")),
        # ---- frozen ablation: substitution retrieval ----
        "substPairsN": subst.get("n_pairs"),
        "substMorganTopFive": r(rep(subst, "morgan").get("top5_accuracy")),
        "substPomTopFive": r(rep(subst, "real_pom").get("top5_accuracy")),
        "substPomPhysTopFive": r(rep(subst, "pom_plus_physics").get("top5_accuracy")),
        # ---- substantivity GBM locked holdout (v10_line_drawn) ----
        "gbmNModelReady": v10.get("part1", {}).get("substantivity", {}).get("n_model_ready"),
        "gbmLockedRho": r(sub_lock.get("spearman_rho")),
        "gbmLockedRtwo": r(sub_lock.get("r2")),
        "gbmLockedMae": r(sub_lock.get("mae")),
        "gbmCvRho": r(sub_cv.get("mean_spearman_rho")),
        # ---- two-arm A/B frozen readout (verbatim, CPU run) ----
        "abMorganInputDimReadout": frozen_m.get("input_embedding_dim"),
        "abPomInputDimReadout": frozen_p.get("input_embedding_dim"),
        "abMorganTripletAcc": r(tr_acc(frozen_m)),
        "abPomTripletAcc": r(tr_acc(frozen_p)),
        "abTripletN": tr_n(frozen_m) or tr_n(frozen_p) or 20,
        "abMorganProspectiveCos": r(pr_cos(frozen_m)),
        "abPomProspectiveCos": r(pr_cos(frozen_p)),
        "abProspectiveN": pr_n(frozen_m) or pr_n(frozen_p) or 40,
        "abMorganValTotal": r(frozen_m.get("final_val_total")),
        "abPomValTotal": r(frozen_p.get("final_val_total")),
        # ---- dataset curation stats ----
        "dsTotalRows": 5708,
        "dsTrainableRows": 5678,
        "dsTrainablePct": 99.47,
        "dsBannedCasBefore": 309,
        "dsBannedCasAfter": 0,
        "dsLiteratureRepairPairs": 106,
        "dsTrajSteps": 49,
        "dsTrajDim": 138,
        "dsTripletN": 20,
        "dsProspectiveN": 40,
        # ---- honest negatives ----
        "negRetrievalTopFive": 0,
        "negCharacterMacroF": 0.134,
    }

    # drop Nones so the builder never emits an empty macro
    dm = {k: v for k, v in dm.items() if v is not None}

    lines = [
        "# Auto-generated by scripts/export_paper_data.py. Do not edit by hand.",
        "# Canonical facts/numbers for the PINO paper (autopaper pipeline).",
        "data_macros:",
    ]
    for k, v in dm.items():
        lines.append(f"  {k}: {v}")

    OUT.write_text("\n".join(lines) + "\n")
    print(f"wrote {OUT} with {len(dm)} data_macros")
    for k, v in dm.items():
        print(f"  \\D{k[0].upper()+k[1:]} = {v}")
    return 0


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