Upload agents/model_agent.py with huggingface_hub

agents/model_agent.py

@@ -0,0 +1,254 @@
+"""
+FORENSIQ — Generative Model Agent
+Detects architecture-specific signatures:
+  - Frequency grid artifacts (8×8, 16×16 periodic patterns from GANs)
+  - Diffusion residuals (spectral notches at step noise harmonics)
+  - Model fingerprinting (frequency-domain generator attribution)
+"""
+
+import numpy as np
+from PIL import Image
+from scipy.signal import find_peaks
+from scipy.ndimage import gaussian_filter
+from typing import Dict, Any
+
+from agents.optical_agent import AgentEvidence
+
+
+# ─── Frequency Grid Artifacts ────────────────────────────────────────
+def analyze_frequency_grid(img: Image.Image) -> Dict[str, Any]:
+    """
+    GAN upsampling (deconv layers) creates periodic grid patterns 
+    visible in 2D FFT as spectral peaks at multiples of 8×8 or 16×16.
+    """
+    gray = np.array(img.convert("L")).astype(np.float64)
+    h, w = gray.shape
+
+    # 2D FFT
+    fft = np.fft.fft2(gray)
+    fft_shift = np.fft.fftshift(fft)
+    magnitude = np.log(np.abs(fft_shift) + 1)
+
+    # Center row/column profiles
+    cy, cx = h // 2, w // 2
+    row_profile = magnitude[cy, :]
+    col_profile = magnitude[:, cx]
+
+    # Find periodic peaks (GAN artifacts show at multiples of N/8, N/16)
+    row_peaks, row_props = find_peaks(row_profile, distance=5, prominence=0.3)
+    col_peaks, col_props = find_peaks(col_profile, distance=5, prominence=0.3)
+
+    # Check for regular spacing (grid artifact signature)
+    def check_periodic(peaks, size):
+        if len(peaks) < 3:
+            return 0.0, []
+        spacings = np.diff(sorted(peaks))
+        # Expected spacing for 8×8 grid: size/8
+        expected_8 = size / 8
+        expected_16 = size / 16
+        matches_8 = np.sum(np.abs(spacings - expected_8) < expected_8 * 0.15)
+        matches_16 = np.sum(np.abs(spacings - expected_16) < expected_16 * 0.15)
+        best = max(matches_8, matches_16)
+        return float(best / max(len(spacings), 1)), spacings.tolist()
+
+    row_periodic, row_spacings = check_periodic(row_peaks, w)
+    col_periodic, col_spacings = check_periodic(col_peaks, h)
+    grid_score = (row_periodic + col_periodic) / 2
+
+    # High-frequency vs mid-frequency energy ratio
+    hf_ring = magnitude.copy()
+    hf_ring[cy - h // 8:cy + h // 8, cx - w // 8:cx + w // 8] = 0
+    hf_energy = float(np.mean(hf_ring))
+    mf_mask = np.zeros_like(magnitude)
+    mf_mask[cy - h // 4:cy + h // 4, cx - w // 4:cx + w // 4] = 1
+    mf_mask[cy - h // 8:cy + h // 8, cx - w // 8:cx + w // 8] = 0
+    mf_energy = float(np.mean(magnitude * mf_mask))
+
+    if grid_score > 0.4:
+        score = 0.7
+        note = f"Periodic grid artifacts detected (GAN signature, periodicity={grid_score:.2f})"
+    elif grid_score > 0.2:
+        score = 0.3
+        note = f"Weak periodic patterns (possible GAN artifacts, periodicity={grid_score:.2f})"
+    else:
+        score = -0.2
+        note = "No periodic grid artifacts (natural frequency spectrum)"
+
+    return {
+        "test": "Frequency Grid Artifacts",
+        "grid_periodicity_score": round(grid_score, 4),
+        "row_peaks": len(row_peaks),
+        "col_peaks": len(col_peaks),
+        "hf_energy": round(hf_energy, 4),
+        "mf_energy": round(mf_energy, 4),
+        "score": score,
+        "note": note,
+        "magnitude_spectrum": magnitude,
+    }
+
+
+# ─── Diffusion Residuals ────────────────────────────────────────────
+def analyze_diffusion_residuals(img: Image.Image) -> Dict[str, Any]:
+    """
+    Diffusion models leave characteristic spectral notches and
+    autocorrelation patterns from the denoising step schedule.
+    """
+    gray = np.array(img.convert("L")).astype(np.float64)
+
+    # Compute radial power spectrum
+    fft = np.fft.fft2(gray)
+    fft_shift = np.fft.fftshift(fft)
+    power = np.abs(fft_shift) ** 2
+
+    h, w = power.shape
+    cy, cx = h // 2, w // 2
+    Y, X = np.mgrid[0:h, 0:w]
+    R = np.sqrt((X - cx) ** 2 + (Y - cy) ** 2).astype(int)
+
+    # Radially averaged power spectrum
+    max_r = min(cy, cx)
+    radial_power = np.zeros(max_r)
+    counts = np.zeros(max_r)
+    for r in range(max_r):
+        mask = R == r
+        if mask.any():
+            radial_power[r] = np.mean(power[mask])
+            counts[r] = np.sum(mask)
+
+    radial_power = np.log(radial_power + 1)
+
+    # Natural images: power ∝ 1/f² → linear decrease in log-log
+    freqs = np.arange(1, max_r)
+    log_freqs = np.log(freqs + 1)
+    log_power = radial_power[1:max_r]
+
+    # Fit linear model (1/f² slope)
+    if len(log_freqs) > 10:
+        coeffs = np.polyfit(log_freqs, log_power, 1)
+        fitted = np.polyval(coeffs, log_freqs)
+        residuals = log_power - fitted
+
+        # Diffusion models: spectral notches (negative dips in residuals)
+        notches, _ = find_peaks(-residuals, prominence=0.5)
+        
+        # Smoothness of spectral rolloff
+        smoothness = float(np.std(residuals))
+        slope = float(coeffs[0])
+    else:
+        notches = []
+        smoothness = 1.0
+        slope = 0.0
+
+    # Natural 1/f² slope is ~ -2 in log-log
+    slope_deviation = abs(slope - (-2.0))
+
+    if len(notches) > 3:
+        score = 0.5
+        note = f"Spectral notches detected ({len(notches)} notches, diffusion signature)"
+    elif smoothness > 1.0 or slope_deviation > 1.5:
+        score = 0.3
+        note = f"Unnatural spectral rolloff (slope={slope:.2f}, deviation={slope_deviation:.2f})"
+    elif smoothness < 0.5 and slope_deviation < 0.5:
+        score = -0.3
+        note = f"Natural 1/f² spectral profile (slope={slope:.2f})"
+    else:
+        score = 0.1
+        note = f"Mild spectral deviation (slope={slope:.2f})"
+
+    return {
+        "test": "Diffusion Residuals",
+        "spectral_notches": len(notches),
+        "spectral_smoothness": round(smoothness, 4),
+        "slope": round(slope, 4),
+        "slope_deviation": round(slope_deviation, 4),
+        "score": score,
+        "note": note,
+    }
+
+
+# ─── Model Fingerprinting ───────────────────────────────────────────
+def analyze_model_fingerprint(img: Image.Image) -> Dict[str, Any]:
+    """
+    Different generators leave unique frequency-domain signatures.
+    Analyzes autocorrelation and spectral texture for attribution.
+    """
+    gray = np.array(img.convert("L")).astype(np.float64)
+
+    # Autocorrelation map
+    fft = np.fft.fft2(gray)
+    power = np.abs(fft) ** 2
+    autocorr = np.real(np.fft.ifft2(power))
+    autocorr = np.fft.fftshift(autocorr)
+    autocorr = autocorr / (autocorr.max() + 1e-9)
+
+    h, w = autocorr.shape
+    cy, cx = h // 2, w // 2
+
+    # Check for periodic peaks in autocorrelation (GAN checkerboard)
+    # Exclude center peak
+    autocorr_masked = autocorr.copy()
+    r_exclude = max(h, w) // 20
+    Y, X = np.mgrid[0:h, 0:w]
+    center_mask = ((X - cx) ** 2 + (Y - cy) ** 2) < r_exclude ** 2
+    autocorr_masked[center_mask] = 0
+
+    max_secondary = float(np.max(autocorr_masked))
+    
+    # High secondary peak = repetitive structure (GAN artifact)
+    if max_secondary > 0.3:
+        score = 0.6
+        note = f"Strong autocorrelation secondary peak ({max_secondary:.3f}) — GAN checkerboard pattern"
+    elif max_secondary > 0.15:
+        score = 0.3
+        note = f"Moderate autocorrelation peak ({max_secondary:.3f}) — possible generator artifacts"
+    else:
+        score = -0.2
+        note = f"Natural autocorrelation structure (peak={max_secondary:.3f})"
+
+    return {
+        "test": "Model Fingerprinting",
+        "max_secondary_peak": round(max_secondary, 4),
+        "score": score,
+        "note": note,
+    }
+
+
+# ─── Main Agent Entry Point ─────────────────────────────────────────
+def run_model_agent(img: Image.Image) -> AgentEvidence:
+    """Run all generative model detection tests."""
+    findings = []
+    scores = []
+
+    for fn in [analyze_frequency_grid, analyze_diffusion_residuals, analyze_model_fingerprint]:
+        try:
+            result = fn(img)
+            findings.append(result)
+            scores.append(result["score"])
+        except Exception as e:
+            findings.append({"test": fn.__name__, "error": str(e), "score": 0})
+
+    avg_score = float(np.mean(scores)) if scores else 0.0
+    confidence = min(1.0, 0.5 + 0.5 * abs(avg_score))
+
+    violations = [f["test"] for f in findings if f.get("score", 0) > 0.2]
+    compliant = [f["test"] for f in findings if f.get("score", 0) < -0.1]
+
+    if violations:
+        rationale = f"Generative model signatures detected: {', '.join(violations)}."
+    elif compliant:
+        rationale = f"No generator artifacts found: {', '.join(compliant)}."
+    else:
+        rationale = "Generator analysis inconclusive."
+
+    for f in findings:
+        if f.get("note"):
+            rationale += f" [{f['test']}]: {f['note']}."
+
+    return AgentEvidence(
+        agent_name="Generative Model Agent",
+        violation_score=np.clip(avg_score, -1, 1),
+        confidence=confidence,
+        failure_prob=max(0.0, 1.0 - len(scores) / 3),
+        rationale=rationale,
+        sub_findings=findings,
+    )