Switch Space backend to extension-compatible deepfake API

README.md

@@ -1,6 +1,6 @@
 ---
-title: DuneNet Model API
-emoji: 🚀
+title: Deepfake Detection API
+emoji: 🧠
 colorFrom: yellow
 colorTo: red
 sdk: docker
@@ -8,18 +8,28 @@ app_port: 7860
 pinned: false
 ---
 
-# DuneNet Model API
+# Deepfake Detection API (Hugging Face Space)
 
-FastAPI backend for DuneNet - Autonomous UGV Perception Platform.
-
-Runs a fine-tuned Segformer (nvidia/mit-b4) model for semantic segmentation of desert terrain, providing:
-- Semantic segmentation masks
-- Traversability maps for autonomous navigation
-- Live simulation inference with costmap grids
+This Space serves a Deepfake detection backend that is compatible with your browser extension.
 
 ## Endpoints
 
-- `GET /` — Health check
-- `POST /predict` — Full segmentation prediction
-- `POST /predict/sim` — Simulation-optimized prediction with traversability grid
-- `GET /model/info` — Model metadata
+- `GET /` - Health check
+- `GET /health` - Extension health endpoint
+- `POST /analyze` - Analyze one frame (`multipart/form-data`, field: `frame`)
+- `POST /reset` - Reset temporal tracker
+
+## Model Files
+
+Put your deepfake checkpoint in:
+
+- `models/best_model.pth` (preferred)
+
+If no compatible checkpoint is found, the API runs in forensic-only mode.
+
+## Extension Backend URL
+
+After deployment, set the extension backend URL to your Space URL:
+
+- `https://<your-username>-<your-space-name>.hf.space`
+

main.py

@@ -1,496 +1,392 @@
-from fastapi import FastAPI, UploadFile, File, HTTPException
-from fastapi.middleware.cors import CORSMiddleware
-from pydantic import BaseModel
-import torch
-import torch.nn.functional as F
-from PIL import Image
 import io
-import base64
+import os
+import time
+
+import cv2
 import numpy as np
-from typing import Optional
-import uvicorn
-import albumentations as A
-from albumentations.pytorch import ToTensorV2
-from transformers import SegformerConfig, SegformerForSemanticSegmentation
+import torch
+import torch.nn as nn
+from efficientnet_pytorch import EfficientNet
+from fastapi import FastAPI, File, HTTPException, UploadFile
+from fastapi.middleware.cors import CORSMiddleware
+
 
-app = FastAPI(title="DuneNet Model API", version="1.0.0")
+app = FastAPI(title="Deepfake Detection API", version="2.0.0")
 
