scripts/decide_epsilon_from_probe.py

#!/usr/bin/env python3
"""Decisão empírica do ε∞ principal pra Sweep MAIN TCC (Análise B do plano).

Lê o CSV merged do probe e avalia 5 critérios pra cada ε candidato.
Recomenda o ε∞ que melhor balanceia os critérios.

Critérios (cada um vale 1 ponto):
  1. ASR razoável mas não saturado
       → mean ASR de PGD/MIM/SAGA está em [0.5, 0.99]?
       → satura cedo = perde diferenciação científica
  2. SSIM mediano > 0.85 (regime "ainda quase imperceptível", Sen 2020 caveat)
       → median SSIM dos 4 ataques (média) > 0.85?
  3. Variância informativa
       → IQR do ASR > 0 em pelo menos 1 ataque (alguma heterogeneidade entre imgs)
  4. Comparabilidade com literatura
       → ε ∈ {4, 8, 16}/255? (RobustBench, Hu 2024, Mahmood 2021)
  5. Hierarquia entre ataques discriminável
       → spread (max ASR - min ASR) entre ataques ≥ 0.10?

Usage:
    python scripts/decide_epsilon_from_probe.py \\
        --csv results/raw/probe_merged_no_tgr.csv

Output: tabela markdown no stdout + arquivo `decide_epsilon_report.md` na pasta de saída.
"""
from __future__ import annotations

import argparse
import sys
from pathlib import Path


def _project_root() -> Path:
    cur = Path(__file__).resolve().parent
    for p in [cur, *cur.parents]:
        if (p / "requirements.txt").exists():
            return p
    raise RuntimeError("project root not found")


PROJECT_ROOT = _project_root()
CANDIDATE_EPS = [4, 6, 8, 10, 12]
LITERATURE_REFS = {4: "RobustBench (Croce 2021)", 8: "Hu 2024", 16: "Mahmood 2021 / TGR"}


def evaluate_epsilon(df, eps_255: int) -> dict:
    """Avalia 1 ε candidato nos 5 critérios. Retorna dict de resultados."""
    sub = df[df["eps_255"] == eps_255]
    if sub.empty:
        return {"eps_255": eps_255, "valid": False}

    # Critério 1: ASR razoável mas não saturado pra ataques iterativos
    iterative = ["PGD", "MIM", "SAGA"]
    iter_asrs = [sub[sub["attack"] == a]["asr"].mean()
                 for a in iterative if a in sub["attack"].values]
    avg_iter_asr = sum(iter_asrs) / max(len(iter_asrs), 1)
    c1 = 0.50 <= avg_iter_asr <= 0.99
    c1_detail = f"avg ASR (PGD/MIM/SAGA) = {avg_iter_asr:.3f}"

    # Critério 2: SSIM mediano alto
    ssim_medians = [sub[sub["attack"] == a]["ssim"].median()
                    for a in sub["attack"].unique()]
    median_ssim = sum(ssim_medians) / len(ssim_medians) if ssim_medians else 0.0
    c2 = median_ssim > 0.85
    c2_detail = f"median SSIM (avg over attacks) = {median_ssim:.3f}"

    # Critério 3: variância informativa (algum ataque tem IQR > 0)
    iqrs = []
    for a in sub["attack"].unique():
        vals = sub[sub["attack"] == a]["asr"].dropna()
        if len(vals) > 0:
            iqrs.append(float(vals.quantile(0.75) - vals.quantile(0.25)))
    max_iqr = max(iqrs) if iqrs else 0.0
    c3 = max_iqr > 0
    c3_detail = f"max IQR ASR = {max_iqr:.3f}"

    # Critério 4: literatura
    c4 = eps_255 in LITERATURE_REFS
    c4_detail = LITERATURE_REFS.get(eps_255, "—")

    # Critério 5: hierarquia discriminável entre ataques
    asr_per_attack = {a: sub[sub["attack"] == a]["asr"].mean()
                      for a in sub["attack"].unique()}
    if asr_per_attack:
        spread = max(asr_per_attack.values()) - min(asr_per_attack.values())
    else:
        spread = 0.0
    c5 = spread >= 0.10
    c5_detail = f"spread ASR = {spread:.3f}"

    score = sum([c1, c2, c3, c4, c5])

    return {
        "eps_255": eps_255,
        "valid": True,
        "score": score,
        "c1": (c1, c1_detail),
        "c2": (c2, c2_detail),
        "c3": (c3, c3_detail),
        "c4": (c4, c4_detail),
        "c5": (c5, c5_detail),
        "asr_per_attack": asr_per_attack,
        "median_ssim": median_ssim,
    }


