#!/usr/bin/env python3 """ Aggregate the EGFR-VHL ablation outputs into one table. Walks generation_logs/ablation_egfr// and for each, parses the summary CSV into batch_report's compute_stats. Emits a TSV with one row per (tag, seed). Also prints a per-cell (scorer x weight) and acceptance summary averaged across seeds. """ import argparse import csv import importlib.util import re import statistics from pathlib import Path import pandas as pd # Locate batch_report.py in scripts/ SCRIPT_DIR = Path(__file__).resolve().parent.parent spec = importlib.util.spec_from_file_location("batch_report", SCRIPT_DIR / "batch_report.py") br = importlib.util.module_from_spec(spec) spec.loader.exec_module(br) REPO_ROOT = SCRIPT_DIR.parent OUT_BASE = REPO_ROOT / "generation_logs" / "ablation_egfr" def parse_tag(tag: str) -> dict: m = re.match(r"egfr_baseline_seed(?P\d+)$", tag) if m: return {"kind": "baseline", "seed": int(m["seed"])} m = re.match(r"egfr_(?P[a-z]+)_w(?P[\d.]+)_seed(?P\d+)$", tag) if m: return {"kind": "main", "scorer": m["sc"], "weight": float(m["w"]), "seed": int(m["seed"])} m = re.match(r"egfr_accept_(?Pon|off)_seed(?P\d+)$", tag) if m: return {"kind": "accept", "accept": m["acc"], "seed": int(m["seed"])} return {} def stats_for_csv(csv_path: Path, protac_module: str) -> dict: br._PROTAC_MODULE = protac_module br._protac_analyzer_state = None br._PROTACDB_FP_CACHE = getattr(br, "_PROTACDB_FP_CACHE", {}) df = br.run_evaluate_full_protac_file(csv_path) if csv_path.suffix == ".txt" \ else br.run_evaluate(csv_path) return br.compute_stats(df) def main(): ap = argparse.ArgumentParser() ap.add_argument("--out", default=str(OUT_BASE / "ablation_results.tsv"), help="output TSV path") ap.add_argument("--module", default="ErlotinibVHL") args = ap.parse_args() rows = [] for run_dir in sorted(OUT_BASE.iterdir()): if not run_dir.is_dir(): continue csvs = list(run_dir.glob("defog_local_edit_*_summary.csv")) if not csvs: print(f"[skip] {run_dir.name}: no summary csv") continue # In case of multiple (resumed runs), take the newest csv_path = sorted(csvs, key=lambda p: p.stat().st_mtime)[-1] try: s = stats_for_csv(csv_path, args.module) except Exception as e: print(f"[fail] {run_dir.name}: {e}") continue meta = parse_tag(run_dir.name) rows.append({ "tag": run_dir.name, **meta, "n_total": s["n_total"], "std_valid": round(s["std_valid_rate"] * s["n_total"] / 100), "asm_valid": s["n_valid"], "asm_pct": round(s["asm_valid_rate"], 1), "sa_mean": round(s["sa_mean"], 2), "qed_mean": round(s["qed_mean"], 3), "dm_mean": round(s["degrad_mean"], 3), "dm_std": round(s["degrad_std"], 3), "dm_ge05": round(s["degrad_ge05"], 1), "mem_ge08": round(s.get("train_sim_pct_ge_0_8", 0), 1), "logp_in": round(s["logp_in_window_pct"], 1), "mw_in": round(s["mw_in_window_pct"], 1), "rotb_in": round(s["rot_in_window_pct"], 1), }) print(f"[ok] {run_dir.name}: asm={s['n_valid']}/{s['n_total']}, " f"dm={s['degrad_mean']:.3f}") if not rows: print("No results to aggregate.") return # Write per-run TSV (union of all keys across baseline / main / accept rows) fieldnames = [] for r in rows: for k in r: if k not in fieldnames: fieldnames.append(k) with open(args.out, "w", newline="") as f: w = csv.DictWriter(f, fieldnames=fieldnames, delimiter="\t", extrasaction="ignore", restval="") w.writeheader() w.writerows(rows) print(f"\nPer-run TSV: {args.out} ({len(rows)} rows)") # Aggregate baseline (shared across all w=1 cells) base_rows = [r for r in rows if r.get("kind") == "baseline"] base_acc_off = base_rows # acceptance=off baseline reuses the baseline runs def agg(group): asm = [g["asm_valid"] for g in group] dm = [g["dm_mean"] for g in group] mem = [g["mem_ge08"] for g in group] if not group: return None return { "asm_m": statistics.mean(asm), "asm_s": statistics.stdev(asm) if len(asm) > 1 else 0, "dm_m": statistics.mean(dm), "dm_s": statistics.stdev(dm) if len(dm) > 1 else 0, "mem_m": statistics.mean(mem), "mem_s": statistics.stdev(mem) if len(mem) > 1 else 0, "n": len(group), } main_rows = [r for r in rows if r.get("kind") == "main"] if main_rows or base_rows: # collect into (scorer, weight) -> aggregate by_cell = {} for r in main_rows: by_cell.setdefault((r["scorer"], r["weight"]), []).append(r) scorers = sorted({r["scorer"] for r in main_rows}) weights = sorted({r["weight"] for r in main_rows} | {1.0}) print("\n## Main ablation (EGFR-VHL, 100 samples, mean ± std over 3 seeds)\n") # Markdown table header_w = "| scorer | " + " | ".join(f"w={w:g}" for w in weights) + " |" sep = "|" + "|".join(["---"] * (len(weights) + 1)) + "|" print("**AsmVal (out of 100)**\n") print(header_w); print(sep) base_agg = agg(base_acc_off) for sc in scorers: cells = [] for w in weights: if w == 1.0: a = base_agg else: a = agg(by_cell.get((sc, w), [])) cells.append(f"{a['asm_m']:.1f} ± {a['asm_s']:.1f}" if a else "—") print(f"| {sc} | " + " | ".join(cells) + " |") print("\n**DegradeMaster mean**\n") print(header_w); print(sep) for sc in scorers: cells = [] for w in weights: if w == 1.0: a = base_agg else: a = agg(by_cell.get((sc, w), [])) cells.append(f"{a['dm_m']:.3f} ± {a['dm_s']:.3f}" if a else "—") print(f"| {sc} | " + " | ".join(cells) + " |") print("\n**Memorization % (> 0.80 Tanimoto)**\n") print(header_w); print(sep) for sc in scorers: cells = [] for w in weights: if w == 1.0: a = base_agg else: a = agg(by_cell.get((sc, w), [])) cells.append(f"{a['mem_m']:.1f} ± {a['mem_s']:.1f}" if a else "—") print(f"| {sc} | " + " | ".join(cells) + " |") # Acceptance ablation acc_on = [r for r in rows if r.get("kind") == "accept" and r["accept"] == "on"] acc_off = base_acc_off # reuse baseline if acc_on or acc_off: print("\n## Acceptance ablation (EGFR-VHL, baseline weights, mean ± std over 3 seeds)\n") print("| accept | AsmVal | DegradeMaster | Mem % |") print("|---|---|---|---|") for name, group in (("off", acc_off), ("on", acc_on)): a = agg(group) if a is None: continue print(f"| {name} | {a['asm_m']:.1f} ± {a['asm_s']:.1f} | " f"{a['dm_m']:.3f} ± {a['dm_s']:.3f} | " f"{a['mem_m']:.1f} ± {a['mem_s']:.1f} |") if __name__ == "__main__": main()