report_adversarial_stats.py

#!/usr/bin/env python3
"""Statistical analysis of adversarial transformation effectiveness.

Auto-detects all adversarial→source pairs in the database and reports
paired score differences with mean, std, Cohen's d, and p-values.

Breakdowns: per detector (overall), per category, per generator family.
Generator families group edit+t2i variants (e.g. grok + grok-t2i = "grok").
"""

import argparse
import re
import sqlite3
from collections import defaultdict

import numpy as np
from scipy import stats

import config

DB_PATH = f"{config.DATA_DIR}/images.db"


def infer_source_generator(adv_model: str) -> str:
    """Infer the source generator from an adversarial model name.

    adversarial-v4-grok -> grok
    adversarial-grok-t2i -> grok-t2i
    adversarial-v2-seedream -> seedream
    """
    name = adv_model
    # Strip "adversarial-" prefix
    name = re.sub(r"^adversarial-", "", name)
    # Strip version tag (v2-, v3-, v4-, etc.)
    name = re.sub(r"^v\d+-", "", name)
    return name


def generator_family(gen: str) -> str:
    """Map a generator to its family (strip -t2i suffix)."""
    return re.sub(r"-t2i$", "", gen)


def discover_pairs(conn, group=None):
    """Auto-discover all adversarial→source pairs with detection data.

    Returns list of (adv_model, source_generator) tuples.
    """
    query = """
        SELECT DISTINCT g_adv.model
        FROM generated_images g_adv
        JOIN images i ON g_adv.base_image_id = i.id
        WHERE g_adv.generator LIKE 'adversarial%'
    """
    params = []
    if group is not None:
        query += ' AND i."group" = ?'
        params.append(group)

    adv_models = [r[0] for r in conn.execute(query, params).fetchall()]

    # Get all source generators in the DB
    src_query = """
        SELECT DISTINCT generator FROM generated_images
        WHERE generator NOT LIKE 'adversarial%'
    """
    source_gens = {r[0] for r in conn.execute(src_query).fetchall()}

    pairs = []
    for adv_model in sorted(adv_models):
        src = infer_source_generator(adv_model)
        if src in source_gens:
            pairs.append((adv_model, src))
    return pairs


def get_paired_scores(conn, adv_model, source_gen, group=None):
    """Get paired (orig_score, adv_score) for each (base_image, detector).

    Truthscan variants are treated as interchangeable (matched via LIKE),
    with truthscan-noc2pa preferred over plain truthscan when both exist.

    Returns list of dicts with keys:
        base_image_id, category, detector, orig_score, adv_score
    """
    query = """
        SELECT
            g_orig.base_image_id,
            i.category,
            d_orig.detector as orig_detector,
            d_adv.detector as adv_detector,
            d_orig.ai_score as orig_score,
            d_adv.ai_score as adv_score
        FROM generated_images g_orig
        JOIN generated_images g_adv
            ON g_orig.base_image_id = g_adv.base_image_id
        JOIN detection_results d_orig
            ON d_orig.image_ref_id = g_orig.image_ref_id
        JOIN detection_results d_adv
            ON d_adv.image_ref_id = g_adv.image_ref_id
            AND (d_adv.detector = d_orig.detector
                 OR (d_orig.detector LIKE 'truthscan%'
                     AND d_adv.detector LIKE 'truthscan%'))
        JOIN images i ON g_orig.base_image_id = i.id
        WHERE g_orig.generator = ?
          AND g_adv.model = ?
    """
    params = [source_gen, adv_model]
    if group is not None:
        query += ' AND i."group" = ?'
        params.append(group)

    raw_rows = [dict(zip(
        ["base_image_id", "category", "orig_detector", "adv_detector",
         "orig_score", "adv_score"],
        row
    )) for row in conn.execute(query, params).fetchall()]

    # Dedup: for each (base_image_id, normalized_detector), prefer rows
    # where truthscan-noc2pa is used (on both sides if possible)
    seen = {}
    seen_priority = {}
    for r in raw_rows:
        norm_det = normalize_detector(r["orig_detector"])
        key = (r["base_image_id"], norm_det)
        priority = (
            (r["orig_detector"] == "truthscan-noc2pa")
            + (r["adv_detector"] == "truthscan-noc2pa")
        )
        if key not in seen or priority > seen_priority[key]:
            seen[key] = {
                "base_image_id": r["base_image_id"],
                "category": r["category"],
                "detector": norm_det,
                "orig_score": r["orig_score"],
                "adv_score": r["adv_score"],
            }
            seen_priority[key] = priority

    return list(seen.values())


def compute_stats(orig_scores, adv_scores):
    """Compute paired difference statistics."""
    orig = np.array(orig_scores)
    adv = np.array(adv_scores)
    deltas = adv - orig
    n = len(deltas)

    if n < 2:
        return None

    mean_orig = np.mean(orig)
    mean_adv = np.mean(adv)
    mean_delta = np.mean(deltas)
    std_delta = np.std(deltas, ddof=1)
    cohens_d = mean_delta / std_delta if std_delta > 0 else 0.0

    t_stat, p_value = stats.ttest_rel(adv, orig)

    return {
        "n": n,
        "mean_orig": mean_orig,
        "mean_adv": mean_adv,
        "mean_delta": mean_delta,
        "std_delta": std_delta,
        "cohens_d": cohens_d,
        "t_stat": t_stat,
        "p_value": p_value,
    }


def format_p(p):
    if p < 0.001:
        return "<.001"
    return f"{p:.3f}"


