app.py

import streamlit as st
import os
from moviepy import VideoFileClip
import torch
from torchvision import transforms
from torch.utils.data.dataset import Dataset
import os
import numpy as np
import cv2
import matplotlib.pyplot as plt
import librosa
from sklearn.preprocessing import StandardScaler
import torch.nn as nn
import torch.nn.functional as F
import torch
import torch.nn as nn
import torch.optim as optim
import matplotlib.pyplot as plt
from torchvision.models import resnet50
import librosa.display
N_MFCC = 30
SEQUENCE_LENGTH = 30  # Number of MFCC frames for each audio segment


def display_video_frames(video_path):
    cap = cv2.VideoCapture(video_path)
    frames = []

    # Read the first 30 frames from the video
    for _ in range(30):
        ret, frame = cap.read()
        if not ret:
            break
        frames.append(frame)

    cap.release()

    # Create a 5x6 grid to display frames
    grid = []
    for i in range(0, 30, 6):
        row = np.hstack(frames[i:i+6])
        grid.append(row)

    # Concatenate the rows to create the final grid image
    grid_image = np.vstack(grid)

    st.image(grid_image, channels="BGR")

class VideoDataset(Dataset):
    def __init__(self, video_paths, audio_paths, labels, transform=None):
        self.video_paths = video_paths
        self.audio_paths = audio_paths
        self.labels = labels
        self.transform = transform
        self.count=30
    def __len__(self):
        return len(self.video_paths)
    def __getitem__(self, idx):
        video_path = self.video_paths[idx]
        audio_path = self.audio_paths[idx]
        label = self.labels[idx]
        d={'A':0,'B':1,'C':1,'D':1}
        frames=[]
        for i,frame in enumerate(self.frame_extract(video_path)):
          frames.append(self.transform(frame))
          if(len(frames) == self.count):
            break
        frames=torch.stack(frames)
        audio = extract_features(audio_path)
        label = d[label]
        return frames, audio, label
    def frame_extract(self,path):
      vidObj = cv2.VideoCapture(path)
      success = 1
      while success:
          success, image = vidObj.read()
          if success:
              yield image
def plot_mfcc(y, sr):
    mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
    plt.figure(figsize=(12, 8))
    librosa.display.specshow(mfccs, sr=sr, x_axis='time')
    plt.colorbar()
    plt.title('MFCC')
    st.pyplot(plt.gcf())
def extract_features(file_path, n_mfcc=N_MFCC):
    audio, sample_rate = librosa.load(file_path, sr=None)
    mfccs = librosa.feature.mfcc(y=audio, sr=sample_rate, n_mfcc=n_mfcc)

    # Normalize the features
    scaler = StandardScaler()
    mfccs_scaled = scaler.fit_transform(mfccs.T)

    # Pad or truncate to match the required sequence length
    if mfccs_scaled.shape[0] < SEQUENCE_LENGTH:
        pad_width = SEQUENCE_LENGTH - mfccs_scaled.shape[0]
        mfccs_scaled = np.pad(mfccs_scaled, ((0, pad_width), (0, 0)), mode='constant')
    else:
        mfccs_scaled = mfccs_scaled[:SEQUENCE_LENGTH, :]

    return mfccs_scaled
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Resize((224,224))
])

