# app.py â DFDA Streamlit app with Premium UI/UX Design
import streamlit as st
import torch
import torch.nn as nn
import torch.nn.functional as F
import torchaudio
import torchaudio.transforms as T
import torchvision.transforms as transforms
import numpy as np
import cv2
from pathlib import Path
from PIL import Image
import plotly.graph_objects as go
import plotly.express as px
import io
import timm
import warnings
import tempfile
import shutil
import os
import importlib.util
import traceback
warnings.filterwarnings('ignore')
# Robust import for download_models
_download_mod = None
try:
import download_models as _dm
download_from_hf = _dm.download_from_hf
MODELS_DIR = _dm.MODELS_DIR
except Exception:
_base = Path(__file__).parent
candidates = [
_base / "download_models.py",
_base / "src" / "download_models.py"
]
loaded = False
tb_text = None
for cand in candidates:
if cand.exists():
try:
spec = importlib.util.spec_from_file_location("download_models", str(cand))
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
download_from_hf = mod.download_from_hf
MODELS_DIR = mod.MODELS_DIR
loaded = True
break
except Exception:
tb_text = traceback.format_exc()
if not loaded:
st.error("Could not import download_models.py. Make sure it exists in the repo root or src/ folder.")
if tb_text:
st.text(tb_text)
raise ImportError("download_models import failed")
# HF repo & filenames
AUDIO_REPO = "jayanthbottu/DFDA"
AUDIO_FILENAME = "Audio.pth"
VIDEO_REPO = "jayanthbottu/DFDA"
VIDEO_FILENAME = "Video.pt"
IMAGE_REPO = "jayanthbottu/DFDA"
IMAGE_FILENAME = "Image.pt"
MODELS_DIR.mkdir(parents=True, exist_ok=True)
# ======================= PAGE CONFIG ============================
st.set_page_config(
page_title="DFDA - Deepfake Detection & Attribution",
page_icon="đ",
layout="wide",
initial_sidebar_state="expanded"
)
# ======================= PREMIUM CSS STYLING ============================
st.markdown("""
""", unsafe_allow_html=True)
# ======================= CONFIG CLASSES (unchanged) ============================
class AudioConfig:
MODEL_PATH = MODELS_DIR / "Audio.pth"
CLASS_NAMES = ["ElevenLabs", "Original", "Tacotron", "Text To Speech", "Voice Conversion"]
SAMPLE_RATE = 16000
N_MELS = 128
N_FFT = 2048
HOP_LENGTH = 512
MAX_AUDIO_LENGTH = 5
DEVICE = torch.device("cpu")
class ImageConfig:
MODEL_PATH = MODELS_DIR / "Image.pt"
CLASS_NAMES = ["DALL-E", "DeepFaceLab", "Face2Face", "FaceShifter", "FaceSwap", "Midjourney",
"NeuralTextures", "Real", "Stable Diffusion", "StyleGAN"]
IMG_SIZE = 299
DEVICE = torch.device("cpu")
class VideoConfig:
MODEL_PATH = MODELS_DIR / "Video.pt"
CLASS_NAMES = ["Deepfakes", "Face2Face", "FaceShifter", "FaceSwap", "NeuralTextures", "Original"]
DEVICE = torch.device("cpu")
FRAME_SKIP = 10
# ======================= HELPER FUNCTIONS (unchanged) ============================
def safe_torch_load(path, map_location="cpu"):
p = str(path)
try:
if hasattr(torch.serialization, "safe_globals"):
with torch.serialization.safe_globals([np.core.multiarray.scalar]):
ckpt = torch.load(p, map_location=map_location)
else:
if hasattr(torch.serialization, "add_safe_globals"):
torch.serialization.add_safe_globals([np.core.multiarray.scalar])
ckpt = torch.load(p, map_location=map_location)
return ckpt
except Exception as e:
try:
ckpt = torch.load(p, map_location=map_location, weights_only=False)
return ckpt
except Exception as e2:
raise RuntimeError(f"torch.load failed: {e} ; fallback failed: {e2}")
def download_and_load_torch_model(repo_id: str, filename: str, expected_device: torch.device):
try:
local_path = download_from_hf(repo_id, filename)
except Exception as e:
raise RuntimeError(f"Download failed for {repo_id}/{filename}: {e}")
if not local_path.exists():
raise FileNotFoundError(f"After download, file not found: {local_path}")
return local_path
# ======================= MODEL CLASSES (unchanged) ============================
class AudioDeepfakeCNN(nn.Module):
def __init__(self, num_classes=5):
super().__init__()
self.conv1 = nn.Sequential(
nn.Conv2d(1, 64, 3, padding=1),
nn.BatchNorm2d(64), nn.ReLU(inplace=True),
nn.Conv2d(64, 64, 3, padding=1),
nn.BatchNorm2d(64), nn.ReLU(inplace=True),
nn.MaxPool2d(2,2),
nn.Dropout2d(0.25)
)
self.conv2 = nn.Sequential(
nn.Conv2d(64, 128, 3, padding=1),
nn.BatchNorm2d(128), nn.ReLU(inplace=True),
nn.Conv2d(128, 128, 3, padding=1),
nn.BatchNorm2d(128), nn.ReLU(inplace=True),
nn.MaxPool2d(2,2),
nn.Dropout2d(0.25)
)
self.conv3 = nn.Sequential(
nn.Conv2d(128, 256, 3, padding=1),
nn.BatchNorm2d(256), nn.ReLU(inplace=True),
nn.Conv2d(256, 256, 3, padding=1),
nn.BatchNorm2d(256), nn.ReLU(inplace=True),
nn.MaxPool2d(2,2),
nn.Dropout2d(0.3)
)
self.conv4 = nn.Sequential(
nn.Conv2d(256, 512, 3, padding=1),
nn.BatchNorm2d(512), nn.ReLU(inplace=True),
nn.Conv2d(512, 512, 3, padding=1),
nn.BatchNorm2d(512), nn.ReLU(inplace=True),
nn.MaxPool2d(2,2),
nn.Dropout2d(0.3)
)
self.gap = nn.AdaptiveAvgPool2d((1,1))
self.fc = nn.Sequential(
nn.Linear(512, 256), nn.ReLU(inplace=True), nn.Dropout(0.5),
nn.Linear(256, 128), nn.ReLU(inplace=True), nn.Dropout(0.4),
nn.Linear(128, num_classes)
)
def forward(self, x):
x = self.conv1(x); x = self.conv2(x); x = self.conv3(x); x = self.conv4(x)
x = self.gap(x); x = x.view(x.size(0), -1); x = self.fc(x); return x
class AttentionBlock(nn.Module):
def __init__(self, in_features):
super(AttentionBlock, self).__init__()
self.attention = nn.Sequential(
nn.Linear(in_features, in_features // 8),
nn.ReLU(),
nn.Linear(in_features // 8, in_features),
nn.Sigmoid()
)
def forward(self, x):
att_weights = self.attention(x)
return x * att_weights
class EnhancedXceptionNet(nn.Module):
def __init__(self, num_classes, dropout_rate=0.3, use_attention=True):
super(EnhancedXceptionNet, self).__init__()
try:
self.base_model = timm.create_model('xception', pretrained=False)
in_features = self.base_model.get_classifier().in_features
self.base_model.reset_classifier(0)
except Exception:
class DummyNet(nn.Module):
def __init__(self):
super().__init__()
self.net = nn.Sequential(
nn.Conv2d(3, 32, 3, padding=1),
nn.ReLU(),
nn.AdaptiveAvgPool2d((1,1)),
nn.Flatten()
)
def forward(self, x):
return self.net(x)
self.base_model = DummyNet()
in_features = 32
self.use_attention = use_attention
if use_attention:
self.attention = AttentionBlock(in_features)
self.classifier = nn.Sequential(
nn.Dropout(dropout_rate),
nn.Linear(in_features, 512),
nn.BatchNorm1d(512),
nn.ReLU(),
nn.Dropout(dropout_rate / 2),
nn.Linear(512, num_classes)
)
def forward(self, x):
features = self.base_model(x)
if self.use_attention:
features = self.attention(features)
output = self.classifier(features)
return output
# ======================= MODEL LOADERS (unchanged) ============================
@st.cache_resource
def load_audio_model():
try:
model = AudioDeepfakeCNN(num_classes=len(AudioConfig.CLASS_NAMES))
checkpoint = safe_torch_load(AudioConfig.MODEL_PATH, map_location=str(AudioConfig.DEVICE))
state = checkpoint['model_state_dict'] if isinstance(checkpoint, dict) and 'model_state_dict' in checkpoint else checkpoint
if isinstance(state, dict):
model_dict = model.state_dict()
filtered_dict = {k: v for k, v in state.items() if k in model_dict and v.shape == model_dict[k].shape}
model_dict.update(filtered_dict)
model.load_state_dict(model_dict, strict=False)
else:
model.load_state_dict(state)
model.to(AudioConfig.DEVICE).eval()
return model
except Exception as e:
st.error(f"Error loading audio model: {e}")
return None
@st.cache_resource
def load_image_model():
try:
model = EnhancedXceptionNet(num_classes=len(ImageConfig.CLASS_NAMES), dropout_rate=0.3, use_attention=True)
checkpoint = safe_torch_load(ImageConfig.MODEL_PATH, map_location=str(ImageConfig.DEVICE))
state_dict = checkpoint['model_state_dict'] if isinstance(checkpoint, dict) and 'model_state_dict' in checkpoint else checkpoint
if isinstance(state_dict, dict):
model_dict = model.state_dict()
filtered_dict = {k: v for k, v in state_dict.items() if k in model_dict and v.shape == model_dict[k].shape}
model_dict.update(filtered_dict)
model.load_state_dict(model_dict, strict=False)
else:
model.load_state_dict(state_dict, strict=False)
model.to(ImageConfig.DEVICE).eval()
return model
except Exception as e:
st.error(f"Error loading image model: {e}")
return None
@st.cache_resource
def load_video_model():
try:
num_classes = len(VideoConfig.CLASS_NAMES)
try:
model = timm.create_model("legacy_xception", pretrained=False, num_classes=num_classes)
except Exception:
class SmallNet(nn.Module):
def __init__(self, n):
super().__init__()
self.net = nn.Sequential(
nn.Conv2d(3, 16, 3, padding=1),
nn.ReLU(),
nn.AdaptiveAvgPool2d((1,1)),
nn.Flatten(),
nn.Linear(16, n)
)
def forward(self, x):
return self.net(x)
model = SmallNet(num_classes)
checkpoint = safe_torch_load(VideoConfig.MODEL_PATH, map_location=str(VideoConfig.DEVICE))
state_dict = checkpoint['model_state_dict'] if isinstance(checkpoint, dict) and 'model_state_dict' in checkpoint else checkpoint
if isinstance(state_dict, dict):
model_dict = model.state_dict()
filtered_dict = {k: v for k, v in state_dict.items() if k in model_dict and v.shape == model_dict[k].shape}
model_dict.update(filtered_dict)
model.load_state_dict(model_dict, strict=False)
else:
model.load_state_dict(state_dict, strict=False)
model.to(VideoConfig.DEVICE).eval()
return model
except Exception as e:
st.error(f"Error loading video model: {e}")
return None
# ======================= PREPROCESSING (unchanged) ============================
def preprocess_audio(audio_path, sample_rate, n_mels, n_fft, hop_length, max_length):
waveform, sr = torchaudio.load(str(audio_path))
if waveform.shape[0] > 1:
waveform = torch.mean(waveform, dim=0, keepdim=True)
if sr != sample_rate:
resampler = T.Resample(sr, sample_rate)
waveform = resampler(waveform)
max_len = sample_rate * max_length
if waveform.shape[1] < max_len:
waveform = torch.nn.functional.pad(waveform, (0, max_len - waveform.shape[1]))
else:
waveform = waveform[:, :max_len]
mel_spec = T.MelSpectrogram(sample_rate, n_fft=n_fft, hop_length=hop_length, n_mels=n_mels, power=2.0)(waveform)
mel_spec_db = T.AmplitudeToDB(top_db=80)(mel_spec)
mel_spec_db = (mel_spec_db - mel_spec_db.mean()) / (mel_spec_db.std() + 1e-8)
return mel_spec_db, mel_spec_db.squeeze().numpy()
def preprocess_image(image_path, size):
trans = transforms.Compose([
transforms.Resize((size, size)),
transforms.ToTensor(),
transforms.Normalize([0.485,0.456,0.406],[0.229,0.224,0.225])
])
img = Image.open(image_path).convert('RGB')
return trans(img).unsqueeze(0)
def extract_frames(video_path, frame_skip=10):
frames = []
cap = cv2.VideoCapture(str(video_path))
count = 0
while cap.isOpened():
ret, frame = cap.read()
if not ret: break
if count % frame_skip == 0:
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frames.append(frame)
count += 1
cap.release()
return frames
def preprocess_frame(frame, size):
trans = transforms.Compose([
transforms.ToPILImage(),
transforms.Resize((size, size)),
transforms.ToTensor(),
transforms.Normalize([0.485,0.456,0.406],[0.229,0.224,0.225])
])
return trans(frame).unsqueeze(0)
# ======================= INFERENCE (unchanged) ============================
def predict_audio(model, audio_path, config):
device = config.DEVICE
try:
x, spec_viz = preprocess_audio(audio_path, config.SAMPLE_RATE, config.N_MELS, config.N_FFT, config.HOP_LENGTH, config.MAX_AUDIO_LENGTH)
x = x.unsqueeze(0).to(device)
with torch.no_grad():
out = F.softmax(model(x), dim=1)[0]
pred_idx = torch.argmax(out).item()
return {
"predicted_class": config.CLASS_NAMES[pred_idx],
"predicted_index": pred_idx,
"confidence": out[pred_idx].item(),
"probabilities": {c: out[i].item() for i, c in enumerate(config.CLASS_NAMES)}
}, spec_viz
except Exception as e:
return {"error": str(e)}, None
def predict_image(model, image_path, config):
device = config.DEVICE
try:
x = preprocess_image(image_path, config.IMG_SIZE).to(device)
with torch.no_grad():
out = F.softmax(model(x), dim=1)[0]
pred_idx = torch.argmax(out).item()
return {
"predicted_class": config.CLASS_NAMES[pred_idx],
"predicted_index": pred_idx,
"confidence": out[pred_idx].item(),
"probabilities": {c: out[i].item() for i, c in enumerate(config.CLASS_NAMES)}
}
except Exception as e:
return {"error": str(e)}
def predict_video(model, video_path, config):
device = config.DEVICE
try:
frames = extract_frames(video_path, config.FRAME_SKIP)
if len(frames) == 0:
return {"error":"No frames extracted"}
probs = []
progress_bar = st.progress(0)
for i, frame in enumerate(frames):
x = preprocess_frame(frame, 300).to(device)
with torch.no_grad():
p = F.softmax(model(x),dim=1)[0].cpu().numpy()
probs.append(p)
progress_bar.progress((i + 1) / len(frames))
probs = np.stack(probs)
avg_prob = probs.mean(axis=0)
pred_idx = np.argmax(avg_prob)
return {
"predicted_class": config.CLASS_NAMES[pred_idx],
"predicted_index": pred_idx,
"confidence": float(avg_prob[pred_idx]),
"probabilities": {c: float(avg_prob[i]) for i, c in enumerate(config.CLASS_NAMES)},
"frame_count": len(frames)
}
except Exception as e:
return {"error": str(e)}
# ======================= VISUALIZATION (enhanced) ============================
def plot_probabilities(probs, class_names):
sorted_items = sorted(probs.items(), key=lambda x: x[1], reverse=True)
classes = [item[0] for item in sorted_items]
values = [item[1] for item in sorted_items]
colors = ['rgba(102, 126, 234, 0.8)' if v == max(values) else 'rgba(118, 75, 162, 0.5)' for v in values]
fig = go.Figure(data=[
go.Bar(
x=values,
y=classes,
orientation='h',
marker=dict(
color=colors,
line=dict(color='rgba(102, 126, 234, 1)', width=2)
),
text=[f'{v*100:.1f}%' for v in values],
textposition='outside',
textfont=dict(size=13, color='#ffffff', family='Space Grotesk', weight=600),
hovertemplate='%{y}
Confidence: %{x:.1%}
Advanced multi-modal forensic analysis powered by deep learning
⨠State-of-the-Art Neural NetworksOur system employs deep convolutional neural networks trained on extensive datasets of authentic and manipulated media. Each modality uses specialized architectures optimized for detecting artifacts specific to different deepfake generation techniques.
Analyze audio samples for voice synthesis and manipulation
Identify AI-generated and manipulated images
Temporal analysis for video manipulation detection
DFDA System