feat: add predict_ensemble, conservative forensic fusion, and exif harvester

api/model_server.py

@@ -59,6 +59,63 @@ TURBO_LUT = np.array([
 ], dtype=np.uint8)
 
 
+def extract_exif_metadata(image: Image.Image) -> Dict[str, Any]:
+    """
+    Extracts camera hardware metadata and detects AI generation software tags
+    embedded within EXIF, XMP, or PNG text segments.
+    """
+    exif_summary = {}
+    generative_tags = []
+    has_exif = False
+
+    gen_keywords = [
+        "firefly", "stable diffusion", "midjourney", "dall-e", "dalle",
+        "comfyui", "automatic1111", "novelai", "invokeai", "photoshop",
+        "generative", "canva", "bing creator"
+    ]
+
+    try:
+        from PIL import ExifTags
+        raw_exif = image.getexif()
+        if raw_exif:
+            for tag_id, val in raw_exif.items():
+                tag_name = ExifTags.TAGS.get(tag_id, str(tag_id))
+                val_str = str(val).strip()
+                if val_str and len(val_str) < 120:
+                    exif_summary[tag_name] = val_str
+                    has_exif = True
+                    for kw in gen_keywords:
+                        if kw in val_str.lower() and kw not in generative_tags:
+                            generative_tags.append(kw)
+    except Exception:
+        pass
+
+    for k, v in image.info.items():
+        if isinstance(v, (str, bytes)):
+            v_str = str(v).strip()
+            if len(v_str) < 150:
+                exif_summary[f"info_{k}"] = v_str
+            for kw in gen_keywords:
+                if kw in v_str.lower() and kw not in generative_tags:
+                    generative_tags.append(kw)
+
+    # Focus on standard photographic capture audit parameters
+    key_tags = [
+        "Make", "Model", "Software", "DateTimeOriginal", "LensModel",
+        "ExposureTime", "FNumber", "ISOSpeedRatings", "FocalLength"
+    ]
+    filtered_exif = {k: exif_summary[k] for k in key_tags if k in exif_summary}
+    if not filtered_exif and exif_summary:
+        filtered_exif = {k: v for i, (k, v) in enumerate(exif_summary.items()) if i < 8}
+
+    return {
+        "has_exif": has_exif,
+        "exif_summary": filtered_exif if filtered_exif else None,
+        "generative_software_tag": ", ".join(generative_tags) if generative_tags else None,
+        "is_generative_exif": len(generative_tags) > 0,
+    }
+
+
 def check_photographic_domain(
     image: Image.Image,
     filename: Optional[str] = None,
@@ -252,6 +309,7 @@ class TamperingModelServer:
         self,
         image: Image.Image,
         filename: Optional[str] = None,
+        sha256_hash: Optional[str] = None,
     ) -> Dict[str, Any]:
         start = time.time()
         tensor = self._preprocess(image)
@@ -314,6 +372,12 @@ class TamperingModelServer:
         blockiness = ood_info.get("compression_blockiness", 1.0)
         warning = ood_info["warning"]
 
+        # EXIF Hardware & Software metadata audit
+        exif_res = extract_exif_metadata(image)
+        if exif_res["is_generative_exif"]:
+            gen_tag_msg = f"AI Generation Metadata Detected in EXIF ({exif_res['generative_software_tag']})"
+            warning = f"{warning} | {gen_tag_msg}" if warning else gen_tag_msg
+
         conf = round(probs[pred_idx].item(), 4)
         if conf < 0.60:
             entropy_msg = f"Low triage certainty ({conf * 100:.1f}%). Model exhibits high entropy across forensic classes."
@@ -335,5 +399,128 @@ class TamperingModelServer:
             "domain_status": domain_status,
             "is_social_media_compressed": is_sm_compressed,
             "compression_blockiness": blockiness,
+            "consensus_status": "single_model",
+            "consensus_message": None,
+            "exif_summary": exif_res["exif_summary"],
+            "generative_software_tag": exif_res["generative_software_tag"],
+            "sha256_hash": sha256_hash,
+            "cached": False,
             "warning": warning,
         }
