Rewriting the complete backend with gradcam and explainability support

backend/app.py

@@ -0,0 +1,206 @@
+from flask import Flask, request, jsonify
+from flask_cors import CORS
+from werkzeug.utils import secure_filename
+import os
+import sys
+import torch
+import re
+import numpy as np
+from PIL import Image
+
+# Add the dfdc_deepfake_challenge directory to the path
+sys.path.append('dfdc_model')
+
+from dfdc_model.kernel_utils import VideoReader, FaceExtractor, confident_strategy, predict_on_video
+from dfdc_model.training.zoo.classifiers import DeepFakeClassifier
+from enhanced_prediction import enhanced_predict_on_video, enhanced_predict_on_image
+
+app = Flask(__name__)
+CORS(app)
+app.config['UPLOAD_FOLDER'] = 'uploads/'
+os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
+
+# Initialize the model and preprocessing pipeline
+def load_models():
+    models = []
+    weights_dir = "dfdc_deepfake_challenge/weights"
+    # Update these model names based on your downloaded weights
+    model_files = [
+        "final_888_DeepFakeClassifier_tf_efficientnet_b7_ns_0_40",
+        # Add more model files if you have them
+    ]
+    
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    
+    for model_file in model_files:
+        model_path = os.path.join(weights_dir, model_file)
+        if os.path.exists(model_path):
+            model = DeepFakeClassifier(encoder="tf_efficientnet_b7_ns").to(device)
+            print(f"Loading state dict {model_path}")
+            checkpoint = torch.load(model_path, map_location="cpu")
+            state_dict = checkpoint.get("state_dict", checkpoint)
+            model.load_state_dict({re.sub("^module.", "", k): v for k, v in state_dict.items()}, strict=True)
+            model.eval()
+            del checkpoint
+            if device == "cuda":
+                models.append(model.half())
+            else:
+                models.append(model)
+    
+    return models
+
+# Initialize the preprocessing pipeline
+def initialize_preprocessing():
+    frames_per_video = 32
+    video_reader = VideoReader()
+    video_read_fn = lambda x: video_reader.read_frames(x, num_frames=frames_per_video)
+    face_extractor = FaceExtractor(video_read_fn)
+    input_size = 380
+    return face_extractor, input_size, frames_per_video
+
+# Load models and initialize preprocessing
+print("Loading models...")
+models = load_models()
+face_extractor, input_size, frames_per_video = initialize_preprocessing()
+print(f"Loaded {len(models)} models")
+
+def predict_on_image(image_path):
+    """Predict on a single image using face extraction and model inference"""
+    try:
+        from dfdc_model.kernel_utils import isotropically_resize_image, put_to_center, normalize_transform
+        import numpy as np
+        import cv2
+        
+        # Read the image
+        image = cv2.imread(image_path)
+        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
+        
+        # Use MTCNN to detect faces
+        from PIL import Image as PILImage
+        img_pil = PILImage.fromarray(image)
+        img_pil = img_pil.resize(size=[s // 2 for s in img_pil.size])
+        
+        device = "cuda" if torch.cuda.is_available() else "cpu"
+        detector = face_extractor.detector
+        
+        batch_boxes, probs = detector.detect(img_pil, landmarks=False)
+        
+        if batch_boxes is None:
+            return 0.5  # No faces detected, return neutral
+            
+        # Process the largest/most confident face
+        best_idx = 0
+        if len(probs) > 1:
+            best_idx = np.argmax(probs)
+            
+        bbox = batch_boxes[best_idx]
+        if bbox is not None:
+            xmin, ymin, xmax, ymax = [int(b * 2) for b in bbox]
+            w = xmax - xmin
+            h = ymax - ymin
+            p_h = h // 3
+            p_w = w // 3
+            crop = image[max(ymin - p_h, 0):ymax + p_h, max(xmin - p_w, 0):xmax + p_w]
+            
+            # Resize and center the face
+            resized_face = isotropically_resize_image(crop, input_size)
+            resized_face = put_to_center(resized_face, input_size)
+            
+            # Prepare for model inference
+            x = torch.tensor(resized_face, device=device).float().unsqueeze(0)
+            x = x.permute((0, 3, 1, 2))
+            x = normalize_transform(x[0] / 255.).unsqueeze(0)
+            
+            # Run prediction
+            with torch.no_grad():
+                preds = []
+                for model in models:
+                    if device == "cuda":
+                        y_pred = model(x.half())
+                    else:
+                        y_pred = model(x)
+                    y_pred = torch.sigmoid(y_pred.squeeze())
+                    preds.append(y_pred.cpu().numpy())
+                return np.mean(preds)
+        
+        return 0.5
+    except Exception as e:
+        print(f"Error predicting on image: {e}")
+        return 0.5
+
+@app.route('/')
+def index():
+    return jsonify({'message': 'Deepfake Detection Flask API is running.'})
+
+@app.route('/analyze/image', methods=['POST'])
+def analyze_image():
+    if 'file' not in request.files:
+        return jsonify({'error': 'No file part'}), 400
+    file = request.files['file']
+    if file.filename == '':
+        return jsonify({'error': 'No selected file'}), 400
+    filename = secure_filename(file.filename)
+    filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
+    file.save(filepath)
+    
+    try:
+        # Use enhanced image prediction with Grad-CAM
+        result = enhanced_predict_on_image(filepath, models, input_size)
+        
+        label = 'fake' if result['prediction'] > 0.5 else 'real'
+        response = {
+            'label': label,
+            'confidence': float(result['prediction']),
+            'details': f'Image analyzed using {len(models)} models with face detection and Grad-CAM',
+            'faces_detected': 1,
+            'processing_time': 'N/A',
+            'gradcam_image': result['gradcam_image'],
+            'fake_frames': []  # Not applicable for images
+        }
+        return jsonify(response)
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+@app.route('/analyze/video', methods=['POST'])
+def analyze_video():
+    if 'file' not in request.files:
+        return jsonify({'error': 'No file part'}), 400
+    file = request.files['file']
+    if file.filename == '':
+        return jsonify({'error': 'No selected file'}), 400
+    filename = secure_filename(file.filename)
+    filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
+    file.save(filepath)
+    
+    try:
+        # Use enhanced video prediction with frame analysis and Grad-CAM
+        result = enhanced_predict_on_video(
+            face_extractor=face_extractor,
+            video_path=filepath,
+            batch_size=frames_per_video,
+            input_size=input_size,
+            models=models,
+            strategy=confident_strategy,
+            apply_compression=False,
+            return_frame_predictions=True
+        )
+        
+        prediction = result['prediction']
+        label = 'fake' if prediction > 0.5 else 'real'
+        
+        response = {
+            'label': label,
+            'confidence': float(prediction),
+            'details': f'Video analyzed using {len(models)} models with confident strategy and frame analysis',
+            'faces_detected': len(result.get('frame_predictions', [])),
+            'processing_time': 'N/A',
+            'fake_frames': result.get('fake_frames', []),
+            'gradcam_image': result.get('gradcam_images', [None])[0],  # Return first Grad-CAM image
+            'frame_predictions': result.get('frame_predictions', [])
+        }
+        return jsonify(response)
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=8000)

backend/dfdc_model/configs/b5.json

@@ -0,0 +1,28 @@
+{
+    "network": "DeepFakeClassifier",
+    "encoder": "tf_efficientnet_b5_ns",
+    "batches_per_epoch": 2500,
+    "size": 380,
+    "fp16": true,
+    "optimizer": {
+        "batch_size": 20,
+        "type": "SGD",
+        "momentum": 0.9,
+        "weight_decay": 1e-4,
+        "learning_rate": 0.01,
+        "nesterov": true,
+        "schedule": {
+            "type": "poly",
+            "mode": "step",
+            "epochs": 30,
+            "params": {"max_iter":  75100}
+        }
+    },
+    "normalize": {
+        "mean": [0.485, 0.456, 0.406],
+        "std": [0.229, 0.224, 0.225]
+    },
+    "losses": {
+        "BinaryCrossentropy": 1
+    }
+}

backend/dfdc_model/configs/b7.json

@@ -0,0 +1,29 @@
+{
+    "network": "DeepFakeClassifier",
+    "encoder": "tf_efficientnet_b7_ns",
+    "batches_per_epoch": 2500,
+    "size": 380,
+    "fp16": true,
+    "optimizer": {
+        "batch_size": 12,
+        "type": "SGD",
+        "momentum": 0.9,
+        "weight_decay": 1e-4,
+        "learning_rate": 0.01,
+        "nesterov": true,
+        "schedule": {
+            "type": "poly",
+            "mode": "step",
+            "epochs": 40,
+            "params": {"max_iter":  100500}
+        }
+    },
+    "normalize": {
+        "mean": [0.485, 0.456, 0.406],
+        "std": [0.229, 0.224, 0.225]
+    },
+    "losses": {
+        "BinaryCrossentropy": 1
+    }
+}
+

backend/dfdc_model/kernel_utils.py

@@ -0,0 +1,359 @@
+import os
+
+import cv2
+import numpy as np
+import torch
+from PIL import Image
+from albumentations.augmentations.functional import image_compression
+from facenet_pytorch.models.mtcnn import MTCNN
+from concurrent.futures import ThreadPoolExecutor
+
+from torchvision.transforms import Normalize
+
+mean = [0.485, 0.456, 0.406]
+std = [0.229, 0.224, 0.225]
+normalize_transform = Normalize(mean, std)
+
+
+class VideoReader:
+    """Helper class for reading one or more frames from a video file."""
+
+    def __init__(self, verbose=True, insets=(0, 0)):
+        """Creates a new VideoReader.
+
+        Arguments:
+            verbose: whether to print warnings and error messages
+            insets: amount to inset the image by, as a percentage of
+                (width, height). This lets you "zoom in" to an image
+                to remove unimportant content around the borders.
+                Useful for face detection, which may not work if the
+                faces are too small.
+        """
+        self.verbose = verbose
+        self.insets = insets
+
+    def read_frames(self, path, num_frames, jitter=0, seed=None):
+        """Reads frames that are always evenly spaced throughout the video.
+
+        Arguments:
+            path: the video file
+            num_frames: how many frames to read, -1 means the entire video
+                (warning: this will take up a lot of memory!)
+            jitter: if not 0, adds small random offsets to the frame indices;
+                this is useful so we don't always land on even or odd frames
+            seed: random seed for jittering; if you set this to a fixed value,
+                you probably want to set it only on the first video
+        """
+        assert num_frames > 0
+
+        capture = cv2.VideoCapture(path)
+        frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
+        if frame_count <= 0: return None
+
+        frame_idxs = np.linspace(0, frame_count - 1, num_frames, endpoint=True, dtype=int)
+        if jitter > 0:
+            np.random.seed(seed)
+            jitter_offsets = np.random.randint(-jitter, jitter, len(frame_idxs))
+            frame_idxs = np.clip(frame_idxs + jitter_offsets, 0, frame_count - 1)
+
+        result = self._read_frames_at_indices(path, capture, frame_idxs)
+        capture.release()
+        return result
+
+    def read_random_frames(self, path, num_frames, seed=None):
+        """Picks the frame indices at random.
+
+        Arguments:
+            path: the video file
+            num_frames: how many frames to read, -1 means the entire video
+                (warning: this will take up a lot of memory!)
+        """
+        assert num_frames > 0
+        np.random.seed(seed)
+
+        capture = cv2.VideoCapture(path)
+        frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
+        if frame_count <= 0: return None
+
+        frame_idxs = sorted(np.random.choice(np.arange(0, frame_count), num_frames))
+        result = self._read_frames_at_indices(path, capture, frame_idxs)
+
+        capture.release()
+        return result
+
+    def read_frames_at_indices(self, path, frame_idxs):
+        """Reads frames from a video and puts them into a NumPy array.
+
+        Arguments:
+            path: the video file
+            frame_idxs: a list of frame indices. Important: should be
+                sorted from low-to-high! If an index appears multiple
+                times, the frame is still read only once.
+
+        Returns:
+            - a NumPy array of shape (num_frames, height, width, 3)
+            - a list of the frame indices that were read
+
+        Reading stops if loading a frame fails, in which case the first
+        dimension returned may actually be less than num_frames.
+
+        Returns None if an exception is thrown for any reason, or if no
+        frames were read.
+        """
+        assert len(frame_idxs) > 0
+        capture = cv2.VideoCapture(path)
+        result = self._read_frames_at_indices(path, capture, frame_idxs)
+        capture.release()
+        return result
+
+    def _read_frames_at_indices(self, path, capture, frame_idxs):
+        try:
+            frames = []
+            idxs_read = []
+            for frame_idx in range(frame_idxs[0], frame_idxs[-1] + 1):
+                # Get the next frame, but don't decode if we're not using it.
+                ret = capture.grab()
+                if not ret:
+                    if self.verbose:
+                        print("Error grabbing frame %d from movie %s" % (frame_idx, path))
+                    break
+
+                # Need to look at this frame?
+                current = len(idxs_read)
+                if frame_idx == frame_idxs[current]:
+                    ret, frame = capture.retrieve()
+                    if not ret or frame is None:
+                        if self.verbose:
+                            print("Error retrieving frame %d from movie %s" % (frame_idx, path))
+                        break
+
+                    frame = self._postprocess_frame(frame)
+                    frames.append(frame)
+                    idxs_read.append(frame_idx)
+
+            if len(frames) > 0:
+                return np.stack(frames), idxs_read
+            if self.verbose:
+                print("No frames read from movie %s" % path)
+            return None
+        except:
+            if self.verbose:
+                print("Exception while reading movie %s" % path)
+            return None
+
+    def read_middle_frame(self, path):
+        """Reads the frame from the middle of the video."""
+        capture = cv2.VideoCapture(path)
+        frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
+        result = self._read_frame_at_index(path, capture, frame_count // 2)
+        capture.release()
+        return result
+
+    def read_frame_at_index(self, path, frame_idx):
+        """Reads a single frame from a video.
+
+        If you just want to read a single frame from the video, this is more
+        efficient than scanning through the video to find the frame. However,
+        for reading multiple frames it's not efficient.
+
+        My guess is that a "streaming" approach is more efficient than a
+        "random access" approach because, unless you happen to grab a keyframe,
+        the decoder still needs to read all the previous frames in order to
+        reconstruct the one you're asking for.
+
+        Returns a NumPy array of shape (1, H, W, 3) and the index of the frame,
+        or None if reading failed.
+        """
+        capture = cv2.VideoCapture(path)
+        result = self._read_frame_at_index(path, capture, frame_idx)
+        capture.release()
+        return result
+
+    def _read_frame_at_index(self, path, capture, frame_idx):
+        capture.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
+        ret, frame = capture.read()
+        if not ret or frame is None:
+            if self.verbose:
+                print("Error retrieving frame %d from movie %s" % (frame_idx, path))
+            return None
+        else:
+            frame = self._postprocess_frame(frame)
+            return np.expand_dims(frame, axis=0), [frame_idx]
+
+    def _postprocess_frame(self, frame):
+        frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+
+        if self.insets[0] > 0:
+            W = frame.shape[1]
+            p = int(W * self.insets[0])
+            frame = frame[:, p:-p, :]
+
+        if self.insets[1] > 0:
+            H = frame.shape[1]
+            q = int(H * self.insets[1])
+            frame = frame[q:-q, :, :]
+
+        return frame
+
+
+class FaceExtractor:
+    def __init__(self, video_read_fn):
+        self.video_read_fn = video_read_fn
+        self.detector = MTCNN(margin=0, thresholds=[0.7, 0.8, 0.8], device="cuda")
+
+    def process_videos(self, input_dir, filenames, video_idxs):
+        videos_read = []
+        frames_read = []
+        frames = []
+        results = []
+        for video_idx in video_idxs:
+            # Read the full-size frames from this video.
+            filename = filenames[video_idx]
+            video_path = os.path.join(input_dir, filename)
+            result = self.video_read_fn(video_path)
+            # Error? Then skip this video.
+            if result is None: continue
+
+            videos_read.append(video_idx)
+
+            # Keep track of the original frames (need them later).
+            my_frames, my_idxs = result
+
+            frames.append(my_frames)
+            frames_read.append(my_idxs)
+            for i, frame in enumerate(my_frames):
+                h, w = frame.shape[:2]
+                img = Image.fromarray(frame.astype(np.uint8))
+                img = img.resize(size=[s // 2 for s in img.size])
+
+                batch_boxes, probs = self.detector.detect(img, landmarks=False)
+
+                faces = []
+                scores = []
+                if batch_boxes is None:
+                    continue
+                for bbox, score in zip(batch_boxes, probs):
+                    if bbox is not None:
+                        xmin, ymin, xmax, ymax = [int(b * 2) for b in bbox]
+                        w = xmax - xmin
+                        h = ymax - ymin
+                        p_h = h // 3
+                        p_w = w // 3
+                        crop = frame[max(ymin - p_h, 0):ymax + p_h, max(xmin - p_w, 0):xmax + p_w]
+                        faces.append(crop)
+                        scores.append(score)
+
+                frame_dict = {"video_idx": video_idx,
+                              "frame_idx": my_idxs[i],
+                              "frame_w": w,
+                              "frame_h": h,
+                              "faces": faces,
+                              "scores": scores}
+                results.append(frame_dict)
+
+        return results
+
+    def process_video(self, video_path):
+        """Convenience method for doing face extraction on a single video."""
+        input_dir = os.path.dirname(video_path)
+        filenames = [os.path.basename(video_path)]
+        return self.process_videos(input_dir, filenames, [0])
+
+
+
+def confident_strategy(pred, t=0.8):
+    pred = np.array(pred)
+    sz = len(pred)
+    fakes = np.count_nonzero(pred > t)
+    # 11 frames are detected as fakes with high probability
+    if fakes > sz // 2.5 and fakes > 11:
+        return np.mean(pred[pred > t])
+    elif np.count_nonzero(pred < 0.2) > 0.9 * sz:
+        return np.mean(pred[pred < 0.2])
+    else:
+        return np.mean(pred)
+
+strategy = confident_strategy
+
+
+def put_to_center(img, input_size):
+    img = img[:input_size, :input_size]
+    image = np.zeros((input_size, input_size, 3), dtype=np.uint8)
+    start_w = (input_size - img.shape[1]) // 2
+    start_h = (input_size - img.shape[0]) // 2
+    image[start_h:start_h + img.shape[0], start_w: start_w + img.shape[1], :] = img
+    return image
+
+
+def isotropically_resize_image(img, size, interpolation_down=cv2.INTER_AREA, interpolation_up=cv2.INTER_CUBIC):
+    h, w = img.shape[:2]
+    if max(w, h) == size:
+        return img
+    if w > h:
+        scale = size / w
+        h = h * scale
+        w = size
+    else:
+        scale = size / h
+        w = w * scale
+        h = size
+    interpolation = interpolation_up if scale > 1 else interpolation_down
+    resized = cv2.resize(img, (int(w), int(h)), interpolation=interpolation)
+    return resized
+
+
+def predict_on_video(face_extractor, video_path, batch_size, input_size, models, strategy=np.mean,
+                     apply_compression=False):
+    batch_size *= 4
+    try:
+        faces = face_extractor.process_video(video_path)
+        if len(faces) > 0:
+            x = np.zeros((batch_size, input_size, input_size, 3), dtype=np.uint8)
+            n = 0
+            for frame_data in faces:
+                for face in frame_data["faces"]:
+                    resized_face = isotropically_resize_image(face, input_size)
+                    resized_face = put_to_center(resized_face, input_size)
+                    if apply_compression:
+                        resized_face = image_compression(resized_face, quality=90, image_type=".jpg")
+                    if n + 1 < batch_size:
+                        x[n] = resized_face
+                        n += 1
+                    else:
+                        pass
+            if n > 0:
+                x = torch.tensor(x, device="cuda").float()
+                # Preprocess the images.
+                x = x.permute((0, 3, 1, 2))
+                for i in range(len(x)):
+                    x[i] = normalize_transform(x[i] / 255.)
+                # Make a prediction, then take the average.
+                with torch.no_grad():
+                    preds = []
+                    for model in models:
+                        y_pred = model(x[:n].half())
+                        y_pred = torch.sigmoid(y_pred.squeeze())
+                        bpred = y_pred[:n].cpu().numpy()
+                        preds.append(strategy(bpred))
+                    return np.mean(preds)
+    except Exception as e:
+        print("Prediction error on video %s: %s" % (video_path, str(e)))
+
+    return 0.5
+
+
+def predict_on_video_set(face_extractor, videos, input_size, num_workers, test_dir, frames_per_video, models,
+                         strategy=np.mean,
+                         apply_compression=False):
+    def process_file(i):
+        filename = videos[i]
+        y_pred = predict_on_video(face_extractor=face_extractor, video_path=os.path.join(test_dir, filename),
+                                  input_size=input_size,
+                                  batch_size=frames_per_video,
+                                  models=models, strategy=strategy, apply_compression=apply_compression)
+        return y_pred
+
+    with ThreadPoolExecutor(max_workers=num_workers) as ex:
+        predictions = ex.map(process_file, range(len(videos)))
+    return list(predictions)
+

backend/dfdc_model/preprocessing/__init__.py

@@ -0,0 +1 @@
+from .face_detector import *

backend/dfdc_model/preprocessing/face_detector.py

@@ -0,0 +1,72 @@
+import os
+os.environ["MKL_NUM_THREADS"] = "1"
+os.environ["NUMEXPR_NUM_THREADS"] = "1"
+os.environ["OMP_NUM_THREADS"] = "1"
+
+from abc import ABC, abstractmethod
+from collections import OrderedDict
+from typing import List
+
+import cv2
+cv2.ocl.setUseOpenCL(False)
+cv2.setNumThreads(0)
+
+from PIL import Image
+from facenet_pytorch.models.mtcnn import MTCNN
+from torch.utils.data import Dataset
+
+
+class VideoFaceDetector(ABC):
+
+    def __init__(self, **kwargs) -> None:
+        super().__init__()
+
+    @property
+    @abstractmethod
+    def _batch_size(self) -> int:
+        pass
+
+    @abstractmethod
+    def _detect_faces(self, frames) -> List:
+        pass
+
+
+class FacenetDetector(VideoFaceDetector):
+
+    def __init__(self, device="cuda:0") -> None:
+        super().__init__()
+        self.detector = MTCNN(margin=0,thresholds=[0.85, 0.95, 0.95], device=device)
+
+    def _detect_faces(self, frames) -> List:
+        batch_boxes, *_ = self.detector.detect(frames, landmarks=False)
+        return [b.tolist() if b is not None else None for b in batch_boxes]
+
+    @property
+    def _batch_size(self):
+        return 32
+
+
+class VideoDataset(Dataset):
+
+    def __init__(self, videos) -> None:
+        super().__init__()
+        self.videos = videos
+
+    def __getitem__(self, index: int):
+        video = self.videos[index]
+        capture = cv2.VideoCapture(video)
+        frames_num = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
+        frames = OrderedDict()
+        for i in range(frames_num):
+            capture.grab()
+            success, frame = capture.retrieve()
+            if not success:
+                continue
+            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+            frame = Image.fromarray(frame)
+            frame = frame.resize(size=[s // 2 for s in frame.size])
+            frames[i] = frame
+        return video, list(frames.keys()), list(frames.values())
+
+    def __len__(self) -> int:
+        return len(self.videos)

backend/dfdc_model/preprocessing/utils.py

@@ -0,0 +1,51 @@
+import json
+import os
+from glob import glob
+from pathlib import Path
+
+
+def get_original_video_paths(root_dir, basename=False):
+    originals = set()
+    originals_v = set()
+    for json_path in glob(os.path.join(root_dir, "*/metadata.json")):
+        dir = Path(json_path).parent
+        with open(json_path, "r") as f:
+            metadata = json.load(f)
+        for k, v in metadata.items():
+            original = v.get("original", None)
+            if v["label"] == "REAL":
+                original = k
+                originals_v.add(original)
+                originals.add(os.path.join(dir, original))
+    originals = list(originals)
+    originals_v = list(originals_v)
+    print(len(originals))
+    return originals_v if basename else originals
+
+
+def get_original_with_fakes(root_dir):
+    pairs = []
+    for json_path in glob(os.path.join(root_dir, "*/metadata.json")):
+        with open(json_path, "r") as f:
+            metadata = json.load(f)
+        for k, v in metadata.items():
+            original = v.get("original", None)
+            if v["label"] == "FAKE":
+                pairs.append((original[:-4], k[:-4] ))
+
+    return pairs
+
+
+def get_originals_and_fakes(root_dir):
+    originals = []
+    fakes = []
+    for json_path in glob(os.path.join(root_dir, "*/metadata.json")):
+        with open(json_path, "r") as f:
+            metadata = json.load(f)
+        for k, v in metadata.items():
+            if v["label"] == "FAKE":
+                fakes.append(k[:-4])
+            else:
+                originals.append(k[:-4])
+
+    return originals, fakes

backend/dfdc_model/training/transforms/albu.py

@@ -0,0 +1,99 @@
+import random
+
+import cv2
+import numpy as np
+from albumentations import DualTransform, ImageOnlyTransform
+from albumentations.augmentations.functional import crop
+
+
+def isotropically_resize_image(img, size, interpolation_down=cv2.INTER_AREA, interpolation_up=cv2.INTER_CUBIC):
+    h, w = img.shape[:2]
+    if max(w, h) == size:
+        return img
+    if w > h:
+        scale = size / w
+        h = h * scale
+        w = size
+    else:
+        scale = size / h
+        w = w * scale
+        h = size
+    interpolation = interpolation_up if scale > 1 else interpolation_down
+    resized = cv2.resize(img, (int(w), int(h)), interpolation=interpolation)
+    return resized
+
+
+class IsotropicResize(DualTransform):
+    def __init__(self, max_side, interpolation_down=cv2.INTER_AREA, interpolation_up=cv2.INTER_CUBIC,
+                 always_apply=False, p=1):
+        super(IsotropicResize, self).__init__(always_apply, p)
+        self.max_side = max_side
+        self.interpolation_down = interpolation_down
+        self.interpolation_up = interpolation_up
+
+    def apply(self, img, interpolation_down=cv2.INTER_AREA, interpolation_up=cv2.INTER_CUBIC, **params):
+        return isotropically_resize_image(img, size=self.max_side, interpolation_down=interpolation_down,
+                                          interpolation_up=interpolation_up)
+
+    def apply_to_mask(self, img, **params):
+        return self.apply(img, interpolation_down=cv2.INTER_NEAREST, interpolation_up=cv2.INTER_NEAREST, **params)
+
+    def get_transform_init_args_names(self):
+        return ("max_side", "interpolation_down", "interpolation_up")
+
+
+class Resize4xAndBack(ImageOnlyTransform):
+    def __init__(self, always_apply=False, p=0.5):
+        super(Resize4xAndBack, self).__init__(always_apply, p)
+
+    def apply(self, img, **params):
+        h, w = img.shape[:2]
+        scale = random.choice([2, 4])
+        img = cv2.resize(img, (w // scale, h // scale), interpolation=cv2.INTER_AREA)
+        img = cv2.resize(img, (w, h),
+                         interpolation=random.choice([cv2.INTER_CUBIC, cv2.INTER_LINEAR, cv2.INTER_NEAREST]))
+        return img
+
+
+class RandomSizedCropNonEmptyMaskIfExists(DualTransform):
+
+    def __init__(self, min_max_height, w2h_ratio=[0.7, 1.3], always_apply=False, p=0.5):
+        super(RandomSizedCropNonEmptyMaskIfExists, self).__init__(always_apply, p)
+
+        self.min_max_height = min_max_height
+        self.w2h_ratio = w2h_ratio
+
+    def apply(self, img, x_min=0, x_max=0, y_min=0, y_max=0, **params):
+        cropped = crop(img, x_min, y_min, x_max, y_max)
+        return cropped
+
+    @property
+    def targets_as_params(self):
+        return ["mask"]
+
+    def get_params_dependent_on_targets(self, params):
+        mask = params["mask"]
+        mask_height, mask_width = mask.shape[:2]
+        crop_height = int(mask_height * random.uniform(self.min_max_height[0], self.min_max_height[1]))
+        w2h_ratio = random.uniform(*self.w2h_ratio)
+        crop_width = min(int(crop_height * w2h_ratio), mask_width - 1)
+        if mask.sum() == 0:
+            x_min = random.randint(0, mask_width - crop_width + 1)
+            y_min = random.randint(0, mask_height - crop_height + 1)
+        else:
+            mask = mask.sum(axis=-1) if mask.ndim == 3 else mask
+            non_zero_yx = np.argwhere(mask)
+            y, x = random.choice(non_zero_yx)
+            x_min = x - random.randint(0, crop_width - 1)
+            y_min = y - random.randint(0, crop_height - 1)
+            x_min = np.clip(x_min, 0, mask_width - crop_width)
+            y_min = np.clip(y_min, 0, mask_height - crop_height)
+
+        x_max = x_min + crop_height
+        y_max = y_min + crop_width
+        y_max = min(mask_height, y_max)
+        x_max = min(mask_width, x_max)
+        return {"x_min": x_min, "x_max": x_max, "y_min": y_min, "y_max": y_max}
+
+    def get_transform_init_args_names(self):
+        return "min_max_height", "height", "width", "w2h_ratio"

backend/dfdc_model/training/zoo/classifiers.py

@@ -0,0 +1,172 @@
+from functools import partial
+
+import numpy as np
+import torch
+from timm.models.efficientnet import tf_efficientnet_b4_ns, tf_efficientnet_b3_ns, \
+    tf_efficientnet_b5_ns, tf_efficientnet_b2_ns, tf_efficientnet_b6_ns, tf_efficientnet_b7_ns
+from torch import nn
+from torch.nn.modules.dropout import Dropout
+from torch.nn.modules.linear import Linear
+from torch.nn.modules.pooling import AdaptiveAvgPool2d
+
+encoder_params = {
+    "tf_efficientnet_b3_ns": {
+        "features": 1536,
+        "init_op": partial(tf_efficientnet_b3_ns, pretrained=True, drop_path_rate=0.2)
+    },
+    "tf_efficientnet_b2_ns": {
+        "features": 1408,
+        "init_op": partial(tf_efficientnet_b2_ns, pretrained=False, drop_path_rate=0.2)
+    },
+    "tf_efficientnet_b4_ns": {
+        "features": 1792,
+        "init_op": partial(tf_efficientnet_b4_ns, pretrained=True, drop_path_rate=0.5)
+    },
+    "tf_efficientnet_b5_ns": {
+        "features": 2048,
+        "init_op": partial(tf_efficientnet_b5_ns, pretrained=True, drop_path_rate=0.2)
+    },
+    "tf_efficientnet_b4_ns_03d": {
+        "features": 1792,
+        "init_op": partial(tf_efficientnet_b4_ns, pretrained=True, drop_path_rate=0.3)
+    },
+    "tf_efficientnet_b5_ns_03d": {
+        "features": 2048,
+        "init_op": partial(tf_efficientnet_b5_ns, pretrained=True, drop_path_rate=0.3)
+    },
+    "tf_efficientnet_b5_ns_04d": {
+        "features": 2048,
+        "init_op": partial(tf_efficientnet_b5_ns, pretrained=True, drop_path_rate=0.4)
+    },
+    "tf_efficientnet_b6_ns": {
+        "features": 2304,
+        "init_op": partial(tf_efficientnet_b6_ns, pretrained=True, drop_path_rate=0.2)
+    },
+    "tf_efficientnet_b7_ns": {
+        "features": 2560,
+        "init_op": partial(tf_efficientnet_b7_ns, pretrained=True, drop_path_rate=0.2)
+    },
+    "tf_efficientnet_b6_ns_04d": {
+        "features": 2304,
+        "init_op": partial(tf_efficientnet_b6_ns, pretrained=True, drop_path_rate=0.4)
+    },
+}
+
+
+def setup_srm_weights(input_channels: int = 3) -> torch.Tensor:
+    """Creates the SRM kernels for noise analysis."""
+    # note: values taken from Zhou et al., "Learning Rich Features for Image Manipulation Detection", CVPR2018
+    srm_kernel = torch.from_numpy(np.array([
+        [  # srm 1/2 horiz
+            [0., 0., 0., 0., 0.],  # noqa: E241,E201
+            [0., 0., 0., 0., 0.],  # noqa: E241,E201
+            [0., 1., -2., 1., 0.],  # noqa: E241,E201
+            [0., 0., 0., 0., 0.],  # noqa: E241,E201
+            [0., 0., 0., 0., 0.],  # noqa: E241,E201
+        ], [  # srm 1/4
+            [0., 0., 0., 0., 0.],  # noqa: E241,E201
+            [0., -1., 2., -1., 0.],  # noqa: E241,E201
+            [0., 2., -4., 2., 0.],  # noqa: E241,E201
+            [0., -1., 2., -1., 0.],  # noqa: E241,E201
+            [0., 0., 0., 0., 0.],  # noqa: E241,E201
+        ], [  # srm 1/12
+            [-1., 2., -2., 2., -1.],  # noqa: E241,E201
+            [2., -6., 8., -6., 2.],  # noqa: E241,E201
+            [-2., 8., -12., 8., -2.],  # noqa: E241,E201
+            [2., -6., 8., -6., 2.],  # noqa: E241,E201
+            [-1., 2., -2., 2., -1.],  # noqa: E241,E201
+        ]
+    ])).float()
+    srm_kernel[0] /= 2
+    srm_kernel[1] /= 4
+    srm_kernel[2] /= 12
+    return srm_kernel.view(3, 1, 5, 5).repeat(1, input_channels, 1, 1)
+
+
+def setup_srm_layer(input_channels: int = 3) -> torch.nn.Module:
+    """Creates a SRM convolution layer for noise analysis."""
+    weights = setup_srm_weights(input_channels)
+    conv = torch.nn.Conv2d(input_channels, out_channels=3, kernel_size=5, stride=1, padding=2, bias=False)
+    with torch.no_grad():
+        conv.weight = torch.nn.Parameter(weights, requires_grad=False)
+    return conv
+
+
+class DeepFakeClassifierSRM(nn.Module):
+    def __init__(self, encoder, dropout_rate=0.5) -> None:
+        super().__init__()
+        self.encoder = encoder_params[encoder]["init_op"]()
+        self.avg_pool = AdaptiveAvgPool2d((1, 1))
+        self.srm_conv = setup_srm_layer(3)
+        self.dropout = Dropout(dropout_rate)
+        self.fc = Linear(encoder_params[encoder]["features"], 1)
+
+    def forward(self, x):
+        noise = self.srm_conv(x)
+        x = self.encoder.forward_features(noise)
+        x = self.avg_pool(x).flatten(1)
+        x = self.dropout(x)
+        x = self.fc(x)
+        return x
+
+
+class GlobalWeightedAvgPool2d(nn.Module):
+    """
+    Global Weighted Average Pooling from paper "Global Weighted Average
+    Pooling Bridges Pixel-level Localization and Image-level Classification"
+    """
+
+    def __init__(self, features: int, flatten=False):
+        super().__init__()
+        self.conv = nn.Conv2d(features, 1, kernel_size=1, bias=True)
+        self.flatten = flatten
+
+    def fscore(self, x):
+        m = self.conv(x)
+        m = m.sigmoid().exp()
+        return m
+
+    def norm(self, x: torch.Tensor):
+        return x / x.sum(dim=[2, 3], keepdim=True)
+
+    def forward(self, x):
+        input_x = x
+        x = self.fscore(x)
+        x = self.norm(x)
+        x = x * input_x
+        x = x.sum(dim=[2, 3], keepdim=not self.flatten)
+        return x
+
+
+class DeepFakeClassifier(nn.Module):
+    def __init__(self, encoder, dropout_rate=0.0) -> None:
+        super().__init__()
+        self.encoder = encoder_params[encoder]["init_op"]()
+        self.avg_pool = AdaptiveAvgPool2d((1, 1))
+        self.dropout = Dropout(dropout_rate)
+        self.fc = Linear(encoder_params[encoder]["features"], 1)
+
+    def forward(self, x):
+        x = self.encoder.forward_features(x)
+        x = self.avg_pool(x).flatten(1)
+        x = self.dropout(x)
+        x = self.fc(x)
+        return x
+
+
+
+
+class DeepFakeClassifierGWAP(nn.Module):
+    def __init__(self, encoder, dropout_rate=0.5) -> None:
+        super().__init__()
+        self.encoder = encoder_params[encoder]["init_op"]()
+        self.avg_pool = GlobalWeightedAvgPool2d(encoder_params[encoder]["features"])
+        self.dropout = Dropout(dropout_rate)
+        self.fc = Linear(encoder_params[encoder]["features"], 1)
+
+    def forward(self, x):
+        x = self.encoder.forward_features(x)
+        x = self.avg_pool(x).flatten(1)
+        x = self.dropout(x)
+        x = self.fc(x)
+        return x

backend/dfdc_model/training/zoo/unet.py

@@ -0,0 +1,151 @@
+from functools import partial
+
+import torch
+from timm.models.efficientnet import tf_efficientnet_b3_ns, tf_efficientnet_b5_ns
+from torch import nn
+from torch.nn import Dropout2d, Conv2d
+from torch.nn.modules.dropout import Dropout
+from torch.nn.modules.linear import Linear
+from torch.nn.modules.pooling import AdaptiveAvgPool2d
+from torch.nn.modules.upsampling import UpsamplingBilinear2d
+
+encoder_params = {
+    "tf_efficientnet_b3_ns": {
+        "features": 1536,
+        "filters": [40, 32, 48, 136, 1536],
+        "decoder_filters": [64, 128, 256, 256],
+        "init_op": partial(tf_efficientnet_b3_ns, pretrained=True, drop_path_rate=0.2)
+    },
+    "tf_efficientnet_b5_ns": {
+        "features": 2048,
+        "filters": [48, 40, 64, 176, 2048],
+        "decoder_filters": [64, 128, 256, 256],
+        "init_op": partial(tf_efficientnet_b5_ns, pretrained=True, drop_path_rate=0.2)
+    },
+}
+
+
+class DecoderBlock(nn.Module):
+    def __init__(self, in_channels, out_channels):
+        super().__init__()
+        self.layer = nn.Sequential(
+            nn.Upsample(scale_factor=2),
+            nn.Conv2d(in_channels, out_channels, 3, padding=1),
+            nn.ReLU(inplace=True)
+        )
+
+    def forward(self, x):
+        return self.layer(x)
+
+
+class ConcatBottleneck(nn.Module):
+    def __init__(self, in_channels, out_channels):
+        super().__init__()
+        self.seq = nn.Sequential(
+            nn.Conv2d(in_channels, out_channels, 3, padding=1),
+            nn.ReLU(inplace=True)
+        )
+
+    def forward(self, dec, enc):
+        x = torch.cat([dec, enc], dim=1)
+        return self.seq(x)
+
+
+class Decoder(nn.Module):
+    def __init__(self, decoder_filters, filters, upsample_filters=None,
+                 decoder_block=DecoderBlock, bottleneck=ConcatBottleneck, dropout=0):
+        super().__init__()
+        self.decoder_filters = decoder_filters
+        self.filters = filters
+        self.decoder_block = decoder_block
+        self.decoder_stages = nn.ModuleList([self._get_decoder(idx) for idx in range(0, len(decoder_filters))])
+        self.bottlenecks = nn.ModuleList([bottleneck(self.filters[-i - 2] + f, f)
+                                          for i, f in enumerate(reversed(decoder_filters))])
+        self.dropout = Dropout2d(dropout) if dropout > 0 else None
+        self.last_block = None
+        if upsample_filters:
+            self.last_block = decoder_block(decoder_filters[0], out_channels=upsample_filters)
+        else:
+            self.last_block = UpsamplingBilinear2d(scale_factor=2)
+
+    def forward(self, encoder_results: list):
+        x = encoder_results[0]
+        bottlenecks = self.bottlenecks
+        for idx, bottleneck in enumerate(bottlenecks):
+            rev_idx = - (idx + 1)
+            x = self.decoder_stages[rev_idx](x)
+            x = bottleneck(x, encoder_results[-rev_idx])
+        if self.last_block:
+            x = self.last_block(x)
+        if self.dropout:
+            x = self.dropout(x)
+        return x
+
+    def _get_decoder(self, layer):
+        idx = layer + 1
+        if idx == len(self.decoder_filters):
+            in_channels = self.filters[idx]
+        else:
+            in_channels = self.decoder_filters[idx]
+        return self.decoder_block(in_channels, self.decoder_filters[max(layer, 0)])
+
+
+def _initialize_weights(module):
+    for m in module.modules():
+        if isinstance(m, nn.Conv2d) or isinstance(m, nn.ConvTranspose2d) or isinstance(m, nn.Linear):
+            m.weight.data = nn.init.kaiming_normal_(m.weight.data)
+            if m.bias is not None:
+                m.bias.data.zero_()
+        elif isinstance(m, nn.BatchNorm2d):
+            m.weight.data.fill_(1)
+            m.bias.data.zero_()
+
+
+class EfficientUnetClassifier(nn.Module):
+    def __init__(self, encoder, dropout_rate=0.5) -> None:
+        super().__init__()
+        self.decoder = Decoder(decoder_filters=encoder_params[encoder]["decoder_filters"],
+                               filters=encoder_params[encoder]["filters"])
+        self.avg_pool = AdaptiveAvgPool2d((1, 1))
+        self.dropout = Dropout(dropout_rate)
+        self.fc = Linear(encoder_params[encoder]["features"], 1)
+        self.final = Conv2d(encoder_params[encoder]["decoder_filters"][0], out_channels=1, kernel_size=1, bias=False)
+        _initialize_weights(self)
+        self.encoder = encoder_params[encoder]["init_op"]()
+
+    def get_encoder_features(self, x):
+        encoder_results = []
+        x = self.encoder.conv_stem(x)
+        x = self.encoder.bn1(x)
+        x = self.encoder.act1(x)
+        encoder_results.append(x)
+        x = self.encoder.blocks[:2](x)
+        encoder_results.append(x)
+        x = self.encoder.blocks[2:3](x)
+        encoder_results.append(x)
+        x = self.encoder.blocks[3:5](x)
+        encoder_results.append(x)
+        x = self.encoder.blocks[5:](x)
+        x = self.encoder.conv_head(x)
+        x = self.encoder.bn2(x)
+        x = self.encoder.act2(x)
+        encoder_results.append(x)
+        encoder_results = list(reversed(encoder_results))
+        return encoder_results
+
+    def forward(self, x):
+        encoder_results = self.get_encoder_features(x)
+        seg = self.final(self.decoder(encoder_results))
+        x = encoder_results[0]
+        x = self.avg_pool(x).flatten(1)
+        x = self.dropout(x)
+        x = self.fc(x)
+        return x, seg
+
+
+if __name__ == '__main__':
+    model = EfficientUnetClassifier("tf_efficientnet_b5_ns")
+    model.eval()
+    with torch.no_grad():
+        input = torch.rand(4, 3, 224, 224)
+        print(model(input))

backend/enhanced_prediction.py

@@ -0,0 +1,226 @@
+import numpy as np
+import torch
+from dfdc_model.kernel_utils import isotropically_resize_image, put_to_center, normalize_transform
+from gradcam import GradCAM, create_gradcam_image, image_to_base64
+
+def enhanced_predict_on_video(face_extractor, video_path, batch_size, input_size, models, strategy=np.mean,
+                              apply_compression=False, return_frame_predictions=True):
+    """Enhanced video prediction that returns frame-level predictions and Grad-CAM"""
+    # Reduce batch size for lower memory usage
+    batch_size = max(8, batch_size * 2)
+    frame_predictions = []
+    fake_frames = []
+    gradcam_images = []
+    
+    try:
+        faces = face_extractor.process_video(video_path)
+        # Sample every Nth frame to reduce memory usage
+        frame_sample_rate = 3  # Only process every 3rd frame
+        if len(faces) > 0:
+            x = np.zeros((batch_size, input_size, input_size, 3), dtype=np.uint8)
+            frame_data_list = []
+            n = 0
+            for idx, frame_data in enumerate(faces):
+                if idx % frame_sample_rate != 0:
+                    continue
+                for face in frame_data["faces"]:
+                    resized_face = isotropically_resize_image(face, input_size)
+                    resized_face = put_to_center(resized_face, input_size)
+                    if apply_compression:
+                        from albumentations.augmentations.functional import image_compression
+                        resized_face = image_compression(resized_face, quality=90, image_type=".jpg")
+                    if n + 1 < batch_size:
+                        x[n] = resized_face
+                        frame_data_list.append({
+                            'frame_idx': frame_data["frame_idx"],
+                            'original_face': face,
+                            'processed_face': resized_face
+                        })
+                        n += 1
+                    else:
+                        break
+            if n > 0:
+                device = "cuda" if torch.cuda.is_available() else "cpu"
+                x = torch.tensor(x, device=device).float()
+                # Preprocess the images.
+                x = x.permute((0, 3, 1, 2))
+                for i in range(len(x)):
+                    x[i] = normalize_transform(x[i] / 255.)
+                
+                # Make predictions for each frame
+                all_preds = []
+                for model in models:
+                    if device == "cuda":
+                        y_pred = model(x[:n].half())
+                    else:
+                        y_pred = model(x[:n])
+                    y_pred = torch.sigmoid(y_pred.squeeze())
+                    frame_preds = y_pred.detach().cpu().numpy()
+                    all_preds.append(frame_preds)
+                
+                    # Average predictions across models
+                    avg_frame_preds = np.mean(all_preds, axis=0)
+                
+                    # Identify fake frames (threshold > 0.5)
+                    for i, pred in enumerate(avg_frame_preds):
+                        if i < len(frame_data_list):
+                            frame_idx = frame_data_list[i]['frame_idx']
+                            frame_predictions.append({
+                                'frame_idx': frame_idx,
+                                'prediction': float(pred)
+                            })
+                        
+                            if pred > 0.5:  # Fake frame
+                                fake_frames.append(frame_idx)
+                
+                    # Generate Grad-CAM for the most suspicious frame
+                    if len(fake_frames) > 0 and len(models) > 0:
+                        # Find the frame with highest fake probability
+                        max_pred_idx = np.argmax(avg_frame_preds)
+                        if max_pred_idx < len(frame_data_list):
+                            try:
+                                # Get the target layer (last convolutional layer)
+                                model = models[0]
+                                target_layer = None
+                                for name, module in model.named_modules():
+                                    if isinstance(module, torch.nn.Conv2d):
+                                        target_layer = module
+                            
+                                if target_layer is not None:
+                                    gradcam = GradCAM(model, target_layer)
+                                    input_tensor = x[max_pred_idx:max_pred_idx+1].clone().detach()
+                                    input_tensor.requires_grad = True
+                                    if device == "cuda":
+                                        input_tensor = input_tensor.half()
+                                    heatmap = gradcam.generate_cam(input_tensor)
+                                    # Create visualization
+                                    original_face = frame_data_list[max_pred_idx]['processed_face']
+                                    # Ensure heatmap is detached before converting to numpy
+                                    if isinstance(heatmap, torch.Tensor):
+                                        heatmap = heatmap.detach().cpu().numpy()
+                                    gradcam_image = create_gradcam_image(original_face, heatmap)
+                                    gradcam_base64 = image_to_base64(gradcam_image)
+                                    gradcam_images.append(gradcam_base64)
+                            except Exception as e:
+                                print(f"Grad-CAM generation error: {e}")
+                
+                    # Return overall prediction using strategy
+                    overall_prediction = strategy(avg_frame_preds)
+                
+                    return {
+                        'prediction': overall_prediction,
+                        'frame_predictions': frame_predictions,
+                        'fake_frames': fake_frames,
+                        'gradcam_images': gradcam_images
+                    }
+                    
+    except Exception as e:
+        print("Enhanced prediction error on video %s: %s" % (video_path, str(e)))
+
+    return {
+        'prediction': 0.5,
+        'frame_predictions': [],
+        'fake_frames': [],
+        'gradcam_images': []
+    }
+
+def enhanced_predict_on_image(image_path, models, input_size=380):
+    """Enhanced image prediction with Grad-CAM"""
+    try:
+        from dfdc_model.kernel_utils import isotropically_resize_image, put_to_center, normalize_transform
+        from facenet_pytorch.models.mtcnn import MTCNN
+        import cv2
+        
+        device = "cuda" if torch.cuda.is_available() else "cpu"
+        detector = MTCNN(margin=0, thresholds=[0.7, 0.8, 0.8], device=device)
+        
+        # Read and process image
+        image = cv2.imread(image_path)
+        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
+        
+        # Detect faces
+        from PIL import Image as PILImage
+        img_pil = PILImage.fromarray(image)
+        img_pil = img_pil.resize(size=[s // 2 for s in img_pil.size])
+        
+        batch_boxes, probs = detector.detect(img_pil, landmarks=False)
+        
+        if batch_boxes is None:
+            return {
+                'prediction': 0.5,
+                'gradcam_image': None
+            }
+            
+        # Process the best face
+        best_idx = 0
+        if len(probs) > 1:
+            best_idx = np.argmax(probs)
+            
+        bbox = batch_boxes[best_idx]
+        if bbox is not None:
+            xmin, ymin, xmax, ymax = [int(b * 2) for b in bbox]
+            w = xmax - xmin
+            h = ymax - ymin
+            p_h = h // 3
+            p_w = w // 3
+            crop = image[max(ymin - p_h, 0):ymax + p_h, max(xmin - p_w, 0):xmax + p_w]
+            
+            # Resize and center
+            resized_face = isotropically_resize_image(crop, input_size)
+            resized_face = put_to_center(resized_face, input_size)
+            
+            # Prepare tensor
+            x = torch.tensor(resized_face, device=device).float().unsqueeze(0)
+            x = x.permute((0, 3, 1, 2))
+            x = normalize_transform(x[0] / 255.).unsqueeze(0)
+            
+            # Predict
+            preds = []
+            for model in models:
+                if device == "cuda":
+                    y_pred = model(x.half())
+                else:
+                    y_pred = model(x)
+                y_pred = torch.sigmoid(y_pred.squeeze())
+                preds.append(y_pred.detach().cpu().numpy())
+            
+            prediction = np.mean(preds)
+            
+            # Generate Grad-CAM if fake
+            gradcam_image = None
+            if prediction > 0.5 and len(models) > 0:
+                try:
+                    model = models[0]
+                    target_layer = None
+                    for name, module in model.named_modules():
+                        if isinstance(module, torch.nn.Conv2d):
+                            target_layer = module
+                    
+                    if target_layer is not None:
+                        gradcam = GradCAM(model, target_layer)
+                        input_tensor = x.clone().detach()
+                        input_tensor.requires_grad = True
+                        if device == "cuda":
+                            input_tensor = input_tensor.half()
+                        heatmap = gradcam.generate_cam(input_tensor)
+                        gradcam_vis = create_gradcam_image(resized_face, heatmap)
+                        gradcam_image = image_to_base64(gradcam_vis)
+                except Exception as e:
+                    print(f"Grad-CAM generation error: {e}")
+            
+            return {
+                'prediction': prediction,
+                'gradcam_image': gradcam_image
+            }
+        
+        return {
+            'prediction': 0.5,
+            'gradcam_image': None
+        }
+        
+    except Exception as e:
+        print(f"Enhanced image prediction error: {e}")
+        return {
+            'prediction': 0.5,
+            'gradcam_image': None
+        }

backend/gradcam.py

@@ -0,0 +1,152 @@
+import torch
+import torch.nn.functional as F
+import cv2
+import numpy as np
+from PIL import Image
+import io
+import base64
+
+class GradCAM:
+    def __init__(self, model, target_layer):
+        self.model = model
+        self.target_layer = target_layer
+        self.gradients = None
+        self.activations = None
+        self.hook_layers()
+    
+    def hook_layers(self):
+        def backward_hook(module, grad_input, grad_output):
+            self.gradients = grad_output[0]
+        
+        def forward_hook(module, input, output):
+            self.activations = output
+        
+        self.target_layer.register_forward_hook(forward_hook)
+        self.target_layer.register_full_backward_hook(backward_hook)
+    
+    def generate_cam(self, input_tensor, class_idx=None):
+        # Forward pass
+        output = self.model(input_tensor)
+        if class_idx is None:
+            # For single output models, use index 0
+            if len(output.shape) == 2 and output.shape[1] == 1:
+                class_idx = 0
+            else:
+                class_idx = torch.argmax(output, dim=1)
+        # Backward pass
+        self.model.zero_grad()
+        if len(output.shape) == 2 and output.shape[1] == 1:
+            # Single output case
+            output[0, 0].backward(retain_graph=True)
+        else:
+            output[0, class_idx].backward(retain_graph=True)
+        # Generate CAM
+        gradients = self.gradients[0]
+        activations = self.activations[0]
+        # Pool the gradients across the spatial dimensions
+        pooled_gradients = torch.mean(gradients, dim=[1, 2])
+        # Weight the channels by corresponding gradients
+        for i in range(activations.shape[0]):
+            activations[i, :, :] *= pooled_gradients[i]
+        # Average the channels of the activations
+        heatmap = torch.mean(activations, dim=0)
+        # ReLU on top of the heatmap
+        heatmap = F.relu(heatmap)
+        # Normalize the heatmap more aggressively for better visibility
+        heatmap_min = torch.min(heatmap)
+        heatmap_max = torch.max(heatmap)
+        if heatmap_max > heatmap_min:
+            heatmap = (heatmap - heatmap_min) / (heatmap_max - heatmap_min)
+            # Apply power function to enhance high values (make them more visible)
+            heatmap = torch.pow(heatmap, 0.5)  # Square root for better contrast
+        else:
+            heatmap = torch.zeros_like(heatmap)
+        # Detach before converting to numpy
+        return heatmap.detach().cpu().numpy()
+
+def create_gradcam_image(original_image, heatmap):
+    """Create a Grad-CAM visualization overlaid on the original image"""
+    # Resize heatmap to match original image size
+    if isinstance(original_image, torch.Tensor):
+        original_image = original_image.cpu().numpy()
+        if original_image.shape[0] == 3:  # CHW format
+            original_image = np.transpose(original_image, (1, 2, 0))
+    
+    # Validate heatmap
+    if heatmap is None or not isinstance(heatmap, np.ndarray) or heatmap.size == 0:
+        print("Invalid heatmap for Grad-CAM visualization")
+        return original_image.astype(np.uint8)
+    
+    # Ensure heatmap is 2D
+    if len(heatmap.shape) > 2:
+        heatmap = np.squeeze(heatmap)
+    
+    # Check for NaN or infinite values
+    if np.any(np.isnan(heatmap)) or np.any(np.isinf(heatmap)):
+        print("Heatmap contains NaN or infinite values")
+        return original_image.astype(np.uint8)
+    
+    h, w = original_image.shape[:2]
+    
+    # Ensure heatmap has valid dimensions before resizing
+    if heatmap.shape[0] == 0 or heatmap.shape[1] == 0:
+        print("Heatmap has zero dimensions, skipping Grad-CAM")
+        return original_image.astype(np.uint8)
+    
+    # Convert heatmap to float32 for better OpenCV compatibility
+    heatmap_float = heatmap.astype(np.float32)
+    
+    try:
+        heatmap_resized = cv2.resize(heatmap_float, (w, h), interpolation=cv2.INTER_LINEAR)
+    except cv2.error as e:
+        print(f"OpenCV resize error: {e}")
+        print(f"Heatmap shape: {heatmap.shape}, Target size: ({w}, {h})")
+        print(f"Heatmap dtype: {heatmap.dtype}, min: {np.min(heatmap)}, max: {np.max(heatmap)}")
+        # Fallback: use numpy resize
+        from scipy import ndimage
+        try:
+            zoom_factors = (h / heatmap.shape[0], w / heatmap.shape[1])
+            heatmap_resized = ndimage.zoom(heatmap, zoom_factors, order=1)
+        except:
+            print("Fallback resize also failed, returning original image")
+            return original_image.astype(np.uint8)
+    
+    # Normalize heatmap to 0-255 with better contrast
+    if np.max(heatmap_resized) > 0:
+        heatmap_resized = heatmap_resized / np.max(heatmap_resized)  # Normalize to 0-1
+        # Apply gamma correction for better visibility
+        heatmap_resized = np.power(heatmap_resized, 0.7)  # Enhance contrast
+    
+    heatmap_resized = np.uint8(255 * heatmap_resized)
+    
+    # Apply colormap (JET gives red for high values)
+    heatmap_colored = cv2.applyColorMap(heatmap_resized, cv2.COLORMAP_JET)
+    
+    # Convert original image to uint8 if needed
+    if original_image.max() <= 1.0:
+        original_image = (original_image * 255).astype(np.uint8)
+    
+    # Ensure both images are in RGB format
+    if len(original_image.shape) == 3:
+        original_image = np.uint8(original_image)
+    heatmap_colored = cv2.cvtColor(heatmap_colored, cv2.COLOR_BGR2RGB)
+    
+    # Create a mask for significant heatmap values (threshold to only show strong activations)
+    mask = heatmap_resized > 50  # Only overlay where heatmap is significant
+    
+    # Create the overlay
+    superimposed = original_image.copy()
+    superimposed[mask] = (0.3 * original_image[mask] + 0.7 * heatmap_colored[mask]).astype(np.uint8)
+    superimposed = np.uint8(superimposed)
+    
+    return superimposed
+
+def image_to_base64(image):
+    """Convert numpy array image to base64 string"""
+    if isinstance(image, np.ndarray):
+        pil_image = Image.fromarray(image)
+        buffer = io.BytesIO()
+        pil_image.save(buffer, format='PNG')
+        img_str = base64.b64encode(buffer.getvalue()).decode()
+        return img_str
+    return None