app.py

import gradio as gr
import cv2
import numpy as np
import tensorflow as tf
import cv2
import json
import numpy as np
from concurrent.futures import ThreadPoolExecutor
import dlib
import tempfile
import shutil
import os
from tensorflow.keras.applications.inception_v3 import preprocess_input


model_path = 'deepfake_detection_model.h5'
model = tf.keras.models.load_model(model_path)

IMG_SIZE = (299, 299) 
MOTION_THRESHOLD = 20  
FRAME_SKIP = 2       
no_of_frames = 10
MAX_FRAMES=no_of_frames

detector = dlib.get_frontal_face_detector()

def extract_faces_from_frame(frame, detector):
    """
    Detects faces in a frame and returns the resized faces.

    Parameters:
    - frame: The video frame to process.
    - detector: Dlib face detector.

    Returns:
    - resized_faces (list): List of resized faces detected in the frame.
    """
    gray_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    faces = detector(gray_frame)
    resized_faces = []

    for face in faces:
        x1, y1, x2, y2 = face.left(), face.top(), face.right(), face.bottom()
        crop_img = frame[y1:y2, x1:x2]
        if crop_img.size != 0:  
            resized_face = cv2.resize(crop_img, IMG_SIZE)
            resized_faces.append(resized_face)

    # Debug: Log the number of faces detected
    #print(f"Detected {len(resized_faces)} faces in current frame")
    return resized_faces

def process_frame(video_path, detector, frame_skip):
    """
    Processes frames to extract motion and face data concurrently.

    Parameters:
    - cap: OpenCV VideoCapture object.
    - detector: Dlib face detector.
    - frame_skip (int): Number of frames to skip for processing.

    Returns:
    - motion_frames (list): List of motion-based face images.
    - all_faces (list): List of all detected faces for fallback.
    """
    prev_frame = None
    frame_count = 0
    motion_frames = []
    all_faces = []
    cap = cv2.VideoCapture(video_path)
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break

        # Skip frames to improve processing speed
        if frame_count % frame_skip != 0:
            frame_count += 1
            continue

        # Debug: Log frame number being processed
        #print(f"Processing frame {frame_count}")

        # # Resize frame to reduce processing time (optional, adjust size as needed)
        # frame = cv2.resize(frame, (640, 360))

        # Extract faces from the current frame
        faces = extract_faces_from_frame(frame, detector)
        all_faces.extend(faces)  # Store all faces detected, including non-motion

        gray_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

        if prev_frame is None:
            prev_frame = gray_frame
            frame_count += 1
            continue

        # Calculate frame difference to detect motion
        frame_diff = cv2.absdiff(prev_frame, gray_frame)
        motion_score = np.sum(frame_diff)

        # Debug: Log the motion score
        #print(f"Motion score: {motion_score}")

        # Check if motion is above the defined threshold and add the face to motion frames
        if motion_score > MOTION_THRESHOLD and faces:
            motion_frames.extend(faces)

        prev_frame = gray_frame
        frame_count += 1

    cap.release()
    return motion_frames, all_faces

