#!/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())