-# CORS middleware for Next.js frontend
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=[
-        "http://localhost:3000",
-        "https://*.vercel.app",
-        "*",
-    ],
+    allow_origins=["*"],
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-# Configuration
-NUM_CLASSES = 10
-IMG_SIZE = 512
-MODEL_NAME = 'nvidia/mit-b4'
-
-CLASS_NAMES = [
-    'Trees', 'Lush Bushes', 'Dry Grass', 'Dry Bushes', 'Ground Clutter',
-    'Flowers', 'Logs', 'Rocks', 'Landscape', 'Sky'
-]
-
-CLASS_COLORS = np.array([
-    [34, 139, 34],    # Trees
-    [0, 255, 127],    # Lush Bushes
-    [189, 183, 107],  # Dry Grass
-    [139, 119, 101],  # Dry Bushes
-    [160, 82, 45],    # Ground Clutter
-    [255, 105, 180],  # Flowers
-    [139, 69, 19],    # Logs
-    [128, 128, 128],  # Rocks
-    [210, 180, 140],  # Landscape
-    [135, 206, 235],  # Sky
-], dtype=np.uint8)
-
-# Traversability mapping
-TRAVERSABILITY = {
-    0: 'no_go',    # Trees
-    1: 'no_go',    # Lush Bushes
-    2: 'go',       # Dry Grass
-    3: 'caution',  # Dry Bushes
-    4: 'caution',  # Ground Clutter
-    5: 'go',       # Flowers
-    6: 'no_go',    # Logs
-    7: 'caution',  # Rocks
-    8: 'go',       # Landscape
-    9: 'sky',      # Sky
-}
-
-TRAV_COLORS = {
-    'go':      np.array([0, 200, 0], dtype=np.uint8),      # Green
-    'caution': np.array([255, 180, 0], dtype=np.uint8),    # Orange
-    'no_go':   np.array([220, 30, 30], dtype=np.uint8),    # Red
-    'sky':     np.array([180, 210, 240], dtype=np.uint8),  # Light blue
-}
-
-# Global model variable
+
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+DETECTION_THRESHOLD = 0.40
+
+
+class DeepfakeEfficientNet(nn.Module):
+    def __init__(self, pretrained: bool = True, dropout: float = 0.5):
+        super().__init__()
+        if pretrained:
+            self.net = EfficientNet.from_pretrained("efficientnet-b0")
+        else:
+            self.net = EfficientNet.from_name("efficientnet-b0")
+
+        in_features = self.net._fc.in_features
+        self.net._fc = nn.Sequential(
+            nn.Dropout(dropout),
+            nn.Linear(in_features, 512),
+            nn.BatchNorm1d(512),
+            nn.ReLU(),
+            nn.Dropout(dropout * 0.7),
+            nn.Linear(512, 256),
+            nn.BatchNorm1d(256),
+            nn.ReLU(),
+            nn.Dropout(dropout * 0.5),
+            nn.Linear(256, 1),
+        )
+
+    def forward(self, rgb_input, freq_input=None):
+        return self.net(rgb_input)
+
+
+class TemporalTracker:
+    def __init__(self, window_size: int = 60, voting_window: int = 10, threshold: float = DETECTION_THRESHOLD):
+        self.window_size = window_size
+        self.voting_window = voting_window
+        self.threshold = threshold
+        self.score_history = []
+        self.frame_votes = []
+
+    def update(self, fake_probability: float):
+        self.score_history.append(float(fake_probability))
+        if len(self.score_history) > self.window_size:
+            self.score_history = self.score_history[-self.window_size :]
+
+        vote = "FAKE" if fake_probability > self.threshold else "REAL"
+        self.frame_votes.append(vote)
+        if len(self.frame_votes) > self.voting_window:
+            self.frame_votes = self.frame_votes[-self.voting_window :]
+
+    def get_temporal_average(self) -> float:
+        if not self.score_history:
+            return 0.0
+        return float(sum(self.score_history) / len(self.score_history))
+
+    def get_stability_score(self) -> float:
+        if len(self.score_history) < 10:
+            return 0.0
+        arr = np.array(self.score_history[-10:], dtype=np.float32)
+        variance = float(np.var(arr))
+        return float(1.0 - min(variance * 4.0, 1.0))
+
+    def get_confidence_level(self) -> str:
+        if len(self.frame_votes) < self.voting_window:
+            return "UNCERTAIN"
+        fake_count = sum(1 for x in self.frame_votes if x == "FAKE")
+        real_count = len(self.frame_votes) - fake_count
+        return "FAKE" if fake_count > real_count else "REAL"
+
+    def reset(self):
+        self.score_history = []
+        self.frame_votes = []
+
+
+class ForensicAnalyzer:
+    def __init__(self, analysis_size=(256, 256)):
+        self.analysis_size = analysis_size
+        self.prev_gray = None
+
+    def analyze(self, frame_bgr: np.ndarray):
+        resized = cv2.resize(frame_bgr, self.analysis_size, interpolation=cv2.INTER_LINEAR)
+
+        frequency = self._analyze_frequency(resized)
+        noise = self._analyze_noise(resized)
+        ela = self._analyze_ela(resized)
+        edge = self._analyze_edges(resized)
+        temporal = self._analyze_temporal(resized)
+
+        score = (
+            0.32 * frequency
+            + 0.20 * noise
+            + 0.18 * ela
+            + 0.18 * edge
+            + 0.12 * temporal
+        )
+
+        return {
+            "fake_probability": float(np.clip(score, 0.0, 1.0)),
+            "scores": {
+                "frequency": float(frequency),
+                "noise": float(noise),
+                "ela": float(ela),
+                "edge": float(edge),
+                "temporal": float(temporal),
+            },
+        }
+
+    def _analyze_frequency(self, frame):
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY).astype(np.float32)
+        magnitude = np.log1p(np.abs(np.fft.fftshift(np.fft.fft2(gray))))
+        h, w = magnitude.shape
+        cy, cx = h // 2, w // 2
+
+        y_grid, x_grid = np.ogrid[:h, :w]
+        dist = np.sqrt((x_grid - cx) ** 2 + (y_grid - cy) ** 2)
+
+        inner = min(h, w) // 8
+        outer = min(h, w) // 3
+        low = magnitude[dist <= inner]
+        high = magnitude[(dist > inner) & (dist <= outer)]
+
+        low_mean = float(low.mean()) if low.size else 0.0
+        high_mean = float(high.mean()) if high.size else 0.0
+        ratio = high_mean / (low_mean + high_mean + 1e-9)
+
+        if ratio < 0.18:
+            return 0.75
+        if ratio < 0.24:
+            return 0.45
+        return 0.10
+
+    def _analyze_noise(self, frame):
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY).astype(np.float32)
+        blurred = cv2.GaussianBlur(gray, (5, 5), 0)
+        residual = gray - blurred
+        std = float(np.std(residual))
+
+        if std < 2.0:
+            return 0.70
+        if std < 4.0:
+            return 0.35
+        return 0.12
+
+    def _analyze_ela(self, frame):
+        encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 90]
+        ok, encoded = cv2.imencode(".jpg", frame, encode_param)
+        if not ok:
+            return 0.0
+
+        recompressed = cv2.imdecode(encoded, cv2.IMREAD_COLOR)
+        if recompressed is None:
+            return 0.0
+
+        diff = cv2.absdiff(frame, recompressed)
+        mean_diff = float(np.mean(diff))
+
+        if mean_diff > 14:
+            return 0.65
+        if mean_diff > 8:
+            return 0.35
+        return 0.08
+
+    def _analyze_edges(self, frame):
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        edges = cv2.Canny(gray, 50, 150)
+        edge_density = float(np.mean(edges > 0))
+        lap_var = float(np.var(cv2.Laplacian(gray, cv2.CV_64F)))
+
+        score = 0.0
+        if edge_density < 0.02:
+            score += 0.45
+        elif edge_density < 0.04:
+            score += 0.20
+
+        if lap_var < 60:
+            score += 0.35
+        elif lap_var < 120:
+            score += 0.15
+
+        return float(np.clip(score, 0.0, 1.0))
+
+    def _analyze_temporal(self, frame):
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY).astype(np.float32)
+        if self.prev_gray is None:
+            self.prev_gray = gray
+            return 0.0
+
+        diff = cv2.absdiff(gray, self.prev_gray)
+        self.prev_gray = gray
+        mean_delta = float(np.mean(diff))
+
+        if mean_delta < 1.2:
+            return 0.40
+        if mean_delta < 2.5:
+            return 0.20
+        return 0.08
+
+    def reset(self):
+        self.prev_gray = None
+
+
 model = None
