Upload agents/sensor_agent.py with huggingface_hub

agents/sensor_agent.py

@@ -0,0 +1,239 @@
+"""
+FORENSIQ — Sensor Characteristics Agent
+Analyzes sensor physics violations:
+  - PRNU (Photo-Response Non-Uniformity) noise residual analysis
+  - Noise structure (Poisson-Gaussian model fit)
+  - Bayer demosaicing artifact detection
+"""
+
+import numpy as np
+from PIL import Image
+from scipy.ndimage import gaussian_filter, uniform_filter
+from scipy.signal import convolve2d
+from dataclasses import dataclass
+from typing import Dict, Any
+
+from agents.optical_agent import AgentEvidence
+
+
+# ─── PRNU Noise Residual ────────────────────────────────────────────
+def analyze_prnu(img: Image.Image) -> Dict[str, Any]:
+    """
+    Extract noise residual fingerprint.
+    Real cameras leave a unique PRNU pattern; AI images have uniform or random noise.
+    Inconsistent local noise variance → splicing / AI generation.
+    """
+    rgb = np.array(img.convert("RGB")).astype(np.float64)
+    
+    noise_residuals = []
+    for c in range(3):
+        channel = rgb[:, :, c]
+        denoised = gaussian_filter(channel, sigma=3.0)
+        residual = channel - denoised
+        noise_residuals.append(residual)
+
+    noise = np.stack(noise_residuals, axis=-1)
+    noise_energy = np.mean(noise ** 2, axis=-1)
+
+    # Local variance map (32x32 blocks)
+    local_var = uniform_filter(noise_energy, size=32)
+    
+    noise_std = float(np.std(local_var))
+    noise_mean = float(np.mean(local_var))
+    uniformity = 1.0 - min(noise_std / (noise_mean + 1e-9), 1.0)
+
+    # Correlation between channels (real sensors have correlated PRNU)
+    r_noise = noise_residuals[0].ravel()
+    g_noise = noise_residuals[1].ravel()
+    b_noise = noise_residuals[2].ravel()
+
+    # Subsample for speed
+    step = max(1, len(r_noise) // 100000)
+    rg_corr = float(np.corrcoef(r_noise[::step], g_noise[::step])[0, 1])
+    rb_corr = float(np.corrcoef(r_noise[::step], b_noise[::step])[0, 1])
+
+    # Real cameras: correlated noise residuals; AI: uncorrelated
+    avg_corr = (rg_corr + rb_corr) / 2
+
+    if uniformity > 0.7 and avg_corr > 0.3:
+        score = -0.4
+        note = "Consistent sensor noise pattern with correlated channels (real camera)"
+    elif uniformity < 0.4:
+        score = 0.5
+        note = "Inconsistent noise regions suggest splicing or AI generation"
+    elif avg_corr < 0.1:
+        score = 0.4
+        note = "Uncorrelated channel noise (atypical for real cameras)"
+    else:
+        score = 0.1
+        note = "Moderate noise consistency"
+
+    return {
+        "test": "PRNU Noise Residual",
+        "noise_uniformity": round(uniformity, 4),
+        "noise_mean": round(noise_mean, 4),
+        "rg_correlation": round(rg_corr, 4),
+        "rb_correlation": round(rb_corr, 4),
+        "score": score,
+        "note": note,
+        "noise_map": noise_energy,
+    }
+
+
+# ─── Noise Structure (Poisson-Gaussian Model) ──────────────────────
+def analyze_noise_structure(img: Image.Image) -> Dict[str, Any]:
+    """
+    Real sensor noise follows σ² = σ²_read + k·I (Poisson-Gaussian).
+    AI images lack this physical noise model.
+    """
+    rgb = np.array(img.convert("RGB")).astype(np.float64)
+    gray = np.mean(rgb, axis=-1)
+
+    # Compute local mean and local variance in blocks
+    block_size = 16
+    h, w = gray.shape
+    h_crop, w_crop = (h // block_size) * block_size, (w // block_size) * block_size
+    gray = gray[:h_crop, :w_crop]
+
+    intensities = []
+    variances = []
+
+    for i in range(0, h_crop, block_size):
+        for j in range(0, w_crop, block_size):
+            block = gray[i:i + block_size, j:j + block_size]
+            intensities.append(float(np.mean(block)))
+            variances.append(float(np.var(block)))
+
+    intensities = np.array(intensities)
+    variances = np.array(variances)
+
+    # Filter out extreme blocks
+    valid = (intensities > 10) & (intensities < 245) & (variances > 0)
+    if np.sum(valid) < 20:
+        return {
+            "test": "Noise Structure (Poisson-Gaussian)",
+            "score": 0.0,
+            "note": "Insufficient data for noise model fitting",
+        }
+
+    I = intensities[valid]
+    V = variances[valid]
+
+    # Fit linear model: V = a + b*I (Poisson-Gaussian)
+    try:
+        coeffs = np.polyfit(I, V, 1)
+        fitted = np.polyval(coeffs, I)
+        residual = float(np.mean((V - fitted) ** 2))
+        r_squared = 1.0 - residual / (np.var(V) + 1e-9)
+    except Exception:
+        r_squared = 0.0
+
+    if r_squared > 0.5:
+        score = -0.3
+        note = f"Noise follows Poisson-Gaussian model (R²={r_squared:.3f}, real sensor)"
+    elif r_squared < 0.1:
+        score = 0.5
+        note = f"Noise does NOT follow sensor physics (R²={r_squared:.3f}, AI-like)"
+    else:
+        score = 0.15
+        note = f"Weak Poisson-Gaussian fit (R²={r_squared:.3f})"
+
+    return {
+        "test": "Noise Structure (Poisson-Gaussian)",
+        "r_squared": round(r_squared, 4),
+        "slope": round(float(coeffs[0]), 6) if r_squared > 0 else None,
+        "intercept": round(float(coeffs[1]), 4) if r_squared > 0 else None,
+        "score": score,
+        "note": note,
+    }
+
+
+# ─── Bayer Demosaicing Artifacts ────────────────────────────────────
+def analyze_bayer_demosaicing(img: Image.Image) -> Dict[str, Any]:
+    """
+    Real cameras: green channel has ~2x samples → lower noise than R/B.
+    Expected: σ_green < σ_red ≈ σ_blue.
+    AI images lack this Bayer pattern artifact.
+    """
+    rgb = np.array(img.convert("RGB")).astype(np.float64)
+    
+    # High-frequency noise per channel
+    noise_std = {}
+    for c, name in enumerate(["red", "green", "blue"]):
+        channel = rgb[:, :, c]
+        denoised = gaussian_filter(channel, sigma=1.5)
+        noise = channel - denoised
+        noise_std[name] = float(np.std(noise))
+
+    green_lower = noise_std["green"] < min(noise_std["red"], noise_std["blue"])
+    rb_similar = abs(noise_std["red"] - noise_std["blue"]) / (
+        max(noise_std["red"], noise_std["blue"]) + 1e-9
+    ) < 0.2
+
+    if green_lower and rb_similar:
+        score = -0.4
+        note = (
+            f"Bayer pattern detected: σ_green({noise_std['green']:.3f}) < "
+            f"σ_red({noise_std['red']:.3f}) ≈ σ_blue({noise_std['blue']:.3f})"
+        )
+    elif green_lower:
+        score = -0.2
+        note = "Green channel is quieter but R/B difference is large"
+    else:
+        score = 0.4
+        note = (
+            f"No Bayer pattern: σ_green({noise_std['green']:.3f}), "
+            f"σ_red({noise_std['red']:.3f}), σ_blue({noise_std['blue']:.3f})"
+        )
+
+    return {
+        "test": "Bayer Demosaicing Artifacts",
+        "noise_red": round(noise_std["red"], 4),
+        "noise_green": round(noise_std["green"], 4),
+        "noise_blue": round(noise_std["blue"], 4),
+        "green_is_lower": green_lower,
+        "rb_similar": rb_similar,
+        "score": score,
+        "note": note,
+    }
+
+
+# ─── Main Agent Entry Point ─────────────────────────────────────────
+def run_sensor_agent(img: Image.Image) -> AgentEvidence:
+    """Run all sensor characteristic tests."""
+    findings = []
+    scores = []
+
+    for fn in [analyze_prnu, analyze_noise_structure, analyze_bayer_demosaicing]:
+        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"Sensor violations: {', '.join(violations)}."
+    elif compliant:
+        rationale = f"Sensor physics consistent: {', '.join(compliant)}."
+    else:
+        rationale = "Sensor analysis inconclusive."
+
+    for f in findings:
+        if f.get("note"):
+            rationale += f" [{f['test']}]: {f['note']}."
+
+    return AgentEvidence(
+        agent_name="Sensor Characteristics 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,
+    )