Upload agents/optical_agent.py with huggingface_hub

agents/optical_agent.py

@@ -0,0 +1,358 @@
+"""
+FORENSIQ — Optical Physics Agent
+Tests violations of lens and optical physics:
+  - Chromatic Aberration analysis (RGB channel misregistration)
+  - Vignetting analysis (cos⁴θ intensity falloff)
+  - Depth-of-Field consistency (expected vs measured blur)
+  - Bokeh microstructure (aperture blade FFT signatures)
+"""
+
+import numpy as np
+from PIL import Image
+from scipy.ndimage import sobel, gaussian_filter, uniform_filter
+from scipy.signal import find_peaks
+from dataclasses import dataclass, field
+from typing import List, Dict, Any, Optional
+
+
+@dataclass
+class AgentEvidence:
+    """Structured evidence output from any forensic agent."""
+    agent_name: str
+    violation_score: float  # -1 (authentic) to +1 (fake)
+    confidence: float       # 0 to 1
+    failure_prob: float     # probability this agent's analysis failed
+    rationale: str
+    sub_findings: List[Dict[str, Any]] = field(default_factory=list)
+    visual_evidence: Optional[Any] = None  # numpy array or PIL image
+
+
+def _to_gray(img: Image.Image) -> np.ndarray:
+    return np.array(img.convert("L")).astype(np.float64)
+
+
+def _to_rgb(img: Image.Image) -> np.ndarray:
+    return np.array(img.convert("RGB")).astype(np.float64)
+
+
+# ─── Chromatic Aberration ────────────────────────────────────────────
+def analyze_chromatic_aberration(img: Image.Image) -> Dict[str, Any]:
+    """
+    Real lenses create slight spatial offsets between R/G/B edges.
+    AI images have perfectly aligned or impossibly misaligned channels.
+    """
+    rgb = _to_rgb(img)
+    r, g, b = rgb[:, :, 0], rgb[:, :, 1], rgb[:, :, 2]
+
+    # Edge magnitude per channel
+    edges = {}
+    for name, ch in [("R", r), ("G", g), ("B", b)]:
+        ex = sobel(ch, axis=0)
+        ey = sobel(ch, axis=1)
+        edges[name] = np.hypot(ex, ey)
+
+    # Flatten for correlation
+    er, eg, eb = edges["R"].ravel(), edges["G"].ravel(), edges["B"].ravel()
+
+    rg_corr = float(np.corrcoef(er, eg)[0, 1])
+    rb_corr = float(np.corrcoef(er, eb)[0, 1])
+    gb_corr = float(np.corrcoef(eg, eb)[0, 1])
+
+    avg_corr = (rg_corr + rb_corr + gb_corr) / 3.0
+
+    # Natural range: moderate correlation (0.80–0.97)
+    # Too perfect (>0.99) = synthetic; too low (<0.70) = heavy editing
+    if avg_corr > 0.99:
+        score = 0.6
+        note = "Suspiciously perfect channel alignment (no natural CA)"
+    elif avg_corr < 0.70:
+        score = 0.5
+        note = "Abnormally low channel correlation (heavy manipulation)"
+    elif 0.80 <= avg_corr <= 0.97:
+        score = -0.4
+        note = "Natural chromatic aberration pattern detected"
+    else:
+        score = 0.2
+        note = "Borderline chromatic aberration"
+
+    return {
+        "test": "Chromatic Aberration",
+        "rg_correlation": round(rg_corr, 4),
+        "rb_correlation": round(rb_corr, 4),
+        "gb_correlation": round(gb_corr, 4),
+        "avg_correlation": round(avg_corr, 4),
+        "score": score,
+        "note": note,
+    }
+
+
+# ─── Vignetting ─────────────────────────────────────────────────────
+def analyze_vignetting(img: Image.Image) -> Dict[str, Any]:
+    """
+    Real photos show cos⁴(θ) intensity falloff from center.
+    AI images often have flat or unnatural brightness profiles.
+    """
+    gray = _to_gray(img)
+    h, w = gray.shape
+    cy, cx = h / 2, w / 2
+
+    # Radial distance for every pixel
+    Y, X = np.mgrid[0:h, 0:w]
+    R = np.sqrt((X - cx) ** 2 + (Y - cy) ** 2)
+    R_max = np.sqrt(cx ** 2 + cy ** 2)
+    R_norm = R / R_max  # 0 at center, 1 at corner
+
+    # Bin by radial distance and compute mean intensity
+    n_bins = 20
+    bin_edges = np.linspace(0, 1, n_bins + 1)
+    radial_means = []
+    for i in range(n_bins):
+        mask = (R_norm >= bin_edges[i]) & (R_norm < bin_edges[i + 1])
+        if mask.any():
+            radial_means.append(float(np.mean(gray[mask])))
+        else:
+            radial_means.append(0)
+
+    radial_means = np.array(radial_means)
+    if radial_means[0] == 0:
+        radial_means[0] = 1.0
+
+    # Expected cos⁴(θ) model: I ∝ cos⁴(arctan(r/f))
+    # Simplified: intensity should decrease toward edges
+    normalized = radial_means / (radial_means[0] + 1e-9)
+
+    # Fit cos⁴ model
+    r_centers = (bin_edges[:-1] + bin_edges[1:]) / 2
+    theta = np.arctan(r_centers * 1.5)  # ~35° max field angle
+    cos4_model = np.cos(theta) ** 4
+
+    residual = float(np.mean((normalized - cos4_model) ** 2))
+
+    # Monotonic decrease check
+    diffs = np.diff(normalized)
+    non_decreasing_frac = float(np.sum(diffs > 0.02) / len(diffs))
+
+    if residual < 0.01 and non_decreasing_frac < 0.3:
+        score = -0.3
+        note = "Natural vignetting pattern (cos⁴θ consistent)"
+    elif residual > 0.05 or non_decreasing_frac > 0.5:
+        score = 0.4
+        note = "Unnatural brightness profile (vignetting absent or inconsistent)"
+    else:
+        score = 0.1
+        note = "Mild vignetting deviation"
+
+    return {
+        "test": "Vignetting (cos⁴θ)",
+        "cos4_residual": round(residual, 5),
+        "non_decreasing_fraction": round(non_decreasing_frac, 3),
+        "score": score,
+        "note": note,
+        "radial_profile": normalized.tolist(),
+    }
+
+
+# ─── Depth of Field Consistency ─────────────────────────────────────
+def analyze_dof_consistency(img: Image.Image) -> Dict[str, Any]:
+    """
+    Real depth-of-field blur varies smoothly with distance.
+    AI images often have inconsistent blur regions.
+    """
+    gray = _to_gray(img)
+    h, w = gray.shape
+
+    # Local blur estimation via Laplacian variance in blocks
+    block_size = max(h, w) // 16
+    if block_size < 8:
+        block_size = 8
+
+    blur_map = np.zeros((h // block_size, w // block_size))
+    for bi in range(blur_map.shape[0]):
+        for bj in range(blur_map.shape[1]):
+            y0, y1 = bi * block_size, (bi + 1) * block_size
+            x0, x1 = bj * block_size, (bj + 1) * block_size
+            block = gray[y0:y1, x0:x1]
+            # Laplacian variance = sharpness metric
+            lap = np.array([[0, 1, 0], [1, -4, 1], [0, 1, 0]], dtype=np.float64)
+            from scipy.signal import convolve2d
+            laplacian = convolve2d(block, lap, mode="valid")
+            blur_map[bi, bj] = float(np.var(laplacian))
+
+    # Smooth blur map
+    if blur_map.size > 1:
+        smooth_blur = gaussian_filter(blur_map, sigma=1.0)
+        # Inconsistency = high-frequency variation in blur map
+        blur_residual = blur_map - smooth_blur
+        inconsistency = float(np.std(blur_residual) / (np.mean(blur_map) + 1e-9))
+    else:
+        inconsistency = 0.0
+
+    if inconsistency < 0.3:
+        score = -0.3
+        note = "Smooth depth-of-field gradient (consistent with real optics)"
+    elif inconsistency > 0.7:
+        score = 0.5
+        note = "Abrupt blur transitions (inconsistent DoF, possible manipulation)"
+    else:
+        score = 0.1
+        note = "Moderate DoF variation"
+
+    return {
+        "test": "Depth-of-Field Consistency",
+        "blur_inconsistency": round(inconsistency, 4),
+        "score": score,
+        "note": note,
+        "blur_map": blur_map,
+    }
+
+
+# ─── Bokeh Microstructure ───────────────────────────────────────────
+def analyze_bokeh(img: Image.Image) -> Dict[str, Any]:
+    """
+    Real bokeh shows aperture blade polygon structure in Fourier domain.
+    AI bokeh is typically smooth circles without blade structure.
+    """
+    gray = _to_gray(img)
+
+    # Find bright point-like regions (potential bokeh)
+    threshold = np.percentile(gray, 97)
+    bright_mask = gray > threshold
+
+    if np.sum(bright_mask) < 100:
+        return {
+            "test": "Bokeh Microstructure",
+            "score": 0.0,
+            "note": "Insufficient bright highlights for bokeh analysis",
+            "bokeh_found": False,
+        }
+
+    # Extract largest bright region
+    from scipy.ndimage import label
+    labeled, n_features = label(bright_mask)
+    if n_features == 0:
+        return {
+            "test": "Bokeh Microstructure",
+            "score": 0.0,
+            "note": "No bokeh regions detected",
+            "bokeh_found": False,
+        }
+
+    # Get largest blob
+    sizes = [np.sum(labeled == i) for i in range(1, n_features + 1)]
+    largest = np.argmax(sizes) + 1
+    blob_mask = labeled == largest
+
+    # Crop around blob
+    ys, xs = np.where(blob_mask)
+    y0, y1, x0, x1 = ys.min(), ys.max(), xs.min(), xs.max()
+    patch = gray[y0:y1 + 1, x0:x1 + 1]
+
+    if patch.shape[0] < 8 or patch.shape[1] < 8:
+        return {
+            "test": "Bokeh Microstructure",
+            "score": 0.0,
+            "note": "Bokeh regions too small for analysis",
+            "bokeh_found": False,
+        }
+
+    # FFT of patch — look for angular structure (blade signatures)
+    fft = np.fft.fft2(patch)
+    fft_shift = np.fft.fftshift(fft)
+    magnitude = np.log(np.abs(fft_shift) + 1)
+
+    # Angular variance in FFT (high = blade structure, low = smooth circle)
+    cy, cx = magnitude.shape[0] // 2, magnitude.shape[1] // 2
+    angles = np.arctan2(
+        np.mgrid[0:magnitude.shape[0], 0:magnitude.shape[1]][0] - cy,
+        np.mgrid[0:magnitude.shape[0], 0:magnitude.shape[1]][1] - cx,
+    )
+    n_angular_bins = 12
+    angular_profile = []
+    for k in range(n_angular_bins):
+        a0 = -np.pi + k * (2 * np.pi / n_angular_bins)
+        a1 = a0 + (2 * np.pi / n_angular_bins)
+        amask = (angles >= a0) & (angles < a1)
+        angular_profile.append(float(np.mean(magnitude[amask])) if amask.any() else 0)
+
+    angular_var = float(np.var(angular_profile))
+
+    if angular_var > 0.1:
+        score = -0.2
+        note = "Aperture blade structure detected in bokeh (consistent with real lens)"
+    else:
+        score = 0.3
+        note = "Smooth circular bokeh without blade structure (possible AI generation)"
+
+    return {
+        "test": "Bokeh Microstructure",
+        "angular_variance": round(angular_var, 4),
+        "score": score,
+        "note": note,
+        "bokeh_found": True,
+    }
+
+
+# ─── Main Agent Entry Point ─────────────────────────────────────────
+def run_optical_agent(img: Image.Image) -> AgentEvidence:
+    """Run all optical physics tests and produce structured evidence."""
+    findings = []
+    scores = []
+
+    try:
+        ca = analyze_chromatic_aberration(img)
+        findings.append(ca)
+        scores.append(ca["score"])
+    except Exception as e:
+        findings.append({"test": "Chromatic Aberration", "error": str(e), "score": 0})
+
+    try:
+        vig = analyze_vignetting(img)
+        findings.append(vig)
+        scores.append(vig["score"])
+    except Exception as e:
+        findings.append({"test": "Vignetting", "error": str(e), "score": 0})
+
+    try:
+        dof = analyze_dof_consistency(img)
+        findings.append(dof)
+        scores.append(dof["score"])
+    except Exception as e:
+        findings.append({"test": "DoF Consistency", "error": str(e), "score": 0})
+
+    try:
+        bokeh = analyze_bokeh(img)
+        findings.append(bokeh)
+        scores.append(bokeh["score"])
+    except Exception as e:
+        findings.append({"test": "Bokeh Microstructure", "error": str(e), "score": 0})
+
+    if scores:
+        avg_score = float(np.mean(scores))
+        confidence = min(1.0, 0.5 + 0.5 * abs(avg_score))
+    else:
+        avg_score = 0.0
+        confidence = 0.1
+
+    # Build rationale
+    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"Optical violations detected in: {', '.join(violations)}."
+    elif compliant:
+        rationale = f"Optical physics consistent: {', '.join(compliant)}."
+    else:
+        rationale = "Optical analysis inconclusive."
+
+    for f in findings:
+        if f.get("note"):
+            rationale += f" [{f['test']}]: {f['note']}."
+
+    return AgentEvidence(
+        agent_name="Optical Physics Agent",
+        violation_score=np.clip(avg_score, -1, 1),
+        confidence=confidence,
+        failure_prob=max(0.0, 1.0 - len(scores) / 4),
+        rationale=rationale,
+        sub_findings=findings,
+    )