-device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-
-class PredictionResponse(BaseModel):
-    prediction: int
-    class_name: str
-    confidence: float
-    device_used: str
-    class_distribution: dict
-    segmentation_mask: str  # base64 encoded image
-    overlay_image: str  # base64 encoded overlay
-    traversability_map: str  # base64 encoded traversability
-    traversability_overlay: str  # base64 encoded traversability overlay
-    traversability_stats: dict  # safe, caution, blocked percentages
-
-class HealthResponse(BaseModel):
-    status: str
-    model_loaded: bool
-    device: str
+model_loaded = False
+tracker = TemporalTracker()
+forensics = ForensicAnalyzer()
+frame_count = 0
 
-@app.on_event("startup")
-async def load_model():
-    """Load the Segformer model on startup"""
+
+def load_checkpoint_model():
     global model
-    try:
-        import os
-        possible_paths = [
-            "models/latest_model_ft.pth",
-            "api_server/models/latest_model_ft.pth",
-            os.path.join(os.path.dirname(__file__), "models/latest_model_ft.pth"),
-        ]
-        
-        model_path = None
-        for path in possible_paths:
-            if os.path.exists(path):
-                model_path = path
-                break
-        
-        if model_path is None:
-            raise FileNotFoundError("latest_model_ft.pth not found in api_server/models/")
-        
-        print(f"Loading Segformer model from: {os.path.abspath(model_path)}")
-        
-        # Build Segformer model
-        config = SegformerConfig.from_pretrained(MODEL_NAME)
-        config.num_labels = NUM_CLASSES
-        model = SegformerForSemanticSegmentation(config)
-        
-        # Load checkpoint
-        checkpoint = torch.load(model_path, map_location=device, weights_only=False)
-        model.load_state_dict(checkpoint['model_state_dict'])
-        
-        model = model.to(device)
+    global model_loaded
+
+    checkpoint_candidates = [
+        os.path.join(os.path.dirname(__file__), "models", "best_model.pth"),
+        os.path.join(os.path.dirname(__file__), "models", "latest_model_ft.pth"),
+    ]
+
+    model = DeepfakeEfficientNet(pretrained=True).to(DEVICE)
+
+    loaded_any = False
+    for path in checkpoint_candidates:
+        if not os.path.exists(path):
+            continue
+
+        try:
+            checkpoint = torch.load(path, map_location=DEVICE, weights_only=False)
+            state_dict = checkpoint.get("model_state_dict", checkpoint)
+            model.load_state_dict(state_dict, strict=False)
+            loaded_any = True
+            print(f"Loaded checkpoint: {path}")
+            break
+        except Exception as ex:
+            print(f"Failed loading checkpoint {path}: {ex}")
+
+    if loaded_any:
         model.eval()