def render_report(results: list[dict], df) -> str:
    """Gera relatório markdown a partir dos resultados."""
    lines = ["# Decisão de ε∞ — relatório do probe", ""]
    lines.append(f"**Modelos**: {df['model'].nunique()}  ")
    lines.append(f"**Ataques**: {sorted(df['attack'].unique())}  ")
    lines.append(f"**ε grid no CSV**: {sorted(df['eps_255'].unique())}/255  ")
    lines.append(f"**Imagens**: {df['image'].nunique()}")
    lines.append("")
    lines.append("## Critérios (cada um vale 1 ponto)")
    lines.append("")
    lines.append("1. ASR razoável mas não saturado: `mean ASR (PGD, MIM, SAGA)` ∈ [0.50, 0.99]")
    lines.append("2. SSIM mediano > 0.85 (Sen 2020 / Liu 2025: regime ~imperceptível)")
    lines.append("3. Variância informativa: max IQR(ASR) > 0 entre ataques")
    lines.append("4. Comparabilidade com literatura: ε ∈ {4, 8, 16}/255")
    lines.append("5. Hierarquia discriminável: max(ASR) − min(ASR) ≥ 0.10")
    lines.append("")

    # Tabela resumo
    lines.append("## Resumo por ε candidato")
    lines.append("")
    lines.append("| ε | Score | C1 ASR | C2 SSIM | C3 IQR | C4 lit | C5 spread |")
    lines.append("|---|---|---|---|---|---|---|")
    valid_results = [r for r in results if r.get("valid")]
    for r in valid_results:
        lines.append(
            f"| **{r['eps_255']}/255** | **{r['score']}/5** "
            f"| {'✓' if r['c1'][0] else '✗'} ({r['c1'][1].split('=')[-1].strip()}) "
            f"| {'✓' if r['c2'][0] else '✗'} ({r['c2'][1].split('=')[-1].strip()}) "
            f"| {'✓' if r['c3'][0] else '✗'} ({r['c3'][1].split('=')[-1].strip()}) "
            f"| {'✓' if r['c4'][0] else '✗'} ({r['c4'][1]}) "
            f"| {'✓' if r['c5'][0] else '✗'} ({r['c5'][1].split('=')[-1].strip()}) |"
        )
    lines.append("")

    # Recomendação
    if valid_results:
        # Maior score; tie-break por proximidade a 8/255 (Hu 2024)
        max_score = max(r["score"] for r in valid_results)
        winners = [r for r in valid_results if r["score"] == max_score]
        winners.sort(key=lambda r: abs(r["eps_255"] - 8))  # tie-break: prefere 8
        winner = winners[0]
        lines.append(f"## ✅ Recomendação: **ε∞ = {winner['eps_255']}/255** ({winner['score']}/5 critérios)")
        if len(winners) > 1:
            others = [str(r["eps_255"]) for r in winners[1:]]
            lines.append(f"Empates: {others}/255 (tie-break aplicado: prefere ε mais próximo de 8/255 = Hu 2024).")
        lines.append("")
        lines.append("### Por ataque a ε escolhido")
        lines.append("")
        lines.append("| Ataque | mean ASR |")
        lines.append("|---|---|")
        for atk, asr in sorted(winner["asr_per_attack"].items(),
                                key=lambda x: -x[1]):
            lines.append(f"| {atk} | {asr:.3f} |")
        lines.append("")

    # Detalhes por ε
    lines.append("## Detalhes por ε")
    for r in valid_results:
        lines.append("")
        lines.append(f"### ε = {r['eps_255']}/255 — score {r['score']}/5")
        for k in ["c1", "c2", "c3", "c4", "c5"]:
            ok, detail = r[k]
            mark = "✓" if ok else "✗"
            lines.append(f"- {mark} **{k.upper()}**: {detail}")

    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--csv", type=Path, required=True,
                        help="CSV merged do probe (output de merge_probe_csvs.py)")
    parser.add_argument("--out", type=Path,
                        default=PROJECT_ROOT / "results" / "figures" / "probe_eps_curve_final" / "decide_epsilon_report.md",
                        help="Path do relatório markdown.")
    args = parser.parse_args()

    if not args.csv.exists():
        print(f"ERROR: CSV não encontrado: {args.csv}")
        return 1

    try:
        import pandas as pd
    except ImportError:
        print("ERROR: pandas necessário")
        return 1

    df = pd.read_csv(args.csv)
    if "eps_255" not in df.columns:
        df["eps_255"] = (df["epsilon"].astype(float) * 255).round().astype(int)

    print(f"Avaliando {len(CANDIDATE_EPS)} ε candidatos: {CANDIDATE_EPS}/255 ...")
    results = [evaluate_epsilon(df, e) for e in CANDIDATE_EPS]

    md = render_report(results, df)
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(md)

    print()
    print(md)
    print()
    print(f"✓ Relatório salvo: {args.out}")
    return 0


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