# Define the model
class MultimodalModel(nn.Module):
    def __init__(self, num_classes=4):
        super(MultimodalModel, self).__init__()

        # ResNet50 for video feature extraction
        resnet = resnet50(pretrained=True)
        self.resnet = nn.Sequential(*list(resnet.children())[:-2])  # Remove last FC and pooling layers
        self.resnet_pool = nn.AdaptiveAvgPool2d((1, 1))  # Adaptive pooling for dynamic output size
        self.resnet_fc = nn.Linear(2048, 512)  # Fully connected layer for ResNet output

        # GRU and LSTM for audio feature extraction
        self.gru = nn.GRU(input_size=30, hidden_size=128, batch_first=True, bidirectional=True)
        self.lstm = nn.LSTM(input_size=256, hidden_size=128, batch_first=True, bidirectional=True)

        # Fully connected layers for combined features
        self.fc1 = nn.Linear(512 + 256, 128)
        self.fc2 = nn.Linear(128, num_classes)

    def forward(self, video_frames, audio_features):
        # Video feature extraction
        batch_size, num_frames, _, _, _ = video_frames.size()
        video_frames = video_frames.view(batch_size * num_frames, 3, 224, 224)
        video_features = self.resnet(video_frames)
        video_features = self.resnet_pool(video_features)
        video_features = torch.flatten(video_features, start_dim=1)
        video_features = self.resnet_fc(video_features)
        video_features = video_features.view(batch_size, num_frames, -1)
        video_features = video_features.mean(dim=1)  # Average pooling over time

        # Audio feature extraction
        audio_features, _ = self.gru(audio_features)
        audio_features, _ = self.lstm(audio_features)
        audio_features = audio_features[:, -1, :]  # Take the last time step

        # Combine features
        combined_features = torch.cat((video_features, audio_features), dim=1)
        x = torch.relu(self.fc1(combined_features))
        x = self.fc2(x)

        return x
def extract_audio(input_video, output_audio):
    video_clip = VideoFileClip(input_video)
    audio_clip = video_clip.audio
    audio_clip.write_audiofile(output_audio)
    video_clip.close()
    audio_clip.close()
model = MultimodalModel(num_classes=2)
model.load_state_dict(torch.load('models\epoch_9.pth', map_location='cpu')) # Added map_location='cpu'
# Move the model to the appropriate device (GPU if available)
device = "cuda" if torch.cuda.is_available() else "cpu"
model.to(device)  # Move the model to the same device as the data
model.eval() #set the model to evaluation mode
st.title("Deepfake Detection using Multimodal Deep Learning")
uploaded_file = st.file_uploader("Choose a file to upload", type=["mp4",'.wmv'])
if uploaded_file is not None:
    # Get the filename
    filename = uploaded_file.name

    # Specify the folder path where you want to save the file
    folder_path = "uploads/"  # Create this folder if it doesn't exist

    # Create the full file path
    file_path = os.path.join(folder_path, filename)

    # Save the file to the specified folder
    with open(file_path, "wb") as f:
        f.write(uploaded_file.getbuffer())

    # Display a success message
    st.success(f"File '{filename}' uploaded successfully to '{folder_path}'")
    input_video=folder_path+filename
    output_audio=input_video.replace("mp4","wav")
    st.write("### :file_folder: Frames in video")
    display_video_frames(input_video)
    try:
        extract_audio(input_video, output_audio)
        st.success(f"Audio extracted successfully: {output_audio}")
    except:
        st.error("Failed to extract audio.")
    y, sr = librosa.load(output_audio)
    st.write('### :loud_sound: Mel-Frequency Cepstral Coefficients (MFCC)')
    plot_mfcc(y, sr)
    video_paths=[]
    video_paths.append(input_video)
    audio_paths=[]
    audio_paths.append(output_audio)
    test_labels=["B"]
    test_dataset = VideoDataset(video_paths, audio_paths, test_labels,transform=transform)
        
    # Get the input data for the 5th sample (index 0)
    frames, audio, label = test_dataset[0]
    # Move the data to the appropriate device (GPU if available)
    device = "cuda" if torch.cuda.is_available() else "cpu"
    frames = frames.unsqueeze(0).to(device)  # Add batch dimension
    audio = torch.tensor(audio).unsqueeze(0).float().to(device) # Add batch dimension and convert to float tensor

    # Make the prediction
    with torch.no_grad():
        outputs = model(frames, audio)
        _, predicted = torch.max(outputs, 1)

    # Print the predicted label
    l={0:'Real',1:'Fake'}
    st.write(f"### :point_right: The Predicted Label for the Given Video: :blue[{l[predicted.item()]}]")
    

    os.remove(input_video)
    os.remove(output_audio)