-        
-        miou = checkpoint.get('miou', 0)
-        epoch = checkpoint.get('epoch', '?')
-        
-        print(f"✓ Segformer model loaded successfully on {device}")
-        print(f"  Epoch: {epoch}, Val mIoU: {miou:.4f}")
-        print(f"  Classes: {NUM_CLASSES}")
-        print(f"  Model: {MODEL_NAME}")
-        
-    except Exception as e:
-        print(f"✗ Error loading model: {e}")
-        import traceback
-        traceback.print_exc()
-        model = None
-
-@app.get("/", response_model=HealthResponse)
-async def health_check():
-    """Health check endpoint"""
+        model_loaded = True
+    else:
+        model_loaded = False
+        print("No compatible deepfake checkpoint found; running forensic-only mode.")
+
+
+@app.on_event("startup")
+async def startup_event():
+    load_checkpoint_model()
+
+
+@app.get("/")
+async def root_health():
     return {
-        "status": "running",
-        "model_loaded": model is not None,
-        "device": str(device)
+        "status": "healthy",
+        "model_loaded": model_loaded,
+        "device": DEVICE,
+        "frame_count": frame_count,
     }
 
-def colorize_mask(class_mask):
-    """Convert class mask to RGB colored image"""
-    h, w = class_mask.shape
-    rgb = np.zeros((h, w, 3), dtype=np.uint8)
-    for c in range(NUM_CLASSES):
-        rgb[class_mask == c] = CLASS_COLORS[c]
-    return rgb
-
-
-def create_overlay(image_np, class_mask, alpha=0.5):
-    """Blend original image with colored segmentation mask"""
-    colored = colorize_mask(class_mask)
-    overlay = (image_np.astype(np.float32) * (1 - alpha) + colored.astype(np.float32) * alpha)
-    return overlay.astype(np.uint8)
-
-
-def create_traversability_map(class_mask):
-    """Generate traversability map from segmentation mask"""
-    h, w = class_mask.shape
-    trav_mask = np.zeros((h, w, 3), dtype=np.uint8)
-    
-    for class_id, category in TRAVERSABILITY.items():
-        region = (class_mask == class_id)
-        trav_mask[region] = TRAV_COLORS[category]
-    
-    return trav_mask
-
-
-def calculate_traversability_stats(class_mask):
-    """Calculate traversability statistics"""
-    total_pixels = class_mask.size
-    sky_pixels = (class_mask == 9).sum()  # Sky class
-    ground_pixels = total_pixels - sky_pixels
-    
-    if ground_pixels == 0:
-        return {'safe': '0%', 'caution': '0%', 'blocked': '0%'}
-    
-    safe_pixels = 0
-    caution_pixels = 0
-    blocked_pixels = 0
-    
-    for class_id, category in TRAVERSABILITY.items():
-        if category == 'sky':
-            continue
-        count = (class_mask == class_id).sum()
-        if category == 'go':
-            safe_pixels += count
-        elif category == 'caution':
-            caution_pixels += count
-        elif category == 'no_go':
-            blocked_pixels += count
-    
+
+@app.get("/health")
+async def health_check():
     return {
-        'safe': f"{(safe_pixels / ground_pixels * 100):.1f}%",
-        'caution': f"{(caution_pixels / ground_pixels * 100):.1f}%",
-        'blocked': f"{(blocked_pixels / ground_pixels * 100):.1f}%"
+        "status": "healthy",
+        "model_loaded": model_loaded,
+        "device": DEVICE,
+        "frame_count": frame_count,
+        "capabilities": {
+            "frame_forensics": True,
+            "temporal_tracking": True,
+            "face_detection": False,
+        },
     }
 
 