def print_stats_table(title, rows_by_key, key_width=16):
    """Print a stats table. rows_by_key: {label: stats_dict}."""
    header = (
        f"{'':>{key_width}} | {'N':>5} | {'Orig':>6} | {'Adv':>6} "
        f"| {'Delta':>7} | {'Std':>6} | {'d':>6} | {'p':>6}"
    )
    print(f"\n{'=' * len(header)}")
    print(title)
    print(f"{'=' * len(header)}")
    print(header)
    print("-" * len(header))

    for label, s in rows_by_key.items():
        if s is None:
            print(f"{label:>{key_width}} |   n<2")
            continue
        print(
            f"{label:>{key_width}} | {s['n']:>5} | {s['mean_orig']:>6.3f} | {s['mean_adv']:>6.3f} "
            f"| {s['mean_delta']:>+7.3f} | {s['std_delta']:>6.3f} | {s['cohens_d']:>+6.2f} "
            f"| {format_p(s['p_value']):>6}"
        )


def normalize_detector(det):
    """Normalize detector name (truthscan variants → truthscan)."""
    if det.startswith("truthscan"):
        return "truthscan"
    return det


def print_pair_report(conn, adv_model, source_gen, group=None):
    """Print full report for one adversarial→source pair."""
    rows = get_paired_scores(conn, adv_model, source_gen, group)
    if not rows:
        print(f"\n  No paired data for {source_gen} → {adv_model}")
        return

    family = generator_family(source_gen)
    print(f"\n{'#' * 100}")
    print(f"  {source_gen} → {adv_model}   (family: {family})")
    print(f"{'#' * 100}")

    # Group by detector
    by_detector = defaultdict(lambda: ([], []))
    # Group by (detector, category)
    by_det_cat = defaultdict(lambda: ([], []))

    for r in rows:
        det = r["detector"]
        cat = r["category"]
        by_detector[det][0].append(r["orig_score"])
        by_detector[det][1].append(r["adv_score"])
        by_det_cat[(det, cat)][0].append(r["orig_score"])
        by_det_cat[(det, cat)][1].append(r["adv_score"])

    # 1. Overall per detector
    det_stats = {}
    for det in sorted(by_detector):
        orig, adv = by_detector[det]
        det_stats[det] = compute_stats(orig, adv)
    print_stats_table("OVERALL (per detector)", det_stats)

    # 2. Per category
    categories = sorted({r["category"] for r in rows})
    for cat in categories:
        cat_stats = {}
        for det in sorted(by_detector):
            key = (det, cat)
            if key in by_det_cat:
                orig, adv = by_det_cat[key]
                cat_stats[det] = compute_stats(orig, adv)
        if cat_stats:
            print_stats_table(f"Category: {cat}", cat_stats)

    print(f"\n  * Delta = adv - orig (negative = scores reduced = evasion success)")
    print(f"  * d = Cohen's d effect size (paired)")
    print(f"  * p = paired t-test p-value")


def main():
    parser = argparse.ArgumentParser(
        description="Statistical analysis of adversarial score differences"
    )
    parser.add_argument(
        "--group", "-g", type=int, help="Filter to image group"
    )
    args = parser.parse_args()

    conn = sqlite3.connect(DB_PATH)

    pairs = discover_pairs(conn, group=args.group)
    if not pairs:
        print("No adversarial pairs found.")
        conn.close()
        return

    print(f"Discovered {len(pairs)} adversarial→source pairs")
    if args.group is not None:
        print(f"Filtering to group {args.group}")

    # Group pairs by adversarial version + generator family
    # e.g. (v4, grok) groups: grok→adversarial-v4-grok + grok-t2i→adversarial-v4-grok-t2i
    from itertools import groupby

    def family_key(pair):
        adv_model, source_gen = pair
        return (adv_model.replace(source_gen, generator_family(source_gen)),
                generator_family(source_gen))

    sorted_pairs = sorted(pairs, key=family_key)
    for (adv_family, src_family), group_pairs in groupby(sorted_pairs, key=family_key):
        group_pairs = list(group_pairs)

        # Print family-level combined report
        all_rows = []
        for adv_model, source_gen in group_pairs:
            rows = get_paired_scores(conn, adv_model, source_gen, group=args.group)
            all_rows.extend(rows)

        if not all_rows:
            continue

        pair_labels = ", ".join(f"{sg}→{am}" for am, sg in group_pairs)
        print(f"\n{'#' * 100}")
        print(f"  Family: {src_family}   [{pair_labels}]")
        print(f"{'#' * 100}")

        # Group by detector
        by_detector = defaultdict(lambda: ([], []))
        by_det_cat = defaultdict(lambda: ([], []))

        for r in all_rows:
            det = r["detector"]
            cat = r["category"]
            by_detector[det][0].append(r["orig_score"])
            by_detector[det][1].append(r["adv_score"])
            by_det_cat[(det, cat)][0].append(r["orig_score"])
            by_det_cat[(det, cat)][1].append(r["adv_score"])

        # Overall per detector
        det_stats = {}
        for det in sorted(by_detector):
            orig, adv = by_detector[det]
            det_stats[det] = compute_stats(orig, adv)
        print_stats_table("OVERALL (per detector)", det_stats)

        # Per category
        categories = sorted({r["category"] for r in all_rows})
        for cat in categories:
            cat_stats = {}
            for det in sorted(by_detector):
                key = (det, cat)
                if key in by_det_cat:
                    orig, adv = by_det_cat[key]
                    cat_stats[det] = compute_stats(orig, adv)
            if cat_stats:
                print_stats_table(f"Category: {cat}", cat_stats)

        print(f"\n  * Delta = adv - orig (negative = scores reduced = evasion success)")
        print(f"  * d = Cohen's d effect size (paired)")
        print(f"  * p = paired t-test p-value")

    conn.close()


if __name__ == "__main__":
    main()