+
+
+def predict_ensemble(
+    server_base: TamperingModelServer,
+    server_large: TamperingModelServer,
+    image: Image.Image,
+    filename: Optional[str] = None,
+    sha256_hash: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    SOTA Conservative Forensic Fusion Ensemble combining Patch-MFFT Base (3.28M)
+    and Patch-MFFT Large (13.94M). Applies tampering priority gating and element-wise
+    maximum likelihood localization mask fusion.
+    """
+    start = time.time()
+
+    res_b = server_base.predict(image, filename=filename, sha256_hash=sha256_hash)
+    res_l = server_large.predict(image, filename=filename, sha256_hash=sha256_hash)
+
+    pb = res_b["class_probabilities"]
+    pl = res_l["class_probabilities"]
+
+    # Conservative Forensic Fusion (Tampering Priority Gating)
+    # If Base detected localized inpainting (p_spliced > 0.45), preserve localized tampering
+    if pb["spliced_tampered"] > 0.45:
+        prediction_label = "spliced_tampered"
+        conf = pb["spliced_tampered"]
+    # If Large or Base detected Full AI with high confidence, preserve generative evidence
+    elif pl["full_ai"] > 0.70 or pb["full_ai"] > 0.70:
+        prediction_label = "full_ai"
+        conf = max(pl["full_ai"], pb["full_ai"])
+    else:
+        ens_auth = 0.40 * pb["authentic"] + 0.60 * pl["authentic"]
+        ens_ai = 0.40 * pb["full_ai"] + 0.60 * pl["full_ai"]
+        ens_splice = 0.40 * pb["spliced_tampered"] + 0.60 * pl["spliced_tampered"]
+        candidates = [("authentic", ens_auth), ("full_ai", ens_ai), ("spliced_tampered", ens_splice)]
+        candidates.sort(key=lambda x: x[1], reverse=True)
+        prediction_label = candidates[0][0]
+        conf = candidates[0][1]
+
+    # Consensus Status
+    if res_b["prediction"] == res_l["prediction"]:
+        consensus_status = "verified_consensus"
+        consensus_message = f"Multi-Model Consensus: Both Base ({res_b['confidence']*100:.1f}%) and Large ({res_l['confidence']*100:.1f}%) agree unanimously on '{prediction_label}'."
+    else:
+        consensus_status = "disputed"
+        consensus_message = (
+            f"Disputed Forensic Case: Base ({res_b['prediction']}, {res_b['confidence']*100:.1f}%) "
+            f"vs Large ({res_l['prediction']}, {res_l['confidence']*100:.1f}%). "
+            f"Conservative forensic fusion prioritized the higher tampering indicator."
+        )
+
+    # Element-Wise Maximum Likelihood Heatmap Fusion
+    t_b = server_base._preprocess(image)
+    t_l = server_large._preprocess(image)
+    with torch.no_grad():
+        _, mask_b = server_base.model(t_b, return_mask=True)
+        _, mask_l = server_large.model(t_l, return_mask=True)
+        m_b = torch.sigmoid(mask_b)[0, 0].cpu().numpy()
+        m_l = torch.sigmoid(mask_l)[0, 0].cpu().numpy()
+        fused_mask = np.maximum(m_b, m_l)
+
+    peak_score = round(float(fused_mask.max()), 4)
+    coverage_ratio = round(float(np.mean(fused_mask > 0.50)), 4)
+
+    lut_idx = np.clip(fused_mask * 255.0, 0, 255).astype(np.int32)
+    heatmap_rgb = TURBO_LUT[lut_idx]
+    mask_pil = Image.fromarray(heatmap_rgb, mode="RGB")
+    buf = io.BytesIO()
+    mask_pil.save(buf, format="PNG")
+    mask_b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
+
+    overlay_rgba = np.zeros((server_base.img_size, server_base.img_size, 4), dtype=np.uint8)
+    overlay_rgba[:, :, :3] = heatmap_rgb
+    alpha_map = np.clip((fused_mask - 0.12) / 0.25 * 210.0, 0, 210).astype(np.uint8)
+    overlay_rgba[:, :, 3] = alpha_map
+    overlay_pil = Image.fromarray(overlay_rgba, mode="RGBA")
+    buf_o = io.BytesIO()
+    overlay_pil.save(buf_o, format="PNG")
+    overlay_b64 = base64.b64encode(buf_o.getvalue()).decode("utf-8")
+
+    class_probs = {
+        "authentic": round(0.40 * pb["authentic"] + 0.60 * pl["authentic"], 4),
+        "full_ai": round(0.40 * pb["full_ai"] + 0.60 * pl["full_ai"], 4),
+        "spliced_tampered": round(0.40 * pb["spliced_tampered"] + 0.60 * pl["spliced_tampered"], 4),
+    }
+
+    warning = res_b.get("warning") or res_l.get("warning")
+    if consensus_status == "disputed":
+        warning = f"{warning} | {consensus_message}" if warning else consensus_message
+
+    elapsed_ms = (time.time() - start) * 1000.0
+
+    return {
+        "prediction": prediction_label,
+        "confidence": round(float(conf), 4),
+        "tampering_detected": prediction_label in ["full_ai", "spliced_tampered"],
+        "class_probabilities": class_probs,
+        "tampering_heatmap_b64": mask_b64,
+        "tampering_overlay_b64": overlay_b64,
+        "tampering_peak_score": peak_score,
+        "tampering_coverage_ratio": coverage_ratio,
+        "frequency_band_contributions": res_l["frequency_band_contributions"],
+        "processing_time_ms": round(elapsed_ms, 2),
+        "model_variant": "ensemble",
+        "is_non_photographic_graphic": res_b["is_non_photographic_graphic"],
+        "domain_status": res_b["domain_status"],
+        "is_social_media_compressed": res_b["is_social_media_compressed"],
+        "compression_blockiness": res_b["compression_blockiness"],
+        "consensus_status": consensus_status,
+        "consensus_message": consensus_message,
+        "exif_summary": res_b.get("exif_summary"),
+        "generative_software_tag": res_b.get("generative_software_tag"),
+        "sha256_hash": sha256_hash,
+        "cached": False,
+        "warning": warning,
+    }