-def numpy_to_base64(image_np):
-    """Convert numpy array to base64 string"""
-    img = Image.fromarray(image_np)
-    buffered = io.BytesIO()
-    img.save(buffered, format="PNG")
-    img_str = base64.b64encode(buffered.getvalue()).decode()
-    return f"data:image/png;base64,{img_str}"
+@app.post("/reset")
+async def reset_state():
+    global frame_count
+
+    tracker.reset()
+    forensics.reset()
+    frame_count = 0
+    return {"success": True, "message": "Detector state reset"}
+
+
+def _prepare_model_tensor(frame_bgr: np.ndarray) -> torch.Tensor:
+    rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)
+    resized = cv2.resize(rgb, (224, 224), interpolation=cv2.INTER_AREA)
+    arr = resized.astype(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)
+    arr = (arr - mean) / std
+
+    chw = np.transpose(arr, (2, 0, 1))
+    tensor = torch.from_numpy(chw).unsqueeze(0).to(DEVICE)
+    return tensor
+
+
+def _run_model(frame_bgr: np.ndarray):
+    if not model_loaded or model is None:
+        return None
 
-@app.post("/predict", response_model=PredictionResponse)
-async def predict(file: UploadFile = File(...)):
-    """Make prediction on uploaded image using Segformer"""
-    if model is None:
-        raise HTTPException(status_code=503, detail="Model not loaded")
-    
     try:
-        # Read and process image
-        contents = await file.read()
-        image = Image.open(io.BytesIO(contents)).convert('RGB')
-        image_np = np.array(image)
-        orig_h, orig_w = image_np.shape[:2]
-        
-        # Preprocessing with albumentations
-        transform = A.Compose([
-            A.Resize(height=IMG_SIZE, width=IMG_SIZE),
-            A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
-            ToTensorV2(),
-        ])
-        
-        aug = transform(image=image_np)
-        tensor = aug['image'].unsqueeze(0).to(device)
-        
-        # Inference
+        tensor = _prepare_model_tensor(frame_bgr)
         with torch.no_grad():
