Spaces:
Paused
Paused
| """ | |
| Model Server for Patch-MFFT Tampering Forensics. | |
| Handles inference, aspect-ratio preserving preprocessing, | |
| 3-class prediction, and pixel-level tampering localization masks. | |
| """ | |
| import base64 | |
| import io | |
| import time | |
| from pathlib import Path | |
| from typing import Any, Dict, Optional | |
| import numpy as np | |
| from PIL import Image | |
| import torch | |
| import torch.nn.functional as F | |
| import sys | |
| sys.path.insert(0, str(Path(__file__).parent.parent / "model" / "src")) | |
| try: | |
| from model.src.model import PatchMFFT, build_patch_mfft | |
| except ImportError: | |
| from model import PatchMFFT, build_patch_mfft | |
| TURBO_LUT = np.array([ | |
| [48, 18, 59], [49, 21, 66], [50, 24, 74], [52, 27, 81], [53, 30, 88], [54, 33, 95], [55, 35, 101], [56, 38, 108], | |
| [57, 41, 114], [58, 44, 121], [59, 47, 127], [60, 50, 133], [60, 53, 139], [61, 55, 145], [62, 58, 150], [63, 61, 156], | |
| [64, 64, 161], [64, 67, 166], [65, 69, 171], [65, 72, 176], [66, 75, 181], [67, 78, 186], [67, 80, 190], [67, 83, 194], | |
| [68, 86, 199], [68, 88, 203], [69, 91, 206], [69, 94, 210], [69, 96, 214], [69, 99, 217], [70, 102, 221], [70, 104, 224], | |
| [70, 107, 227], [70, 109, 230], [70, 112, 232], [70, 115, 235], [70, 117, 237], [70, 120, 240], [70, 122, 242], [70, 125, 244], | |
| [70, 127, 246], [70, 130, 248], [69, 132, 249], [69, 135, 251], [69, 137, 252], [68, 140, 253], [67, 142, 253], [66, 145, 254], | |
| [65, 147, 254], [64, 150, 254], [63, 152, 254], [62, 155, 254], [60, 157, 253], [59, 160, 252], [57, 162, 252], [56, 165, 251], | |
| [54, 168, 249], [52, 170, 248], [51, 172, 246], [49, 175, 245], [47, 177, 243], [45, 180, 241], [43, 182, 239], [42, 185, 237], | |
| [40, 187, 235], [38, 189, 233], [37, 192, 230], [35, 194, 228], [33, 196, 225], [32, 198, 223], [30, 201, 220], [29, 203, 218], | |
| [28, 205, 215], [27, 207, 212], [26, 209, 210], [25, 211, 207], [24, 213, 204], [24, 215, 202], [23, 217, 199], [23, 218, 196], | |
| [23, 220, 194], [23, 222, 191], [24, 224, 189], [24, 225, 186], [25, 227, 184], [26, 228, 182], [27, 229, 180], [29, 231, 177], | |
| [30, 232, 175], [32, 233, 172], [34, 235, 169], [36, 236, 166], [39, 237, 163], [41, 238, 160], [44, 239, 157], [47, 240, 154], | |
| [50, 241, 151], [53, 243, 148], [56, 244, 145], [59, 244, 141], [63, 245, 138], [66, 246, 135], [70, 247, 131], [74, 248, 128], | |
| [77, 249, 124], [81, 249, 121], [85, 250, 118], [89, 251, 114], [93, 251, 111], [97, 252, 108], [101, 252, 104], [105, 253, 101], | |
| [109, 253, 98], [113, 253, 95], [116, 254, 92], [120, 254, 89], [124, 254, 86], [128, 254, 83], [132, 254, 80], [135, 254, 77], | |
| [139, 254, 75], [142, 254, 72], [146, 254, 70], [149, 254, 68], [152, 254, 66], [155, 253, 64], [158, 253, 62], [161, 252, 61], | |
| [164, 252, 59], [166, 251, 58], [169, 251, 57], [172, 250, 55], [174, 249, 55], [177, 248, 54], [179, 248, 53], [182, 247, 53], | |
| [185, 245, 52], [187, 244, 52], [190, 243, 52], [192, 242, 51], [195, 241, 51], [197, 239, 51], [200, 238, 51], [202, 237, 51], | |
| [205, 235, 52], [207, 234, 52], [209, 232, 52], [212, 231, 53], [214, 229, 53], [216, 227, 53], [218, 226, 54], [221, 224, 54], | |
| [223, 222, 54], [225, 220, 55], [227, 218, 55], [229, 216, 56], [231, 215, 56], [232, 213, 56], [234, 211, 57], [236, 209, 57], | |
| [237, 207, 57], [239, 205, 57], [240, 203, 58], [242, 200, 58], [243, 198, 58], [244, 196, 58], [246, 194, 58], [247, 192, 57], | |
| [248, 190, 57], [249, 188, 57], [249, 186, 56], [250, 183, 55], [251, 181, 55], [251, 179, 54], [252, 176, 53], [252, 174, 52], | |
| [253, 171, 51], [253, 169, 50], [253, 166, 49], [253, 163, 48], [254, 161, 47], [254, 158, 46], [254, 155, 45], [254, 152, 44], | |
| [253, 149, 43], [253, 146, 41], [253, 143, 40], [253, 140, 39], [252, 137, 38], [252, 134, 36], [251, 131, 35], [251, 128, 34], | |
| [250, 125, 32], [250, 122, 31], [249, 119, 30], [248, 116, 28], [247, 113, 27], [247, 110, 26], [246, 107, 24], [245, 104, 23], | |
| [244, 101, 22], [243, 99, 21], [242, 96, 20], [241, 93, 19], [239, 90, 17], [238, 88, 16], [237, 85, 15], [236, 82, 14], | |
| [234, 80, 13], [233, 77, 13], [232, 75, 12], [230, 73, 11], [229, 70, 10], [227, 68, 10], [226, 66, 9], [224, 64, 8], | |
| [222, 62, 8], [221, 60, 7], [219, 58, 7], [217, 56, 6], [215, 54, 6], [214, 52, 5], [212, 50, 5], [210, 48, 5], | |
| [208, 47, 4], [206, 45, 4], [203, 43, 3], [201, 41, 3], [199, 40, 3], [197, 38, 2], [195, 36, 2], [192, 35, 2], | |
| [190, 33, 2], [187, 31, 1], [185, 30, 1], [182, 28, 1], [180, 27, 1], [177, 25, 1], [174, 24, 1], [172, 22, 1], | |
| [169, 21, 1], [166, 20, 1], [163, 18, 1], [160, 17, 1], [157, 16, 1], [154, 14, 1], [151, 13, 1], [148, 12, 1], | |
| [145, 11, 1], [142, 10, 1], [139, 9, 1], [135, 8, 1], [132, 7, 1], [129, 6, 2], [125, 5, 2], [122, 4, 2] | |
| ], 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, | |
| ) -> Dict[str, Any]: | |
| """ | |
| Inspects whether an input image conforms to optical photographic statistics | |
| or represents Out-of-Distribution (OOD) digital graphics (e.g. draw.io diagrams, | |
| software screenshots, vector line art, pure blank canvas, or stylized anime/art). | |
| """ | |
| img_rgb = image.convert("RGB") | |
| arr = np.array(img_rgb, dtype=np.float32) | |
| h, w, _ = arr.shape | |
| total_pixels = h * w | |
| std_dev = float(np.std(arr)) | |
| # Color quantization | |
| quantized = (arr // 32).astype(np.int32) | |
| hash_vals = quantized[:, :, 0] * 64 + quantized[:, :, 1] * 8 + quantized[:, :, 2] | |
| unique, counts = np.unique(hash_vals, return_counts=True) | |
| top_color_ratio = float(np.max(counts)) / total_pixels | |
| unique_bins = len(unique) | |
| # Exact background flat pixel ratio | |
| flat_pixels_exact = ( | |
| float(np.max(np.unique(arr.reshape(-1, 3), axis=0, return_counts=True)[1])) | |
| / total_pixels | |
| ) | |
| # Pure NumPy Grayscale (Rec. 601 luma) | |
| gray = 0.299 * arr[:, :, 0] + 0.587 * arr[:, :, 1] + 0.114 * arr[:, :, 2] | |
| # Color saturation in pure NumPy | |
| max_c = np.max(arr, axis=2) | |
| min_c = np.min(arr, axis=2) | |
| sat = np.where(max_c > 0, (max_c - min_c) / (max_c + 1e-6) * 255.0, 0.0) | |
| high_sat_ratio = float(np.mean(sat > 100)) | |
| is_ood = False | |
| domain_status = "photographic" | |
| warning = None | |
| # Check 1: Filename metadata indication | |
| fname_lower = (filename or "").lower() | |
| art_keywords = [ | |
| "ghibli", "chatgpt", "dall-e", "dalle", "midjourney", "anime", | |
| "cartoon", "illustration", "sketch", "draw.io", "diagram", | |
| "flowchart", "figma", "screenshot", "clipart", "render" | |
| ] | |
| found_keywords = [k for k in art_keywords if k in fname_lower] | |
| if found_keywords: | |
| is_ood = True | |
| domain_status = "stylized_digital_graphic" | |
| kw_str = ", ".join(found_keywords) | |
| warning = ( | |
| f"Non-photographic / stylized digital artwork detected ({kw_str}). " | |
| f"Forensics models are calibrated on camera optics; synthetic cartoon styling alters natural sensor noise." | |
| ) | |
| elif std_dev < 1.5: | |
| is_ood = True | |
| domain_status = "solid_canvas_ood" | |
| warning = ( | |
| "Input is a solid blank canvas with zero sensor noise variance (not an optical photograph)." | |
| ) | |
| elif flat_pixels_exact > 0.50 and unique_bins < 100: | |
| is_ood = True | |
| domain_status = "non_photographic_graphic" | |
| warning = ( | |
| f"Input appears to be a digital diagram or vector graphic (e.g., draw.io, UI screenshot) " | |
| f"with {flat_pixels_exact * 100:.1f}% flat background. Forensics models require camera optical physics." | |
| ) | |
| elif unique_bins < 25 and top_color_ratio > 0.40: | |
| is_ood = True | |
| domain_status = "non_photographic_graphic" | |
| warning = ( | |
| f"Input appears to be a digital sketch or illustration with only {unique_bins} discrete color bins. " | |
| f"Forensics models require camera optical physics." | |
| ) | |
| elif high_sat_ratio > 0.85: | |
| is_ood = True | |
| domain_status = "stylized_digital_graphic" | |
| warning = ( | |
| f"Hyper-saturated digital illustration profile detected ({high_sat_ratio * 100:.1f}% saturated pixels). " | |
| f"Non-photographic artwork." | |
| ) | |
| # 8x8 DCT JPEG Blockiness calculation (Pure NumPy, vectorized across horizontal and vertical axes) | |
| h_blocks = h // 8 | |
| w_blocks = w // 8 | |
| blockiness = 1.0 | |
| if h_blocks > 2 and w_blocks > 2: | |
| b_rows = np.arange(1, h_blocks) * 8 | |
| boundary_diff_h = np.mean(np.abs(gray[b_rows, :] - gray[b_rows - 1, :])) | |
| interior_diff_h = np.mean(np.abs(gray[b_rows - 1, :] - gray[b_rows - 2, :])) | |
| blk_h = boundary_diff_h / (interior_diff_h + 1e-6) | |
| b_cols = np.arange(1, w_blocks) * 8 | |
| boundary_diff_v = np.mean(np.abs(gray[:, b_cols] - gray[:, b_cols - 1])) | |
| interior_diff_v = np.mean(np.abs(gray[:, b_cols - 1] - gray[:, b_cols - 2])) | |
| blk_v = boundary_diff_v / (interior_diff_v + 1e-6) | |
| blockiness = float((blk_h + blk_v) / 2.0) | |
| # Social media compression indicators | |
| social_media_keywords = [ | |
| "whatsapp", "telegram", "messenger", "instagram", "discord", | |
| "facebook", "fb_", "signal", "wechat", "viber" | |
| ] | |
| has_sm_keyword = any(k in fname_lower for k in social_media_keywords) | |
| is_social_media_compressed = has_sm_keyword or (blockiness > 1.25) | |
| compression_warning = None | |
| if is_social_media_compressed: | |
| compression_warning = ( | |
| f"Social media lossy compression detected (8x8 DCT blockiness {blockiness:.2f}). " | |
| f"Messaging platforms (WhatsApp/Telegram) discard high-frequency Fourier harmonics (see Manuscript Table 7). " | |
| f"Treat 'Authentic' classifications with caution." | |
| ) | |
| full_warning = warning | |
| if compression_warning: | |
| full_warning = f"{full_warning} | {compression_warning}" if full_warning else compression_warning | |
| return { | |
| "is_non_photographic_graphic": is_ood, | |
| "domain_status": domain_status, | |
| "is_social_media_compressed": is_social_media_compressed, | |
| "compression_blockiness": round(blockiness, 3), | |
| "warning": full_warning, | |
| } | |
| class TamperingModelServer: | |
| def __init__( | |
| self, | |
| checkpoint_path: Optional[str] = None, | |
| variant: str = "large", | |
| img_size: int = 384, | |
| ): | |
| self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") | |
| self.variant = variant | |
| self.img_size = img_size | |
| print(f"[TamperingModelServer] Initializing Patch-MFFT ({variant}) on {self.device}") | |
| self.model = build_patch_mfft(variant, img_size=img_size, num_classes=3) | |
| self.model = self.model.to(self.device) | |
| self.is_loaded = False | |
| if checkpoint_path and Path(checkpoint_path).exists(): | |
| self.load_checkpoint(checkpoint_path) | |
| else: | |
| print(f"[TamperingModelServer] Running with initialized weights (no checkpoint at {checkpoint_path})") | |
| self.model.eval() | |
| self.warmup() | |
| def load_checkpoint(self, path: str): | |
| try: | |
| state = torch.load(path, map_location=self.device, weights_only=True) | |
| except Exception: | |
| state = torch.load(path, map_location=self.device, weights_only=False) | |
| raw_state = state["model_state_dict"] if isinstance(state, dict) and "model_state_dict" in state else state | |
| self.model.load_state_dict(raw_state, strict=True) | |
| self.is_loaded = True | |
| print(f"[TamperingModelServer] Successfully loaded weights with strict=True from {path}") | |
| def warmup(self, iters: int = 2): | |
| dummy = torch.randn(1, 3, self.img_size, self.img_size).to(self.device) | |
| with torch.no_grad(): | |
| for _ in range(iters): | |
| _ = self.model(dummy, return_mask=True) | |
| print(f"[TamperingModelServer] Warmup completed ({iters} iterations).") | |
| def _preprocess(self, image: Image.Image) -> torch.Tensor: | |
| # Standardize directly to model input resolution via bilinear resampling, | |
| # perfectly matching the training dataset pipeline (Cell 14 of kaggle_patch_mfft_pipeline) | |
| img_res = image.convert("RGB").resize((self.img_size, self.img_size), Image.Resampling.BILINEAR) | |
| img_array = np.array(img_res, dtype=np.float32) / 255.0 | |
| mean = np.array([0.485, 0.456, 0.406], dtype=np.float32) | |
| std = np.array([0.229, 0.224, 0.225], dtype=np.float32) | |
| img_array = ((img_array - mean) / std).astype(np.float32) | |
| tensor = torch.from_numpy(img_array).permute(2, 0, 1).unsqueeze(0) | |
| return tensor.to(self.device) | |
| def predict( | |
| self, | |
| image: Image.Image, | |
| filename: Optional[str] = None, | |
| sha256_hash: Optional[str] = None, | |
| ) -> Dict[str, Any]: | |
| start = time.time() | |
| tensor = self._preprocess(image) | |
| logits, mask = self.model(tensor, return_mask=True) | |
| probs = F.softmax(logits, dim=-1)[0] | |
| pred_idx = torch.argmax(probs).item() | |
| classes = ["authentic", "full_ai", "spliced_tampered"] | |
| prediction_label = classes[pred_idx] | |
| class_probs = { | |
| "authentic": round(probs[0].item(), 4), | |
| "full_ai": round(probs[1].item(), 4), | |
| "spliced_tampered": round(probs[2].item(), 4), | |
| } | |
| # Raw evaluated pixel-level tampering probability from dense convolutional decoder (loc): | |
| # Directly evaluated from multi-frequency feature map, without artificial color or class overrides. | |
| mask_np = torch.sigmoid(mask)[0, 0].cpu().numpy() | |
| peak_score = round(float(mask_np.max()), 4) | |
| coverage_ratio = round(float(np.mean(mask_np > 0.50)), 4) | |
| # 1. Standalone publication-style Turbo false-color RGB heatmap (Matches Figure 2 Column d in paper) | |
| # 0.0 (Deep Navy/Indigo): Pristine Authentic | 0.50 (Lime Green): Transition | 1.0 (Crimson Red): High Tampering | |
| lut_idx = np.clip(mask_np * 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") | |
| # 2. Seamless RGBA Overlay: Authentic background is 100% transparent (alpha=0), | |
| # only tampering hotspots glow in vivid Turbo colors without discoloring the original photograph | |
| overlay_rgba = np.zeros((self.img_size, self.img_size, 4), dtype=np.uint8) | |
| overlay_rgba[:, :, :3] = heatmap_rgb | |
| # Threshold: authentic pixels (p < 0.12) are completely transparent; ramps smoothly up to 210 alpha for hotspots | |
| alpha_map = np.clip((mask_np - 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_overlay = io.BytesIO() | |
| overlay_pil.save(buf_overlay, format="PNG") | |
| overlay_b64 = base64.b64encode(buf_overlay.getvalue()).decode("utf-8") | |
| # Frequency bands | |
| bands = self.model.decomposer(tensor) | |
| band_mags = [float(b.abs().mean().item()) for b in bands] | |
| band_dict = { | |
| f"band_{i+1}": round(band_mags[i], 4) for i in range(len(band_mags)) | |
| } | |
| elapsed_ms = (time.time() - start) * 1000.0 | |
| # Pre-flight domain inspection (draw.io, screenshots, blank canvas, anime/art, social media compression) | |
| ood_info = check_photographic_domain(image, filename=filename) | |
| domain_status = ood_info["domain_status"] | |
| is_non_photographic = ood_info["is_non_photographic_graphic"] | |
| is_sm_compressed = ood_info.get("is_social_media_compressed", False) | |
| 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." | |
| warning = f"{warning} | {entropy_msg}" if warning else entropy_msg | |
| return { | |
| "prediction": prediction_label, | |
| "confidence": conf, | |
| "tampering_detected": pred_idx in [1, 2], | |
| "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": band_dict, | |
| "processing_time_ms": round(elapsed_ms, 2), | |
| "model_variant": self.variant, | |
| "is_non_photographic_graphic": is_non_photographic, | |
| "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, | |
| } | |