Upload agents/optical_agent.py with huggingface_hub

agents/optical_agent.py

@@ -292,39 +292,264 @@ def analyze_bokeh(img: Image.Image) -> Dict[str, Any]:
     }
 
 
+# ─── Lens Distortion Analysis ────────────────────────────────────────
+def analyze_lens_distortion(img: Image.Image) -> Dict[str, Any]:
+    """
+    Real lenses produce barrel/pincushion distortion following Brown-Conrady model.
+    AI images often have perfectly rectilinear geometry or impossible distortion.
+    """
+    gray = _to_gray(img)
+    h, w = gray.shape
+
+    # Edge detection
+    ex = sobel(gray, axis=1)
+    ey = sobel(gray, axis=0)
+    edge_mag = np.hypot(ex, ey)
+
+    # Threshold strong edges
+    threshold = np.percentile(edge_mag, 90)
+    strong_edges = edge_mag > threshold
+
+    # Analyze edge straightness in radial bands
+    cy, cx = h / 2, w / 2
+    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
+
+    # Compare edge density at different radial distances
+    inner_edges = float(np.mean(strong_edges[R_norm < 0.3]))
+    mid_edges = float(np.mean(strong_edges[(R_norm >= 0.3) & (R_norm < 0.7)]))
+    outer_edges = float(np.mean(strong_edges[R_norm >= 0.7]))
+
+    # Real lenses: edges slightly softer at corners due to distortion
+    # AI: uniform edge sharpness across frame
+    edge_ratio = outer_edges / (inner_edges + 1e-9)
+
+    if 0.5 < edge_ratio < 0.9:
+        score = -0.3
+        note = f"Natural edge falloff at periphery (ratio={edge_ratio:.3f}, lens distortion present)"
+    elif edge_ratio > 0.95:
+        score = 0.3
+        note = f"Unnaturally uniform edges across frame (ratio={edge_ratio:.3f}, no lens distortion)"
+    else:
+        score = 0.1
+        note = f"Edge distribution ratio={edge_ratio:.3f}"
+
+    return {
+        "test": "Lens Distortion",
+        "edge_ratio_outer_inner": round(edge_ratio, 4),
+        "inner_edge_density": round(inner_edges, 4),
+        "outer_edge_density": round(outer_edges, 4),
+        "score": score,
+        "note": note,
+    }
+
+
+# ─── CA Radial Pattern Analysis ─────────────────────────────────────
+def analyze_ca_radial_pattern(img: Image.Image) -> Dict[str, Any]:
+    """
+    Real chromatic aberration increases radially from center (more at corners).
+    AI images have spatially uniform or random channel misregistration.
+    """
+    rgb = _to_rgb(img)
+    h, w, _ = rgb.shape
+    cy, cx = h / 2, w / 2
+
+    r, g, b = rgb[:, :, 0], rgb[:, :, 1], rgb[:, :, 2]
+
+    # Compute local channel difference in blocks
+    block_size = max(32, min(h, w) // 8)
+    center_diffs = []
+    edge_diffs = []
+
+    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)
+
+    for bi in range(0, h - block_size, block_size):
+        for bj in range(0, w - block_size, block_size):
+            block_r = r[bi:bi + block_size, bj:bj + block_size]
+            block_g = g[bi:bi + block_size, bj:bj + block_size]
+            block_b = b[bi:bi + block_size, bj:bj + block_size]
+
+            # Local RG difference as proxy for CA
+            rg_diff = float(np.std(block_r - block_g))
+            rb_diff = float(np.std(block_r - block_b))
+            ca_magnitude = (rg_diff + rb_diff) / 2
+
+            block_center_r = R[bi + block_size // 2, bj + block_size // 2] / R_max
+
+            if block_center_r < 0.4:
+                center_diffs.append(ca_magnitude)
+            elif block_center_r > 0.6:
+                edge_diffs.append(ca_magnitude)
+
+    if center_diffs and edge_diffs:
+        center_ca = float(np.mean(center_diffs))
+        edge_ca = float(np.mean(edge_diffs))
+        ca_increase = edge_ca / (center_ca + 1e-9)
+
+        # Real lenses: CA increases toward edges (ratio > 1.1)
+        if ca_increase > 1.15:
+            score = -0.3
+            note = f"CA increases radially (edge/center={ca_increase:.2f}, natural lens behavior)"
+        elif ca_increase < 0.9:
+            score = 0.3
+            note = f"CA decreases toward edges (ratio={ca_increase:.2f}, unnatural)"
+        else:
+            score = 0.1
+            note = f"Flat CA distribution (ratio={ca_increase:.2f})"
+    else:
+        ca_increase = 1.0
+        score = 0.0
+        note = "Insufficient data for radial CA analysis"
+
+    return {
+        "test": "CA Radial Pattern",
+        "ca_edge_center_ratio": round(ca_increase, 4),
+        "score": score,
+        "note": note,
+    }
+
+
+# ─── Specular Reflection Map ────────────────────────────────────────
+def analyze_specular_reflections(img: Image.Image) -> Dict[str, Any]:
+    """
+    Real specular reflections follow Phong/Blinn-Phong model with
+    consistent highlight shapes. AI often has inconsistent specularity.
+    """
+    rgb = _to_rgb(img)
+    gray = np.mean(rgb, axis=-1)
+
+    # Detect specular highlights (very bright, near-white pixels)
+    highlight_threshold = np.percentile(gray, 98)
+    highlight_mask = gray > highlight_threshold
+
+    # Compute saturation (low saturation = specular)
+    max_c = np.max(rgb, axis=-1)
+    min_c = np.min(rgb, axis=-1)
+    saturation = (max_c - min_c) / (max_c + 1e-9)
+
+    specular_mask = highlight_mask & (saturation < 0.2)
+    n_specular = int(np.sum(specular_mask))
+    specular_fraction = float(n_specular / (gray.size + 1e-9))
+
+    if n_specular < 50:
+        return {
+            "test": "Specular Reflections",
+            "score": 0.0,
+            "note": "Insufficient specular highlights for analysis",
+            "specular_count": n_specular,
+        }
+
+    # Check if specular highlights are compact (real) vs diffuse (AI)
+    from scipy.ndimage import label
+    labeled, n_features = label(specular_mask)
+    if n_features > 0:
+        sizes = [int(np.sum(labeled == i)) for i in range(1, min(n_features + 1, 100))]
+        avg_size = float(np.mean(sizes))
+        size_std = float(np.std(sizes))
+        size_cv = size_std / (avg_size + 1e-9)  # coefficient of variation
+    else:
+        size_cv = 0.0
+        avg_size = 0.0
+
+    # Real highlights: varied sizes (large CV), AI: uniform sizes
+    if size_cv > 1.0:
+        score = -0.2
+        note = f"Varied specular highlight sizes (CV={size_cv:.2f}, natural)"
+    elif size_cv < 0.3 and n_features > 3:
+        score = 0.3
+        note = f"Suspiciously uniform highlight sizes (CV={size_cv:.2f})"
+    else:
+        score = 0.0
+        note = f"Specular analysis neutral (CV={size_cv:.2f})"
+
+    return {
+        "test": "Specular Reflections",
+        "specular_count": n_specular,
+        "n_highlights": n_features,
+        "size_cv": round(size_cv, 4),
+        "avg_size": round(avg_size, 2),
+        "score": score,
+        "note": note,
+    }
+
+
+# ─── Purple Fringing Detection ──────────────────────────────────────
+def analyze_purple_fringing(img: Image.Image) -> Dict[str, Any]:
+    """
+    Real cameras exhibit purple/magenta fringing at high-contrast edges
+    due to chromatic aberration. AI images rarely reproduce this artifact.
+    """
+    rgb = _to_rgb(img)
+    gray = np.mean(rgb, axis=-1)
+
+    # Find high-contrast edges
+    edge = np.hypot(sobel(gray, axis=0), sobel(gray, axis=1))
+    edge_mask = edge > np.percentile(edge, 95)
+
+    # Check for purple/magenta hue at edges
+    r, g, b = rgb[:, :, 0], rgb[:, :, 1], rgb[:, :, 2]
+
+    # Purple = high R, low G, high B
+    purple_score_map = (r + b - 2 * g) / (r + g + b + 1e-9)
+    edge_purple = purple_score_map[edge_mask]
+
+    if len(edge_purple) < 100:
+        return {
+            "test": "Purple Fringing",
+            "score": 0.0,
+            "note": "Insufficient high-contrast edges for fringing analysis",
+        }
+
+    mean_purple = float(np.mean(edge_purple))
+    purple_fraction = float(np.mean(edge_purple > 0.1))
+
+    if purple_fraction > 0.05:
+        score = -0.3
+        note = f"Purple fringing detected at {purple_fraction:.1%} of edges (real lens artifact)"
+    elif purple_fraction < 0.01 and mean_purple < 0.02:
+        score = 0.2
+        note = "No purple fringing (uncommon in real photography, possible AI)"
+    else:
+        score = 0.0
+        note = f"Minimal fringing (fraction={purple_fraction:.3f})"
+
+    return {
+        "test": "Purple Fringing",
+        "purple_fraction": round(purple_fraction, 4),
+        "mean_purple_score": round(mean_purple, 4),
+        "score": score,
+        "note": note,
+    }
+
+
 # ─── 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})
+    tests = [
+        analyze_chromatic_aberration,
+        analyze_vignetting,
+        analyze_dof_consistency,
+        analyze_bokeh,
+        analyze_lens_distortion,
+        analyze_ca_radial_pattern,
+        analyze_specular_reflections,
+        analyze_purple_fringing,
+    ]
+
+    for fn in tests:
+        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})
 
     if scores:
         avg_score = float(np.mean(scores))
@@ -352,7 +577,7 @@ def run_optical_agent(img: Image.Image) -> 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),
+        failure_prob=max(0.0, 1.0 - len(scores) / len(tests)),
         rationale=rationale,
         sub_findings=findings,
     )