CupOFMoCA / CupOFMoCA-PROTAC /scripts /runs /aggregate_ablation.py

Add the PROTAC release

f12d3b3 verified 8 days ago

7.7 kB

#!/usr/bin/env python3
"""
Aggregate the EGFR-VHL ablation outputs into one table.
Walks generation_logs/ablation_egfr/<tag>/ 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<seed>\d+)$", tag)
    if m:
        return {"kind": "baseline", "seed": int(m["seed"])}
    m = re.match(r"egfr_(?P<sc>[a-z]+)_w(?P<w>[\d.]+)_seed(?P<seed>\d+)$", tag)
    if m:
        return {"kind": "main", "scorer": m["sc"],
                "weight": float(m["w"]), "seed": int(m["seed"])}
    m = re.match(r"egfr_accept_(?P<acc>on|off)_seed(?P<seed>\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()

	#!/usr/bin/env python3
	"""
	Aggregate the EGFR-VHL ablation outputs into one table.

	Walks generation_logs/ablation_egfr/<tag>/ 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<seed>\d+)$", tag)
	if m:
	return {"kind": "baseline", "seed": int(m["seed"])}
	m = re.match(r"egfr_(?P<sc>[a-z]+)_w(?P<w>[\d.]+)_seed(?P<seed>\d+)$", tag)
	if m:
	return {"kind": "main", "scorer": m["sc"],
	"weight": float(m["w"]), "seed": int(m["seed"])}
	m = re.match(r"egfr_accept_(?P<acc>on\|off)_seed(?P<seed>\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("\nDegradeMaster 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("\nMemorization % (> 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()