"""A-EYE backend — HYBRID entrypoint (real-use tuned).
Real-world feedback: model 49 discriminates better than the PatchGuard image
verdict, but its heatmap (old model 24) was weak. So:
- VERDICT = model 49 (the proven champion discriminator)
- + composite rescue: if 49 says REAL but PatchGuard finds a CONCENTRATED
high-confidence AI region (real photo with an inserted/AI object), escalate
to AI — the case 49 structurally misses.
- HEATMAP = dense PatchGuard (model 63) map, rendered vivid, only when AI.
Same /api/analyze contract as the existing app (schemas.AnalyzeResponse), so the
Expo app needs no changes. New file; existing backend files are untouched.
Run (from a-eye/backend):
venv\\Scripts\\python -m uvicorn main_hybrid:app --host 0.0.0.0 --port 8000
"""
from __future__ import annotations
import base64
import io
import os
import sys
import time
from contextlib import asynccontextmanager
from typing import Any, AsyncIterator
import numpy as np
import torch
import torchvision.transforms as T
from PIL import Image, ImageOps
from scipy.ndimage import gaussian_filter
from fastapi import FastAPI, File, HTTPException, UploadFile
from fastapi.concurrency import run_in_threadpool
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
import provenance_gate
import synthid_gate
from heatmap_nextgen import overlay as render_overlay, pure_heatmap, residual_var
from patchguard import PATCHGUARD_ARCH, PatchGuardDetector
from schemas import AnalyzeResponse, HealthResponse, VersionResponse
from zero_shot_v4 import CLIP_MEAN, CLIP_STD, ZeroShotV4Detector
# Model 63 = dense-decoder architecture from the detector repo. Reuse that exact
# (trained) code so loading can never drift from how the checkpoint was produced.
_DETECTOR_ROOT = os.getenv("AEYE_DETECTOR_ROOT", r"C:\Users\user\Desktop\a-eye(vit_clip_LoRA)")
if os.path.isdir(_DETECTOR_ROOT) and _DETECTOR_ROOT not in sys.path:
sys.path.insert(0, _DETECTOR_ROOT)
from aeye_next.models.dense_patchguard import DensePatchGuardDetector
MAX_IMAGE_BYTES = 28 * 1024 * 1024
DEVICE = os.environ.get("AEYE_DEVICE", "cpu")
INPUT_SIZE = 224
VERDICT_PATH = os.getenv("VERDICT_MODEL_PATH", "./models/best49.pt")
HEATMAP_PATH = os.getenv("HEATMAP_MODEL_PATH", "./models/best63.pt")
HEATMAP_KIND = os.getenv("HEATMAP_MODEL_KIND", "dense").lower()
# model 49 architecture (matches the app's model_loader.MODEL_49)
MODEL_49_ARCH = dict(
clip_backbone="clip-vit-l-14", clip_layer=13, semantic_dim=512, forensic_dim=256,
frequency_dim=192, fft_bins=48, image_size=224, num_classes=2, num_sources=20,
dropout=0.26, source_grl_lambda=0.0, freeze_clip=True,
)
# model 63 = dense localization decoder (replaces 61 as heatmap + rescue). Native
# 32x32 dense patch map -> sharper localization; image head powers the rescue gate.
DENSE_ARCH = dict(
clip_backbone="clip-vit-l-14", clip_layer=13, semantic_dim=512, forensic_dim=256,
frequency_dim=192, fft_bins=48, num_classes=2, num_sources=2,
dropout=0.26, source_grl_lambda=0.0, freeze_clip=True, patch_hidden=256,
patch_topk_frac=0.06, decoder_grid=32, decoder_channels=192, decoder_blocks=3,
decoder_dropout=0.04,
)
# model 53 = SigLIP-large/256 verdict head. DIFFERENT backbone than 49/63 (which
# share CLIP-L), so it makes uncorrelated errors -> used as a 2nd rescue. Measured:
# adds ~+9pt AI recall over 49 alone with ~0 extra real false positives. (Does NOT
# fix GPT — it shares the SD/FLUX-only training gap.) Loaded via the detector repo.
# OFF by default: the 53 ensemble was validated and does NOT help under the
# real-precision priority — at an FPR-safe gate it adds ~0 recall, and a low gate
# triples real false positives. Kept for re-use after a GPT-data retrain fixes 53.
ENABLE_SIGLIP = os.getenv("ENABLE_SIGLIP", "0") == "1"
SIGLIP_PATH = os.getenv("SIGLIP_MODEL_PATH", "./models/best53.pt")
SIGLIP_SIZE = 256
SIGLIP_THR = float(os.getenv("SIGLIP_THRESHOLD", "0.90"))
SIGLIP_53_ARCH = dict(
clip_backbone="siglip-large-256", clip_layer=13, semantic_dim=512, forensic_dim=256,
frequency_dim=192, fft_bins=48, num_classes=2, num_sources=2, freeze_clip=True,
source_grl_lambda=0.0, dropout=0.26, image_size=SIGLIP_SIZE,
)
# composite rescue: only when model 49 says REAL but PatchGuard's TRAINED image
# verdict (p_ai) is confidently AI. Measured on model 63: real-paste/control p_ai
# clears 0.72 only ~3% of the time while AI inserts clear it ~36% — model 49 alone
# catches just ~9% of inserts, so this rescue is the main insert detector.
RESCUE_P59 = float(os.getenv("PATCHGUARD_RESCUE_THRESHOLD", "0.72"))
# real-guard: model 49 sometimes false-positives on real CAMERA photos (out of its
# training distribution). When 49 only *weakly* says AI, PatchGuard (model 61, which
# has stronger real precision) confidently says REAL, there is no provenance and no
# localized insert, we trust REAL — a camera photo must stay real.
REAL_GUARD_MARGIN = 0.12 # only override when score49 is within this of the threshold
REAL_GUARD_P59 = 0.20 # model 61 must be confidently real
REAL_GUARD_PEAK = 0.45 # and find no localized insert
def _read_threshold(path: str, default: float) -> float:
try:
ck = torch.load(path, map_location="cpu", weights_only=False)
metrics = ck.get("metrics", {}) if isinstance(ck, dict) else {}
for key in ("calibrated_threshold", "threshold"):
v = metrics.get(key)
if isinstance(v, (int, float)) and 0.0 < float(v) < 1.0:
return float(v)
except Exception:
pass
return default
class HybridDetector:
def __init__(self) -> None:
self.device = torch.device(DEVICE)
# VERDICT preprocessing must match how the models were trained/evaluated:
# resize the shorter side to 256, then center-crop 224 (aspect preserved).
# The old square Resize((224,224)) DISTORTED non-square images (screenshots
# especially), which measurably hurt detection — e.g. a screenshotted GPT
# image scored 0.60 squished vs 0.77 center-cropped.
self.transform = T.Compose([
T.Resize(256, interpolation=T.InterpolationMode.BICUBIC),
T.CenterCrop(INPUT_SIZE),
T.ToTensor(),
T.Normalize(CLIP_MEAN, CLIP_STD),
])
# HEATMAP preprocessing keeps the WHOLE image (square resize) so the overlay
# lines up with the full displayed photo instead of only the center crop.
self.transform_full = T.Compose([
T.Resize((INPUT_SIZE, INPUT_SIZE), interpolation=T.InterpolationMode.BICUBIC),
T.ToTensor(),
T.Normalize(CLIP_MEAN, CLIP_STD),
])
# SigLIP (model 53) takes 256px input (its own backbone resolution).
self.transform_53 = T.Compose([
T.Resize(round(SIGLIP_SIZE / 0.875), interpolation=T.InterpolationMode.BICUBIC),
T.CenterCrop(SIGLIP_SIZE),
T.ToTensor(),
T.Normalize(CLIP_MEAN, CLIP_STD),
])
self.verdict = ZeroShotV4Detector(**MODEL_49_ARCH)
self._load(self.verdict, VERDICT_PATH)
self.threshold = _read_threshold(VERDICT_PATH, 0.4625)
if HEATMAP_KIND == "dense" or os.path.basename(HEATMAP_PATH).lower().startswith("best63"):
self.patch = DensePatchGuardDetector(**DENSE_ARCH)
else:
self.patch = PatchGuardDetector(**PATCHGUARD_ARCH)
self._load(self.patch, HEATMAP_PATH)
# 2nd verdict head — SigLIP (model 53). OFF by default (see ENABLE_SIGLIP).
self.siglip = None
if ENABLE_SIGLIP:
self.siglip = ZeroShotV4Detector(**SIGLIP_53_ARCH)
self._load(self.siglip, SIGLIP_PATH)
def _load(self, model: torch.nn.Module, path: str) -> None:
ck = torch.load(path, map_location="cpu", weights_only=False)
state = ck.get("model_trainable_state_dict", ck.get("model_state_dict", ck))
model.load_trainable_state_dict(state)
model.to(self.device).eval()
@torch.no_grad()
def _dense_patch(self, pil: Image.Image) -> np.ndarray:
"""Model 63's native dense localization map (32x32) over the WHOLE image
(full-frame resize) so the heatmap aligns with the displayed photo."""
x = self.transform_full(pil).unsqueeze(0).to(self.device)
out = self.patch(x)
return torch.sigmoid(out["patch_logits"].float())[0].cpu().numpy()
@torch.no_grad()
def _hires_loc(self, original: Image.Image, base: np.ndarray | None = None) -> np.ndarray:
"""Dense 32x32 localization with horizontal-flip TTA (averaging the map over the
image and its mirror steadies it), fused with the noise-residual map and
smoothed. `base` reuses the patch map already computed in analyze() so only the
mirror pass is extra."""
ph = self._dense_patch(original) if base is None else base
ph_flip = self._dense_patch(original.transpose(Image.FLIP_LEFT_RIGHT))[:, ::-1]
ph = (ph + ph_flip) / 2.0
gh = ph.shape[0]
rv = residual_var(original, gh)
rng = float(np.ptp(rv))
rv = (rv - rv.min()) / (rng + 1e-6) if rng > 1e-6 else np.zeros_like(rv)
return gaussian_filter(ph * (0.5 + 0.5 * rv), 0.7)
@torch.no_grad()
def analyze(self, image_bytes: bytes, filename: str | None = None) -> dict[str, Any]:
started = time.perf_counter()
# Phone photos often carry EXIF orientation instead of rotated pixels.
# Normalize it before model inference and overlay rendering so the
# original/heatmap tabs do not appear rotated relative to each other.
original = ImageOps.exif_transpose(Image.open(io.BytesIO(image_bytes))).convert("RGB")
x = self.transform(original).unsqueeze(0).to(self.device)
provenance = provenance_gate.check_bytes(image_bytes)
filename_signals = provenance_gate.check_filename(filename)
signals = sorted(set((provenance.get("signals") or []) + filename_signals))
# Optional SynthID pixel-watermark check (no-op unless SYNTHID_API_* env set).
# Unlike byte/metadata provenance, this survives screenshots / re-encoding.
synthid = synthid_gate.check_image(image_bytes)
if synthid.get("signals"):
signals = sorted(set(signals + synthid["signals"]))
provenance.setdefault("details", {})["synthid"] = synthid.get("details")
provenance["signals"] = signals
if filename_signals:
provenance.setdefault("details", {})["filename_hint"] = filename
out49 = self.verdict(x)
logits = out49["logits"] if isinstance(out49, dict) else out49
score49 = float(torch.softmax(logits.float().reshape(-1)[:2], dim=0)[1])
score = score49
pg = self.patch(x)
patch = torch.sigmoid(pg["patch_logits"].float())[0].cpu().numpy() # 16x16
peak = float(patch.max())
p_ai_59 = float(torch.sigmoid(pg["z_img"].float())[0])
# 2nd verdict head (SigLIP/53) — only when enabled (off by default).
s53, ai53 = 0.0, False
if self.siglip is not None:
out53 = self.siglip(self.transform_53(original).unsqueeze(0).to(self.device))
logits53 = out53["logits"] if isinstance(out53, dict) else out53
s53 = float(torch.softmax(logits53.float().reshape(-1)[:2], dim=0)[1])
ai53 = s53 >= SIGLIP_THR
source = "model"
verdict = "AI" if score >= self.threshold else "REAL"
model_verdict = verdict # the pixel model's OWN call, before provenance/rescue
reason = "model 49 verdict"
patchguard_would_rescue = model_verdict == "REAL" and p_ai_59 >= RESCUE_P59
rescued = False
if signals:
verdict = "AI"
score = max(score, 0.99)
source = "provenance"
reason = "AI provenance found (" + ", ".join(str(s) for s in signals) + ")"
rescued = patchguard_would_rescue
if model_verdict == "AI" or patchguard_would_rescue or ai53:
reason += "; pixel model also positive"
elif patchguard_would_rescue:
verdict = "AI"
score = max(score, p_ai_59)
source = "patchguard_rescue"
rescued = True
reason = "model 49 + AI-insert rescue (PatchGuard)"
elif model_verdict == "REAL" and ai53:
# 49 said REAL, but the SigLIP head (different backbone) flags AI.
verdict = "AI"
score = max(score, s53)
source = "siglip_rescue"
rescued = True
reason = "model 49 REAL but SigLIP(53) ensemble flags AI"
elif (model_verdict == "AI" and not ai53 and score49 < self.threshold + REAL_GUARD_MARGIN
and p_ai_59 < REAL_GUARD_P59 and peak < REAL_GUARD_PEAK):
# camera/real photo that model 49 only weakly flags as AI, with no
# provenance and no localized insert, and PatchGuard says clearly REAL.
verdict = "REAL"
source = "real_guard"
score = min(score, self.threshold - 0.01)
reason = "model 49 weak-AI overridden by PatchGuard real-guard (camera/real)"
# Downscale for display so the base64 payload stays small. A full-res
# PNG of a phone photo is 10-25MB and crashes the app on decode. 1024px
# is plenty for the heatmap overlay; the real-verdict thumbnail is smaller.
display = _downscale(original, 768)
if verdict == "AI":
# Always show the REAL model heatmap — even when provenance is what flagged
# the image. (Previously the provenance-only case painted a flat red wash;
# the user prefers a real localized heatmap. For a fully-AI image the model
# has no single region to point at, so this highlights its most-suspect
# area — approximate, but a real heatmap rather than a flat blanket.)
# Dense localization over the WHOLE image (residual-fused + flip-TTA), so
# the overlay lines up with the full displayed photo.
loc = self._hires_loc(original)
# Always render VIVID — the hot region is normalized to full color, no
# confidence dimming (product preference: heatmap should never look faint).
conf = 1.0
heatmap_b64 = _png_b64(pure_heatmap(loc, display.size, conf, blanket=False))
overlay_b64 = _png_b64(render_overlay(display, loc, conf, blanket=False))
flat = np.sort(patch.ravel())[::-1]
k = max(1, int(round(flat.size * 0.08)))
structure_score = float(flat[:k].mean())
detail_score = peak
else:
# proper PNG (the frontend labels data-uris image/png; a JPEG payload
# there can crash the native image decoder on some phones)
clean = _png_b64(_downscale(original, 600))
heatmap_b64 = clean
overlay_b64 = clean
structure_score = detail_score = score
reason = "model 49 verdict - real, heatmap suppressed"
return {
"score": score,
"verdict": verdict,
"heatmap_b64": heatmap_b64,
"overlay_b64": overlay_b64,
"model_version": ("49+53" if ENABLE_SIGLIP else "49") + "_verdict+63_heatmap" + ("+provenance" if source == "provenance" else "") + ("+rescue" if rescued else ""),
"elapsed_ms": int((time.perf_counter() - started) * 1000),
"structure_score": structure_score,
"context_score": score,
"detail_score": detail_score,
"threshold": self.threshold,
"reason": reason,
"source": source,
"provenance": provenance,
"ensemble": {"model49_score": score49, "siglip53_score": s53, "patchguard_score": p_ai_59, "patch_peak": peak, "rescue_threshold": RESCUE_P59},
}
def _png_b64(image: Image.Image) -> str:
buf = io.BytesIO()
image.save(buf, format="PNG")
return base64.b64encode(buf.getvalue()).decode("ascii")
def _downscale(image: Image.Image, max_side: int) -> Image.Image:
w, h = image.size
if max(w, h) <= max_side:
return image
s = max_side / float(max(w, h))
return image.resize((max(1, int(w * s)), max(1, int(h * s))), Image.Resampling.BILINEAR)
_detector: HybridDetector | None = None
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
global _detector
print(f"[hybrid] loading 49 (verdict) + 63 (dense heatmap/rescue) on {DEVICE} ...", flush=True)
_detector = HybridDetector()
print(f"[hybrid] ready: verdict=model49 thr={_detector.threshold} heatmap=model63 dense", flush=True)
yield
app = FastAPI(title="A-EYE Inference API (hybrid)", version="3.0.0", lifespan=lifespan)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True,
allow_methods=["*"], allow_headers=["*"])
_WEBAPP_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "webapp")
_TEST_PAGE = """
A-EYE hybrid