-            use_fp16 = device.type == 'cuda'
-            with torch.amp.autocast(device_type=device.type, enabled=use_fp16):
-                outputs = model(pixel_values=tensor)
-            
-            # Get logits and resize
-            logits = F.interpolate(
-                outputs.logits,
-                size=(IMG_SIZE, IMG_SIZE),
-                mode='bilinear',
-                align_corners=False
-            )
-            
-            # Get probabilities
-            probs = torch.softmax(logits, dim=1).squeeze().cpu().numpy()
-            
-            # Get prediction mask
-            pred_mask = np.argmax(probs, axis=0).astype(np.uint8)
-            
-            # Resize prediction to original image size
-            pred_mask_orig = np.array(
-                Image.fromarray(pred_mask).resize((orig_w, orig_h), Image.NEAREST)
-            )
-            
-            # Calculate class distribution
-            class_dist = {}
-            total_pixels = pred_mask_orig.size
-            for c in range(NUM_CLASSES):
-                count = (pred_mask_orig == c).sum()
-                if count > 0:
-                    class_dist[CLASS_NAMES[c]] = f"{(count / total_pixels * 100):.1f}%"
-            
-            # Get dominant class
-            dominant_class = np.bincount(pred_mask_orig.flatten()).argmax()
-            confidence = probs[dominant_class].mean()
-            
-            # Generate visualizations
-            colored_mask = colorize_mask(pred_mask_orig)
-            overlay = create_overlay(image_np, pred_mask_orig, alpha=0.5)
-            
-            # Generate traversability map
-            print(f"Generating traversability map...")
-            trav_map = create_traversability_map(pred_mask_orig)
-            print(f"Traversability map shape: {trav_map.shape}")
-            
-            trav_overlay = create_overlay(image_np, pred_mask_orig, alpha=0.6)
-            # Replace with traversability colors
-            for class_id, category in TRAVERSABILITY.items():
-                region = (pred_mask_orig == class_id)
-                trav_overlay[region] = (
-                    image_np[region].astype(np.float32) * 0.4 + 
-                    TRAV_COLORS[category].astype(np.float32) * 0.6
-                ).astype(np.uint8)
-            
-            trav_stats = calculate_traversability_stats(pred_mask_orig)
-            print(f"Traversability stats: {trav_stats}")
-            
-            # Convert to base64
-            mask_base64 = numpy_to_base64(colored_mask)
-            overlay_base64 = numpy_to_base64(overlay)
-            trav_map_base64 = numpy_to_base64(trav_map)
-            trav_overlay_base64 = numpy_to_base64(trav_overlay)
-            print(f"All images converted to base64 successfully")
-        
-        return {
-            "prediction": int(dominant_class),
-            "class_name": CLASS_NAMES[dominant_class],
-            "confidence": float(confidence),
-            "device_used": str(device),
-            "class_distribution": class_dist,
-            "segmentation_mask": mask_base64,
-            "overlay_image": overlay_base64,
-            "traversability_map": trav_map_base64,
-            "traversability_overlay": trav_overlay_base64,
-            "traversability_stats": trav_stats
-        }
-    except Exception as e:
-        import traceback
-        traceback.print_exc()
-        raise HTTPException(status_code=500, detail=f"Prediction error: {str(e)}")
-
-
-# ═══════════════════════════════════════════════════════════════
-#  Simulation Live Inference
-# ═══════════════════════════════════════════════════════════════
-
-class SimPredictionResponse(BaseModel):
-    segmentation_mask: str
-    traversability_map: str
-    traversability_overlay: str
-    traversability_stats: dict
-    traversability_grid: list
-    class_distribution: dict
-    inference_time_ms: float
-    dominant_class: str
-    confidence: float
-
-
-def create_traversability_grid(class_mask, grid_cols=12, grid_rows=8):
-    """Create a coarse traversability grid from the prediction mask.
-    Uses the bottom 65 % of the image (ground portion, excluding sky).
-    Returns 2-D list of costmap values: 0 = go, 5 = caution, 10 = no_go.
-    """
-    h, w = class_mask.shape
-    ground_start = int(h * 0.35)
-    ground_mask = class_mask[ground_start:, :]
-    gh, gw = ground_mask.shape
-
-    cell_h = max(1, gh // grid_rows)
-    cell_w = max(1, gw // grid_cols)
-
-    grid = []
-    for r in range(grid_rows):
-        row = []
-        for c in range(grid_cols):
-            y0 = r * cell_h
-            y1 = min((r + 1) * cell_h, gh)
-            x0 = c * cell_w
-            x1 = min((c + 1) * cell_w, gw)
-
-            cell = ground_mask[y0:y1, x0:x1]
-            if cell.size == 0:
-                row.append(0)
-                continue
-
-            go_count = caution_count = no_go_count = 0
-            for cid in range(NUM_CLASSES):
-                cnt = int((cell == cid).sum())
-                cat = TRAVERSABILITY[cid]
-                if cat == 'go':
-                    go_count += cnt
-                elif cat == 'caution':
-                    caution_count += cnt
-                elif cat == 'no_go':
-                    no_go_count += cnt
-
-            total = go_count + caution_count + no_go_count
-            if total == 0:
-                row.append(0)
-            elif no_go_count / total > 0.3:
-                row.append(10)
-            elif caution_count / total > 0.3:
-                row.append(5)
-            else:
-                row.append(0)
-        grid.append(row)
-    return grid
-
-
-@app.post("/predict/sim", response_model=SimPredictionResponse)
-async def predict_sim(file: UploadFile = File(...)):
-    """Prediction endpoint optimised for simulation live inference.
-    Returns a traversability grid suitable for direct costmap updates."""
-    if model is None:
-        raise HTTPException(status_code=503, detail="Model not loaded")
-
-    import time
-    t0 = time.time()
+            logits = model(tensor).squeeze()
+            prob = torch.sigmoid(logits).item()
+        return float(np.clip(prob, 0.0, 1.0))
+    except Exception as ex:
+        print(f"Model inference failed: {ex}")
+        return None
 
-    try:
-        contents = await file.read()
-        image = Image.open(io.BytesIO(contents)).convert('RGB')
-        image_np = np.array(image)
-        orig_h, orig_w = image_np.shape[:2]
 
-        transform = A.Compose([
-            A.Resize(height=IMG_SIZE, width=IMG_SIZE),
-            A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
-            ToTensorV2(),
-        ])
+@app.post("/analyze")
+async def analyze_frame(frame: UploadFile = File(None), file: UploadFile = File(None)):
+    global frame_count
 
-        aug = transform(image=image_np)
-        tensor = aug['image'].unsqueeze(0).to(device)
+    start = time.time()
 
-        with torch.no_grad():
-            use_fp16 = device.type == 'cuda'
-            with torch.amp.autocast(device_type=device.type, enabled=use_fp16):
-                outputs = model(pixel_values=tensor)
-
-            logits = F.interpolate(
-                outputs.logits,
-                size=(IMG_SIZE, IMG_SIZE),
-                mode='bilinear',
-                align_corners=False,
-            )
-
-            probs = torch.softmax(logits, dim=1).squeeze().cpu().numpy()
-            pred_mask = np.argmax(probs, axis=0).astype(np.uint8)
-            pred_mask_orig = np.array(
-                Image.fromarray(pred_mask).resize((orig_w, orig_h), Image.NEAREST)
-            )
-
-        # Visualisations
-        colored_mask = colorize_mask(pred_mask_orig)
-        trav_map = create_traversability_map(pred_mask_orig)
-
-        trav_overlay_img = image_np.copy()
-        for cid, category in TRAVERSABILITY.items():
-            region = (pred_mask_orig == cid)
-            trav_overlay_img[region] = (
-                image_np[region].astype(np.float32) * 0.4
-                + TRAV_COLORS[category].astype(np.float32) * 0.6
-            ).astype(np.uint8)
-
-        trav_stats = calculate_traversability_stats(pred_mask_orig)
-        trav_grid = create_traversability_grid(pred_mask_orig)
-
-        class_dist = {}
-        total_pixels = pred_mask_orig.size
-        for cid in range(NUM_CLASSES):
-            cnt = int((pred_mask_orig == cid).sum())
-            if cnt > 0:
-                class_dist[CLASS_NAMES[cid]] = f"{cnt / total_pixels * 100:.1f}%"
-
-        dominant = int(np.bincount(pred_mask_orig.flatten()).argmax())
-        conf = float(probs[dominant].mean())
-        elapsed = (time.time() - t0) * 1000
+    uploaded = frame or file
+    if uploaded is None:
+        raise HTTPException(status_code=400, detail="No frame provided. Use multipart form field 'frame'.")
+
+    raw = await uploaded.read()
+    if not raw:
+        raise HTTPException(status_code=400, detail="Empty file")
+
+    np_bytes = np.frombuffer(raw, np.uint8)
+    image = cv2.imdecode(np_bytes, cv2.IMREAD_COLOR)
+    if image is None:
+        raise HTTPException(status_code=400, detail="Invalid image format")
+
+    frame_count += 1
+
+    forensic_result = forensics.analyze(image)
+    forensic_prob = float(forensic_result["fake_probability"])
+
+    model_prob = _run_model(image)
+    if model_prob is None:
+        combined_prob = forensic_prob
+        analysis_mode = "frame_only"
+    else:
+        combined_prob = float(np.clip(0.70 * model_prob + 0.30 * forensic_prob, 0.0, 1.0))
+        analysis_mode = "model+frame"
+
+    tracker.update(combined_prob)
+
+    elapsed_ms = (time.time() - start) * 1000.0
 
-        return {
-            "segmentation_mask": numpy_to_base64(colored_mask),
-            "traversability_map": numpy_to_base64(trav_map),
-            "traversability_overlay": numpy_to_base64(trav_overlay_img),
-            "traversability_stats": trav_stats,
-            "traversability_grid": trav_grid,
-            "class_distribution": class_dist,
-            "inference_time_ms": round(elapsed, 1),
-            "dominant_class": CLASS_NAMES[dominant],
-            "confidence": conf,
-        }
-    except Exception as e:
-        import traceback
-        traceback.print_exc()
-        raise HTTPException(status_code=500, detail=f"Sim prediction error: {str(e)}")
-
-
-@app.get("/model/info")
-async def model_info():
-    """Get model information"""
-    if model is None:
-        raise HTTPException(status_code=503, detail="Model not loaded")
-    
     return {
-        "model_type": str(type(model).__name__),
-        "device": str(device),
-        "parameters": sum(p.numel() for p in model.parameters() if hasattr(model, 'parameters'))
+        "success": True,
+        "analysis_mode": analysis_mode,
+        "faces_detected": 0,
+        "fake_probability": combined_prob,
+        "model_probability": model_prob,
+        "frame_forensic_probability": forensic_prob,
+        "real_probability": float(1.0 - combined_prob),
+        "confidence_level": tracker.get_confidence_level(),
+        "temporal_average": tracker.get_temporal_average(),
+        "stability_score": tracker.get_stability_score(),
+        "frame_count": frame_count,
+        "processing_time_ms": round(elapsed_ms, 1),
+        "forensic_scores": forensic_result["scores"],
     }
 
+
 if __name__ == "__main__":
-    import os
+    import uvicorn
+
     port = int(os.environ.get("PORT", 7860))
     uvicorn.run(app, host="0.0.0.0", port=port)
+

requirements.txt

@@ -1,10 +1,7 @@
 fastapi==0.109.0
 uvicorn[standard]==0.27.0
-torch>=2.6.0
-torchvision>=0.21.0
-pillow>=10.2.0
-python-multipart>=0.0.6
 numpy>=1.26.0
-pydantic>=2.5.0
-albumentations>=1.3.0
-transformers>=4.30.0
+python-multipart>=0.0.6
+opencv-python-headless>=4.10.0
+efficientnet-pytorch>=0.7.1
+