def select_well_distributed_frames(motion_frames, all_faces, no_of_frames):
    """
    Selects well-distributed frames from the detected motion and fallback faces.

    Parameters:
    - motion_frames (list): List of frames with detected motion.
    - all_faces (list): List of all detected faces.
    - no_of_frames (int): Required number of frames.

    Returns:
    - final_frames (list): List of selected frames.
    """
    # Case 1: Motion frames exceed the required number
    if len(motion_frames) >= no_of_frames:
        interval = len(motion_frames) // no_of_frames
        distributed_motion_frames = [motion_frames[i * interval] for i in range(no_of_frames)]
        return distributed_motion_frames

    # Case 2: Motion frames are less than the required number
    needed_frames = no_of_frames - len(motion_frames)

    # If all frames together are still less than needed, return all frames available
    if len(motion_frames) + len(all_faces) < no_of_frames:
        #print(f"Returning all available frames: {len(motion_frames) + len(all_faces)}")
        return motion_frames + all_faces

    interval = max(1, len(all_faces) // needed_frames)
    additional_faces = [all_faces[i * interval] for i in range(needed_frames)]

    combined_frames = motion_frames + additional_faces
    interval = max(1, len(combined_frames) // no_of_frames)
    final_frames = [combined_frames[i * interval] for i in range(no_of_frames)]
    return final_frames

def extract_frames(no_of_frames, video_path):
  motion_frames, all_faces = process_frame(video_path, detector, FRAME_SKIP)
  final_frames = select_well_distributed_frames(motion_frames, all_faces, no_of_frames)
  return final_frames


def predict_video(model, video_path):
    """
    Predict if a video is REAL or FAKE using the trained model.

    Parameters:
    - model: The loaded deepfake detection model.
    - video_path: Path to the video file to be processed.

    Returns:
    - str: 'REAL' or 'FAKE' based on the model's prediction.
    """
    # Extract frames from the video
    frames = extract_frames(no_of_frames, video_path)
    original_frames = frames

    # Convert the frames list to a 5D tensor (1, time_steps, height, width, channels)
    if len(frames) < MAX_FRAMES:
        # Pad with zero arrays to match MAX_FRAMES
        while len(frames) < MAX_FRAMES:
            frames.append(np.zeros((299, 299, 3), dtype=np.float32))

    frames = frames[:MAX_FRAMES] 
    frames = np.array(frames)    
    frames = preprocess_input(frames) 

    # Expand dims to fit the model input shape
    input_data = np.expand_dims(frames, axis=0)  # Shape becomes (1, MAX_FRAMES, 299, 299, 3)

    # Predict using the model
    prediction = model.predict(input_data)
    probability = prediction[0][0]  # Get the probability for the first (and only) sample
    # Convert probability to class label
    if probability >=0.6:
        predicted_label='FAKE'
    else:
        predicted_label = 'REAL'
        probability=1-probability
    return original_frames, predicted_label, probability

def display_frames_and_prediction(video_file):
    # Ensure file is completely uploaded and of correct type
    if video_file is None:
        return [], "<div style='color: red;'>No video uploaded!</div>", ""
    
    # Check file size (10 MB limit)
    if os.path.getsize(video_file) > 10 * 1024 * 1024:  # 10 MB
        return [], "<div style='color: red;'>File size exceeds 10 MB limit!</div>", ""
    
    # Check file extension
    if not video_file.endswith('.mp4'):
        return [], "<div style='color: red;'>Only .mp4 files are allowed!</div>", ""
    
    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as temp_file:
        temp_file_path = temp_file.name

    with open(video_file, 'rb') as src_file:
        with open(temp_file_path, 'wb') as dest_file:
            shutil.copyfileobj(src_file, dest_file)

    frames, predicted_label, confidence = predict_video(model, temp_file_path)
    os.remove(temp_file_path)

    confidence_text = f"Confidence: {confidence:.2%}"

    prediction_style = (
    f"<div style='color: {'green' if predicted_label == 'REAL' else 'red'}; "
    "text-align: center; font-size: 24px; font-weight: bold; "
    "border: 2px solid; padding: 10px; border-radius: 5px;'>"
    f"{predicted_label}</div>"
)


    return frames, prediction_style, confidence_text

iface = gr.Interface(
    fn=display_frames_and_prediction,
    inputs=gr.File(label="Upload Video", interactive=True),
    outputs=[
        gr.Gallery(label="Extracted Frames"), 
        gr.HTML(label="Prediction"), 
        gr.Textbox(label="Confidence", interactive=False)
    ],
    title="Deepfake Detection",
    description="Upload a video to determine if it is REAL or FAKE based on the deepfake detection model.",
    css="app.css",
    examples=[
        ["examples/abarnvbtwb.mp4"], 
        ["examples/aapnvogymq.mp4"],
    ],
    cache_examples="lazy"
)

iface.launch()