Update app.py
Browse files
app.py
CHANGED
|
@@ -1,221 +1,138 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
import altair as alt
|
| 2 |
import numpy as np
|
| 3 |
import pandas as pd
|
| 4 |
-
import streamlit as st
|
| 5 |
-
|
| 6 |
import streamlit as st
|
| 7 |
import cv2
|
| 8 |
import torch
|
| 9 |
-
import numpy as np
|
| 10 |
-
import os
|
| 11 |
-
import tempfile
|
| 12 |
-
import time
|
| 13 |
from transformers import AutoImageProcessor, AutoModelForImageClassification
|
| 14 |
from collections import deque
|
| 15 |
import tensorflow as tf
|
| 16 |
-
from tensorflow.keras.preprocessing import image
|
| 17 |
from tensorflow.keras.models import load_model
|
|
|
|
|
|
|
| 18 |
import urllib.request
|
| 19 |
import shutil
|
| 20 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 21 |
class CNNDeepfakeDetector:
|
| 22 |
def __init__(self):
|
| 23 |
-
|
| 24 |
-
|
| 25 |
-
# Initialize CNN model for deepfake detection
|
| 26 |
-
with st.spinner("Loading CNN deepfake detection model..."):
|
| 27 |
-
try:
|
| 28 |
-
self.model = load_model('cnn_model.h5')
|
| 29 |
-
st.success("CNN model loaded successfully!")
|
| 30 |
-
except Exception as e:
|
| 31 |
-
st.error(f"Error loading CNN model: {e}")
|
| 32 |
-
st.warning("Please make sure 'cnn_model.h5' is in the current directory.")
|
| 33 |
-
self.model = None
|
| 34 |
-
|
| 35 |
-
def classify_image(self, face_img):
|
| 36 |
-
"""Classify a face image as real or fake using CNN model"""
|
| 37 |
-
try:
|
| 38 |
-
if self.model is None:
|
| 39 |
-
return "Model Not Loaded", 0.0
|
| 40 |
-
|
| 41 |
-
# Resize to target size
|
| 42 |
-
img_resized = cv2.resize(face_img, (128, 128))
|
| 43 |
-
|
| 44 |
-
# Preprocess the image
|
| 45 |
-
img_array = img_resized / 255.0
|
| 46 |
-
img_array = np.expand_dims(img_array, axis=0)
|
| 47 |
-
|
| 48 |
-
# Make prediction
|
| 49 |
-
prediction = self.model.predict(img_array)
|
| 50 |
-
confidence = float(prediction[0][0])
|
| 51 |
-
|
| 52 |
-
# In this model, <0.5 means Real, >=0.5 means Fake
|
| 53 |
-
label = 'Real' if confidence < 0.5 else 'Fake'
|
| 54 |
-
|
| 55 |
-
# Adjust confidence to be relative to the prediction
|
| 56 |
-
if label == 'Fake':
|
| 57 |
-
confidence = confidence # Already between 0.5-1.0
|
| 58 |
-
else:
|
| 59 |
-
confidence = 1.0 - confidence # Convert 0.0-0.5 to 0.5-1.0
|
| 60 |
-
|
| 61 |
-
return label, confidence
|
| 62 |
-
|
| 63 |
-
except Exception as e:
|
| 64 |
-
st.error(f"Error in CNN classification: {e}")
|
| 65 |
-
return "Error", 0.0
|
| 66 |
|
| 67 |
class DeepfakeDetector:
|
| 68 |
def __init__(self):
|
| 69 |
st.info("Initializing Deepfake Detector... This may take a moment.")
|
| 70 |
-
|
| 71 |
-
#
|
| 72 |
with st.spinner("Loading deepfake detection model..."):
|
| 73 |
-
self.image_processor =
|
| 74 |
-
|
| 75 |
-
|
| 76 |
-
'Adieee5/deepfake-detection-f3net-cross')
|
| 77 |
-
|
| 78 |
-
# Face detection model setup
|
| 79 |
with st.spinner("Loading face detection model..."):
|
| 80 |
-
|
| 81 |
-
|
| 82 |
-
|
| 83 |
-
|
| 84 |
-
|
| 85 |
-
|
| 86 |
-
|
| 87 |
-
self.use_dnn = True
|
| 88 |
-
st.success("Using DNN face detector (better for close-up faces)")
|
| 89 |
-
except Exception as e:
|
| 90 |
-
st.warning(f"Could not load DNN model: {e}")
|
| 91 |
-
self.use_dnn = False
|
| 92 |
-
|
| 93 |
-
if not self.use_dnn:
|
| 94 |
-
# Fallback to Haar cascade
|
| 95 |
-
cascade_path = cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
|
| 96 |
-
if os.path.exists(cascade_path):
|
| 97 |
-
self.face_cascade = cv2.CascadeClassifier(cascade_path)
|
| 98 |
st.warning("Using Haar cascade face detector as fallback")
|
| 99 |
else:
|
| 100 |
-
st.error(f"Cascade file not found
|
| 101 |
|
| 102 |
-
# Initialize CNN
|
| 103 |
self.cnn_detector = CNNDeepfakeDetector()
|
| 104 |
|
| 105 |
# Face tracking/smoothing parameters
|
| 106 |
-
self.face_history = {}
|
| 107 |
-
self.face_history_max_size = 10
|
| 108 |
-
self.face_ttl = 5
|
| 109 |
-
self.next_face_id = 0
|
| 110 |
-
|
| 111 |
-
# Result smoothing
|
| 112 |
-
self.result_buffer_size = 5 # Number of classifications to average
|
| 113 |
-
|
| 114 |
-
# Performance metrics
|
| 115 |
self.processing_times = deque(maxlen=30)
|
| 116 |
|
| 117 |
st.success("Models loaded successfully!")
|
| 118 |
|
| 119 |
def detect_faces_haar(self, frame):
|
| 120 |
-
"""Detect faces using Haar cascade"""
|
| 121 |
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
| 122 |
-
faces = self.face_cascade.detectMultiScale(
|
| 123 |
-
gray, scaleFactor=1.1, minNeighbors=5, minSize=(30, 30))
|
| 124 |
-
|
| 125 |
-
# Convert to list of (x,y,w,h,confidence) format for consistency
|
| 126 |
return [(x, y, w, h, 0.8) for (x, y, w, h) in faces]
|
| 127 |
|
| 128 |
-
def classify_frame(self, face_img, model_type="vit"):
|
| 129 |
-
"""Classify a face image as real or fake"""
|
| 130 |
-
try:
|
| 131 |
-
if model_type == "cnn":
|
| 132 |
-
return self.cnn_detector.classify_image(face_img)
|
| 133 |
-
|
| 134 |
-
# Default to ViT model
|
| 135 |
-
# Resize image if too small
|
| 136 |
-
h, w = face_img.shape[:2]
|
| 137 |
-
if h < 224 or w < 224:
|
| 138 |
-
scale = max(224/h, 224/w)
|
| 139 |
-
face_img = cv2.resize(face_img, (int(w*scale), int(h*scale)))
|
| 140 |
-
|
| 141 |
-
# Make sure we have valid image data
|
| 142 |
-
if face_img.size == 0:
|
| 143 |
-
return "Unknown", 0.0
|
| 144 |
-
|
| 145 |
-
# Process with ViT model
|
| 146 |
-
inputs = self.image_processor(images=face_img, return_tensors="pt")
|
| 147 |
-
outputs = self.model(**inputs)
|
| 148 |
-
logits = outputs.logits
|
| 149 |
-
|
| 150 |
-
# Get prediction and confidence
|
| 151 |
-
probs = torch.nn.functional.softmax(logits, dim=1)
|
| 152 |
-
pred = torch.argmax(logits, dim=1).item()
|
| 153 |
-
|
| 154 |
-
# The model has two classes: 0=Fake, 1=Real
|
| 155 |
-
label = 'Real' if pred == 1 else 'Fake'
|
| 156 |
-
confidence = probs[0][pred].item()
|
| 157 |
-
|
| 158 |
-
return label, confidence
|
| 159 |
-
|
| 160 |
-
except Exception as e:
|
| 161 |
-
st.error(f"Error in classification: {e}")
|
| 162 |
-
return "Error", 0.0
|
| 163 |
-
|
| 164 |
def detect_faces_dnn(self, frame):
|
| 165 |
-
"""Detect faces using DNN method"""
|
| 166 |
height, width = frame.shape[:2]
|
| 167 |
-
blob = cv2.dnn.blobFromImage(cv2.resize(frame, (300, 300)), 1.0,
|
| 168 |
-
(300, 300), (104.0, 177.0, 123.0))
|
| 169 |
self.face_net.setInput(blob)
|
| 170 |
detections = self.face_net.forward()
|
| 171 |
-
|
| 172 |
faces = []
|
| 173 |
for i in range(detections.shape[2]):
|
| 174 |
confidence = detections[0, 0, i, 2]
|
| 175 |
-
if confidence > 0.5:
|
| 176 |
box = detections[0, 0, i, 3:7] * np.array([width, height, width, height])
|
| 177 |
(x1, y1, x2, y2) = box.astype("int")
|
| 178 |
-
# Ensure box is within frame boundaries
|
| 179 |
x1, y1 = max(0, x1), max(0, y1)
|
| 180 |
x2, y2 = min(width, x2), min(height, y2)
|
| 181 |
w, h = x2 - x1, y2 - y1
|
| 182 |
-
if w > 0 and h > 0:
|
| 183 |
faces.append((x1, y1, w, h, confidence))
|
| 184 |
-
|
| 185 |
return faces
|
| 186 |
|
| 187 |
def calculate_iou(self, box1, box2):
|
| 188 |
-
"""Calculate Intersection over Union for two boxes"""
|
| 189 |
-
# Convert boxes from (x, y, w, h) to (x1, y1, x2, y2)
|
| 190 |
box1_x1, box1_y1, box1_w, box1_h = box1
|
| 191 |
box2_x1, box2_y1, box2_w, box2_h = box2
|
| 192 |
-
|
| 193 |
box1_x2, box1_y2 = box1_x1 + box1_w, box1_y1 + box1_h
|
| 194 |
box2_x2, box2_y2 = box2_x1 + box2_w, box2_y1 + box2_h
|
| 195 |
-
|
| 196 |
-
# Calculate area of intersection rectangle
|
| 197 |
x_left = max(box1_x1, box2_x1)
|
| 198 |
y_top = max(box1_y1, box2_y1)
|
| 199 |
x_right = min(box1_x2, box2_x2)
|
| 200 |
y_bottom = min(box1_y2, box2_y2)
|
| 201 |
-
|
| 202 |
if x_right < x_left or y_bottom < y_top:
|
| 203 |
return 0.0
|
| 204 |
-
|
| 205 |
intersection_area = (x_right - x_left) * (y_bottom - y_top)
|
| 206 |
-
|
| 207 |
-
# Calculate area of both boxes
|
| 208 |
box1_area = box1_w * box1_h
|
| 209 |
box2_area = box2_w * box2_h
|
| 210 |
-
|
| 211 |
-
# Calculate IoU
|
| 212 |
-
iou = intersection_area / float(box1_area + box2_area - intersection_area)
|
| 213 |
-
return iou
|
| 214 |
|
| 215 |
def track_faces(self, faces):
|
| 216 |
matched_faces = []
|
| 217 |
unmatched_detections = list(range(len(faces)))
|
| 218 |
-
|
| 219 |
if not self.face_history:
|
| 220 |
for face in faces:
|
| 221 |
face_id = self.next_face_id
|
|
@@ -264,50 +181,28 @@ class DeepfakeDetector:
|
|
| 264 |
'result_history': deque(maxlen=self.result_buffer_size)
|
| 265 |
}
|
| 266 |
matched_faces.append((face_id, faces[i]))
|
| 267 |
-
|
| 268 |
return matched_faces
|
| 269 |
|
| 270 |
def smooth_face_position(self, face_id):
|
| 271 |
-
"""Calculate smoothed position for a tracked face"""
|
| 272 |
positions = self.face_history[face_id]['positions']
|
| 273 |
-
|
| 274 |
if len(positions) == 1:
|
| 275 |
return positions[0]
|
| 276 |
-
|
| 277 |
-
# Weight recent positions more heavily
|
| 278 |
total_weight = 0
|
| 279 |
x, y, w, h = 0, 0, 0, 0
|
| 280 |
-
|
| 281 |
for i, pos in enumerate(positions):
|
| 282 |
-
|
| 283 |
-
weight = 2 ** i # Positions are stored newest to oldest
|
| 284 |
total_weight += weight
|
| 285 |
-
|
| 286 |
x += pos[0] * weight
|
| 287 |
y += pos[1] * weight
|
| 288 |
w += pos[2] * weight
|
| 289 |
h += pos[3] * weight
|
| 290 |
-
|
| 291 |
-
# Calculate weighted average
|
| 292 |
-
x = int(x / total_weight)
|
| 293 |
-
y = int(y / total_weight)
|
| 294 |
-
w = int(w / total_weight)
|
| 295 |
-
h = int(h / total_weight)
|
| 296 |
-
|
| 297 |
-
return (x, y, w, h)
|
| 298 |
|
| 299 |
def update_face_classification(self, face_id, label, confidence):
|
| 300 |
-
"""Update classification history for a face"""
|
| 301 |
self.face_history[face_id]['result_history'].append((label, confidence))
|
| 302 |
-
|
| 303 |
-
# Calculate the smoothed result
|
| 304 |
-
if not self.face_history[face_id]['result_history']:
|
| 305 |
-
return label, confidence
|
| 306 |
-
|
| 307 |
real_votes = 0
|
| 308 |
fake_votes = 0
|
| 309 |
total_confidence = 0.0
|
| 310 |
-
|
| 311 |
for result_label, result_conf in self.face_history[face_id]['result_history']:
|
| 312 |
if result_label == "Real":
|
| 313 |
real_votes += 1
|
|
@@ -315,358 +210,215 @@ class DeepfakeDetector:
|
|
| 315 |
elif result_label == "Fake":
|
| 316 |
fake_votes += 1
|
| 317 |
total_confidence += result_conf
|
| 318 |
-
|
| 319 |
-
# Determine majority vote
|
| 320 |
if real_votes >= fake_votes:
|
| 321 |
smoothed_label = "Real"
|
| 322 |
label_confidence = real_votes / len(self.face_history[face_id]['result_history'])
|
| 323 |
else:
|
| 324 |
smoothed_label = "Fake"
|
| 325 |
label_confidence = fake_votes / len(self.face_history[face_id]['result_history'])
|
| 326 |
-
|
| 327 |
-
# Average confidence weighted by vote consistency
|
| 328 |
avg_confidence = (total_confidence / len(self.face_history[face_id]['result_history'])) * label_confidence
|
| 329 |
-
|
| 330 |
-
# Store the smoothed result
|
| 331 |
self.face_history[face_id]['label'] = smoothed_label
|
| 332 |
self.face_history[face_id]['confidence'] = avg_confidence
|
| 333 |
-
|
| 334 |
return smoothed_label, avg_confidence
|
| 335 |
|
| 336 |
def process_video(self, video_path, stframe, status_text, progress_bar, detector_type="dnn", model_type="vit"):
|
| 337 |
-
"""Process video with Streamlit output"""
|
| 338 |
use_dnn_current = detector_type == "dnn" and self.use_dnn
|
| 339 |
-
|
| 340 |
cap = cv2.VideoCapture(video_path)
|
| 341 |
if not cap.isOpened():
|
| 342 |
st.error(f"Error: Cannot open video source")
|
| 343 |
return
|
| 344 |
-
|
| 345 |
-
# Get video properties
|
| 346 |
frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
| 347 |
frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
| 348 |
fps = cap.get(cv2.CAP_PROP_FPS)
|
| 349 |
total_frames = 250 if video_path != 0 else 0
|
| 350 |
-
|
| 351 |
-
# Display video info
|
| 352 |
-
if video_path != 0: # If not webcam
|
| 353 |
status_text.text(f"Video Info: {frame_width}x{frame_height}, {fps:.1f} FPS, {total_frames} frames")
|
| 354 |
else:
|
| 355 |
status_text.text(f"Webcam: {frame_width}x{frame_height}")
|
| 356 |
-
|
| 357 |
-
# Reset tracking data for new video
|
| 358 |
self.face_history = {}
|
| 359 |
self.next_face_id = 0
|
| 360 |
self.processing_times = deque(maxlen=30)
|
| 361 |
-
|
| 362 |
frame_count = 0
|
| 363 |
-
process_every_n_frames = 2
|
| 364 |
-
|
| 365 |
-
# For face detection stats
|
| 366 |
face_stats = {"Real": 0, "Fake": 0, "Unknown": 0}
|
| 367 |
|
| 368 |
-
# Main processing loop
|
| 369 |
while True:
|
| 370 |
start_time = time.time()
|
| 371 |
-
|
| 372 |
ret, frame = cap.read()
|
| 373 |
if not ret:
|
| 374 |
status_text.text("End of video reached")
|
| 375 |
break
|
| 376 |
-
|
| 377 |
frame_count += 1
|
| 378 |
-
|
| 379 |
if frame_count == 250:
|
| 380 |
st.success("Video Processed Successfully!")
|
| 381 |
break
|
| 382 |
-
|
| 383 |
-
if video_path != 0: # If not webcam, update progress
|
| 384 |
progress = min(float(frame_count) / float(max(total_frames, 1)), 1.0)
|
| 385 |
progress_bar.progress(progress)
|
| 386 |
-
|
| 387 |
process_frame = (frame_count % process_every_n_frames == 0)
|
| 388 |
-
|
| 389 |
-
# Store original frame for face extraction
|
| 390 |
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
| 391 |
|
| 392 |
if process_frame:
|
| 393 |
-
|
| 394 |
-
if use_dnn_current:
|
| 395 |
-
faces = self.detect_faces_dnn(frame)
|
| 396 |
-
else:
|
| 397 |
-
faces = self.detect_faces_haar(frame)
|
| 398 |
-
|
| 399 |
-
# Track faces across frames
|
| 400 |
tracked_faces = self.track_faces(faces)
|
| 401 |
-
|
| 402 |
-
|
| 403 |
for face_id, (x, y, w, h, face_confidence) in tracked_faces:
|
| 404 |
-
if face_id
|
| 405 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 406 |
|
|
|
|
|
|
|
| 407 |
sx, sy, sw, sh = self.smooth_face_position(face_id)
|
| 408 |
-
# Draw rectangle around face with smoothed coordinates
|
| 409 |
cv2.rectangle(frame, (sx, sy), (sx+sw, sy+sh), (0, 255, 255), 2)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 410 |
|
| 411 |
-
# Only process classification for real detections (not predicted)
|
| 412 |
-
if w > 20 and h > 20 and face_id in self.face_history:
|
| 413 |
-
try:
|
| 414 |
-
# Extract face using smoothed coordinates for better consistency
|
| 415 |
-
face = frame_rgb[sy:sy+sh, sx:sx+sw]
|
| 416 |
-
|
| 417 |
-
# Skip processing if face is too small after smoothing
|
| 418 |
-
if face.size == 0 or face.shape[0] < 20 or face.shape[1] < 20:
|
| 419 |
-
continue
|
| 420 |
-
|
| 421 |
-
# Process only every N frames or if this is a new face
|
| 422 |
-
if frame_count % process_every_n_frames == 0 or \
|
| 423 |
-
len(self.face_history[face_id]['result_history']) == 0:
|
| 424 |
-
# Classify the face using the selected model
|
| 425 |
-
label, confidence = self.classify_frame(face, model_type)
|
| 426 |
-
|
| 427 |
-
# Update and smooth results
|
| 428 |
-
label, confidence = self.update_face_classification(face_id, label, confidence)
|
| 429 |
-
else:
|
| 430 |
-
# Use last stored result
|
| 431 |
-
label = self.face_history[face_id]['label'] or "Unknown"
|
| 432 |
-
confidence = self.face_history[face_id]['confidence']
|
| 433 |
-
|
| 434 |
-
# Update stats
|
| 435 |
-
if label in face_stats:
|
| 436 |
-
face_stats[label] += 1
|
| 437 |
-
|
| 438 |
-
# Display results
|
| 439 |
-
result_text = f"{label}: {confidence:.2f}"
|
| 440 |
-
text_color = (0, 255, 0) if label == "Real" else (0, 0, 255)
|
| 441 |
-
|
| 442 |
-
# Add text background for better visibility
|
| 443 |
-
cv2.rectangle(frame, (sx, sy+sh), (sx+len(result_text)*11, sy+sh+25), (0, 0, 0), -1)
|
| 444 |
-
cv2.putText(frame, result_text, (sx, sy+sh+20),
|
| 445 |
-
cv2.FONT_HERSHEY_SIMPLEX, 0.7, text_color, 2)
|
| 446 |
-
|
| 447 |
-
# Draw face ID
|
| 448 |
-
cv2.putText(frame, f"ID:{face_id}", (sx, sy-5),
|
| 449 |
-
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 1)
|
| 450 |
-
except Exception as e:
|
| 451 |
-
st.error(f"Error processing face: {e}")
|
| 452 |
-
|
| 453 |
-
# Measure processing time
|
| 454 |
process_time = time.time() - start_time
|
| 455 |
self.processing_times.append(process_time)
|
| 456 |
avg_time = sum(self.processing_times) / len(self.processing_times)
|
| 457 |
effective_fps = 1.0 / avg_time if avg_time > 0 else 0
|
| 458 |
|
| 459 |
-
|
| 460 |
-
if video_path != 0: # If not webcam
|
| 461 |
progress_percent = (frame_count / total_frames) * 100 if total_frames > 0 else 0
|
| 462 |
cv2.putText(frame, f"Frame: {frame_count}/{total_frames} ({progress_percent:.1f}%)",
|
| 463 |
(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
|
| 464 |
else:
|
| 465 |
cv2.putText(frame, f"Frame: {frame_count}",
|
| 466 |
(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
|
| 467 |
-
|
| 468 |
-
# Show detector info and performance
|
| 469 |
detector_name = "DNN" if use_dnn_current else "Haar Cascade"
|
| 470 |
model_name = "ViT" if model_type == "vit" else "CNN"
|
| 471 |
cv2.putText(frame, f"Detector: {detector_name} | Model: {model_name} | FPS: {effective_fps:.1f}",
|
| 472 |
(10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
|
| 473 |
-
|
| 474 |
-
# Show tracking info
|
| 475 |
cv2.putText(frame, f"Tracked faces: {len(self.face_history)}",
|
| 476 |
(10, 90), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
|
| 477 |
-
|
| 478 |
-
# Display the frame in Streamlit
|
| 479 |
stframe.image(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB), channels="RGB")
|
| 480 |
-
|
| 481 |
-
# Update stats
|
| 482 |
status_text.text(f"Real: {face_stats['Real']} | Fake: {face_stats['Fake']} | FPS: {effective_fps:.1f}")
|
| 483 |
-
|
| 484 |
-
# Check if stop button is pressed
|
| 485 |
if st.session_state.get('stop_button', False):
|
| 486 |
break
|
| 487 |
-
|
| 488 |
-
# Clean up
|
| 489 |
cap.release()
|
| 490 |
return face_stats
|
| 491 |
|
| 492 |
-
# Function to ensure sample video exists
|
| 493 |
def ensure_sample_video():
|
| 494 |
sample_dir = "sample_videos"
|
| 495 |
sample_path = os.path.join(sample_dir, "Sample.mp4")
|
| 496 |
-
|
| 497 |
-
# Create directory if it doesn't exist
|
| 498 |
if not os.path.exists(sample_dir):
|
| 499 |
os.makedirs(sample_dir)
|
| 500 |
-
|
| 501 |
-
# If sample video doesn't exist, download it
|
| 502 |
if not os.path.exists(sample_path):
|
| 503 |
try:
|
| 504 |
with st.spinner("Downloading sample video..."):
|
| 505 |
-
# URL to a public domain sample video that contains faces
|
| 506 |
sample_url = "https://storage.googleapis.com/deepfake-demo/sample_deepfake.mp4"
|
| 507 |
-
|
| 508 |
-
# Download the file
|
| 509 |
with urllib.request.urlopen(sample_url) as response, open(sample_path, 'wb') as out_file:
|
| 510 |
shutil.copyfileobj(response, out_file)
|
| 511 |
-
|
| 512 |
st.success("Sample video downloaded successfully!")
|
| 513 |
except Exception as e:
|
| 514 |
st.error(f"Failed to download sample video: {e}")
|
| 515 |
return None
|
| 516 |
-
|
| 517 |
return sample_path
|
| 518 |
|
| 519 |
def main():
|
| 520 |
st.set_page_config(page_title="Deepfake Detector", layout="wide")
|
| 521 |
-
|
| 522 |
-
# App title and description
|
| 523 |
st.title("Deepfake Detection App")
|
| 524 |
st.markdown("""
|
| 525 |
This app uses computer vision and deep learning to detect deepfake videos.
|
| 526 |
Upload a video or use your webcam to detect if faces are real or manipulated.
|
| 527 |
""")
|
| 528 |
|
| 529 |
-
# Initialize session state for the detector and variables
|
| 530 |
if 'detector' not in st.session_state:
|
| 531 |
st.session_state.detector = None
|
| 532 |
-
|
| 533 |
if 'stop_button' not in st.session_state:
|
| 534 |
st.session_state.stop_button = False
|
| 535 |
-
|
| 536 |
if 'use_sample' not in st.session_state:
|
| 537 |
st.session_state.use_sample = False
|
| 538 |
-
|
| 539 |
if 'sample_path' not in st.session_state:
|
| 540 |
st.session_state.sample_path = None
|
| 541 |
|
| 542 |
-
# Initialize the detector
|
| 543 |
if st.session_state.detector is None:
|
| 544 |
st.session_state.detector = DeepfakeDetector()
|
| 545 |
|
| 546 |
-
# Create sidebar for options
|
| 547 |
st.sidebar.title("Options")
|
| 548 |
-
|
| 549 |
-
|
| 550 |
-
|
| 551 |
-
["Upload Video", "Use Webcam", "Try Sample Video"]
|
| 552 |
-
)
|
| 553 |
-
|
| 554 |
-
detector_type = st.sidebar.selectbox(
|
| 555 |
-
"Face Detector",
|
| 556 |
-
["DNN (better for close-ups)", "Haar Cascade (faster)"],
|
| 557 |
-
index=0 if st.session_state.detector.use_dnn else 1
|
| 558 |
-
)
|
| 559 |
detector_option = "dnn" if "DNN" in detector_type else "haar"
|
| 560 |
-
|
| 561 |
-
# Model selection option
|
| 562 |
-
model_type = st.sidebar.selectbox(
|
| 563 |
-
"Deepfake Detection Model",
|
| 564 |
-
["Vision Transformer (ViT)", "F3 Net Model"],
|
| 565 |
-
index=0
|
| 566 |
-
)
|
| 567 |
model_option = "vit" if "Vision" in model_type else "cnn"
|
| 568 |
|
| 569 |
-
# Main content area
|
| 570 |
col1, col2 = st.columns([3, 1])
|
| 571 |
-
|
| 572 |
with col1:
|
| 573 |
-
# Video display area
|
| 574 |
video_placeholder = st.empty()
|
| 575 |
-
|
| 576 |
with col2:
|
| 577 |
-
# Status and controls
|
| 578 |
status_text = st.empty()
|
| 579 |
progress_bar = st.empty()
|
| 580 |
-
|
| 581 |
-
# Results section
|
| 582 |
st.subheader("Results")
|
| 583 |
results_area = st.empty()
|
| 584 |
-
|
| 585 |
-
# Stop button
|
| 586 |
if st.button("Stop Processing"):
|
| 587 |
st.session_state.stop_button = True
|
| 588 |
|
| 589 |
-
# Process based on selected option
|
| 590 |
if input_option == "Upload Video":
|
| 591 |
uploaded_file = st.sidebar.file_uploader("Choose a video file", type=["mp4", "avi", "mov", "mkv"])
|
| 592 |
-
|
| 593 |
if uploaded_file is not None:
|
| 594 |
st.session_state.stop_button = False
|
| 595 |
-
|
| 596 |
-
# Save uploaded file to temp file
|
| 597 |
tfile = tempfile.NamedTemporaryFile(delete=False)
|
| 598 |
tfile.write(uploaded_file.read())
|
| 599 |
video_path = tfile.name
|
| 600 |
-
|
| 601 |
-
|
| 602 |
-
|
| 603 |
-
video_path,
|
| 604 |
-
video_placeholder,
|
| 605 |
-
status_text,
|
| 606 |
-
progress_bar,
|
| 607 |
-
detector_option,
|
| 608 |
-
model_option
|
| 609 |
-
)
|
| 610 |
-
|
| 611 |
-
# Display results
|
| 612 |
-
results_df = {
|
| 613 |
-
"Category": ["Real Faces", "Fake Faces"],
|
| 614 |
-
"Count": [face_stats["Real"], face_stats["Fake"]]
|
| 615 |
-
}
|
| 616 |
results_area.dataframe(results_df)
|
| 617 |
-
|
| 618 |
-
# Clean up temp file
|
| 619 |
os.unlink(video_path)
|
| 620 |
-
|
| 621 |
elif input_option == "Use Webcam":
|
| 622 |
-
# Reset stop button
|
| 623 |
st.session_state.stop_button = False
|
| 624 |
-
|
| 625 |
if st.sidebar.button("Start Webcam"):
|
| 626 |
-
|
| 627 |
-
|
| 628 |
-
|
| 629 |
-
video_placeholder,
|
| 630 |
-
status_text,
|
| 631 |
-
progress_bar,
|
| 632 |
-
detector_option,
|
| 633 |
-
model_option
|
| 634 |
-
)
|
| 635 |
-
|
| 636 |
-
# Display results after stopping
|
| 637 |
-
results_df = {
|
| 638 |
-
"Category": ["Real Faces", "Fake Faces"],
|
| 639 |
-
"Count": [face_stats["Real"], face_stats["Fake"]]
|
| 640 |
-
}
|
| 641 |
results_area.dataframe(results_df)
|
| 642 |
-
|
| 643 |
elif input_option == "Try Sample Video":
|
| 644 |
-
# Reset stop button
|
| 645 |
st.session_state.stop_button = False
|
| 646 |
-
|
| 647 |
-
# Get or download the sample video
|
| 648 |
sample_path = ensure_sample_video()
|
| 649 |
-
|
| 650 |
-
|
| 651 |
-
|
| 652 |
-
|
| 653 |
-
|
| 654 |
-
sample_path,
|
| 655 |
-
video_placeholder,
|
| 656 |
-
status_text,
|
| 657 |
-
progress_bar,
|
| 658 |
-
detector_option,
|
| 659 |
-
model_option
|
| 660 |
-
)
|
| 661 |
-
|
| 662 |
-
# Display results
|
| 663 |
-
results_df = {
|
| 664 |
-
"Category": ["Real Faces", "Fake Faces"],
|
| 665 |
-
"Count": [face_stats["Real"], face_stats["Fake"]]
|
| 666 |
-
}
|
| 667 |
-
results_area.dataframe(results_df)
|
| 668 |
-
else:
|
| 669 |
-
st.sidebar.error("Failed to load sample video. Please try uploading your own video instead.")
|
| 670 |
|
| 671 |
if __name__ == "__main__":
|
| 672 |
-
main()
|
|
|
|
| 1 |
+
import os
|
| 2 |
+
os.environ['TF_ENABLE_ONEDNN_OPTS'] = '0' # Disable oneDNN to avoid numerical differences warning
|
| 3 |
+
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3' # Suppress TensorFlow logs except critical errors
|
| 4 |
+
|
| 5 |
+
import logging
|
| 6 |
+
logging.getLogger('tensorflow').setLevel(logging.ERROR) # Further suppress TensorFlow warnings
|
| 7 |
+
|
| 8 |
import altair as alt
|
| 9 |
import numpy as np
|
| 10 |
import pandas as pd
|
|
|
|
|
|
|
| 11 |
import streamlit as st
|
| 12 |
import cv2
|
| 13 |
import torch
|
|
|
|
|
|
|
|
|
|
|
|
|
| 14 |
from transformers import AutoImageProcessor, AutoModelForImageClassification
|
| 15 |
from collections import deque
|
| 16 |
import tensorflow as tf
|
|
|
|
| 17 |
from tensorflow.keras.models import load_model
|
| 18 |
+
import tempfile
|
| 19 |
+
import time
|
| 20 |
import urllib.request
|
| 21 |
import shutil
|
| 22 |
|
| 23 |
+
# Cached model loading functions
|
| 24 |
+
@st.cache_resource
|
| 25 |
+
def load_cnn_model():
|
| 26 |
+
try:
|
| 27 |
+
model = load_model('cnn_model.h5')
|
| 28 |
+
st.success("CNN model loaded successfully!")
|
| 29 |
+
return model
|
| 30 |
+
except Exception as e:
|
| 31 |
+
st.error(f"Error loading CNN model: {e}")
|
| 32 |
+
st.warning("Please make sure 'cnn_model.h5' is in the current directory.")
|
| 33 |
+
return None
|
| 34 |
+
|
| 35 |
+
@st.cache_resource
|
| 36 |
+
def load_vit_components():
|
| 37 |
+
image_processor = AutoImageProcessor.from_pretrained('Adieee5/deepfake-detection-f3net-cross', use_fast=True)
|
| 38 |
+
model = AutoModelForImageClassification.from_pretrained('Adieee5/deepfake-detection-f3net-cross')
|
| 39 |
+
return image_processor, model
|
| 40 |
+
|
| 41 |
+
@st.cache_resource
|
| 42 |
+
def load_face_net():
|
| 43 |
+
model_file = "deploy.prototxt"
|
| 44 |
+
weights_file = "res10_300x300_ssd_iter_140000.caffemodel"
|
| 45 |
+
if os.path.exists(model_file) and os.path.exists(weights_file):
|
| 46 |
+
return cv2.dnn.readNetFromCaffe(model_file, weights_file)
|
| 47 |
+
return None
|
| 48 |
+
|
| 49 |
+
@st.cache_resource
|
| 50 |
+
def load_haar_cascade():
|
| 51 |
+
cascade_path = cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
|
| 52 |
+
if os.path.exists(cascade_path):
|
| 53 |
+
return cv2.CascadeClassifier(cascade_path)
|
| 54 |
+
return None
|
| 55 |
+
|
| 56 |
class CNNDeepfakeDetector:
|
| 57 |
def __init__(self):
|
| 58 |
+
self.model = load_cnn_model()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 59 |
|
| 60 |
class DeepfakeDetector:
|
| 61 |
def __init__(self):
|
| 62 |
st.info("Initializing Deepfake Detector... This may take a moment.")
|
| 63 |
+
|
| 64 |
+
# Load ViT components
|
| 65 |
with st.spinner("Loading deepfake detection model..."):
|
| 66 |
+
self.image_processor, self.model = load_vit_components()
|
| 67 |
+
|
| 68 |
+
# Load face detection models
|
|
|
|
|
|
|
|
|
|
| 69 |
with st.spinner("Loading face detection model..."):
|
| 70 |
+
self.face_net = load_face_net()
|
| 71 |
+
self.use_dnn = self.face_net is not None
|
| 72 |
+
if self.use_dnn:
|
| 73 |
+
st.success("Using DNN face detector (better for close-up faces)")
|
| 74 |
+
else:
|
| 75 |
+
self.face_cascade = load_haar_cascade()
|
| 76 |
+
if self.face_cascade:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 77 |
st.warning("Using Haar cascade face detector as fallback")
|
| 78 |
else:
|
| 79 |
+
st.error(f"Cascade file not found")
|
| 80 |
|
| 81 |
+
# Initialize CNN detector
|
| 82 |
self.cnn_detector = CNNDeepfakeDetector()
|
| 83 |
|
| 84 |
# Face tracking/smoothing parameters
|
| 85 |
+
self.face_history = {}
|
| 86 |
+
self.face_history_max_size = 10
|
| 87 |
+
self.face_ttl = 5
|
| 88 |
+
self.next_face_id = 0
|
| 89 |
+
self.result_buffer_size = 5
|
|
|
|
|
|
|
|
|
|
|
|
|
| 90 |
self.processing_times = deque(maxlen=30)
|
| 91 |
|
| 92 |
st.success("Models loaded successfully!")
|
| 93 |
|
| 94 |
def detect_faces_haar(self, frame):
|
|
|
|
| 95 |
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
| 96 |
+
faces = self.face_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=5, minSize=(30, 30))
|
|
|
|
|
|
|
|
|
|
| 97 |
return [(x, y, w, h, 0.8) for (x, y, w, h) in faces]
|
| 98 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 99 |
def detect_faces_dnn(self, frame):
|
|
|
|
| 100 |
height, width = frame.shape[:2]
|
| 101 |
+
blob = cv2.dnn.blobFromImage(cv2.resize(frame, (300, 300)), 1.0, (300, 300), (104.0, 177.0, 123.0))
|
|
|
|
| 102 |
self.face_net.setInput(blob)
|
| 103 |
detections = self.face_net.forward()
|
|
|
|
| 104 |
faces = []
|
| 105 |
for i in range(detections.shape[2]):
|
| 106 |
confidence = detections[0, 0, i, 2]
|
| 107 |
+
if confidence > 0.5:
|
| 108 |
box = detections[0, 0, i, 3:7] * np.array([width, height, width, height])
|
| 109 |
(x1, y1, x2, y2) = box.astype("int")
|
|
|
|
| 110 |
x1, y1 = max(0, x1), max(0, y1)
|
| 111 |
x2, y2 = min(width, x2), min(height, y2)
|
| 112 |
w, h = x2 - x1, y2 - y1
|
| 113 |
+
if w > 0 and h > 0:
|
| 114 |
faces.append((x1, y1, w, h, confidence))
|
|
|
|
| 115 |
return faces
|
| 116 |
|
| 117 |
def calculate_iou(self, box1, box2):
|
|
|
|
|
|
|
| 118 |
box1_x1, box1_y1, box1_w, box1_h = box1
|
| 119 |
box2_x1, box2_y1, box2_w, box2_h = box2
|
|
|
|
| 120 |
box1_x2, box1_y2 = box1_x1 + box1_w, box1_y1 + box1_h
|
| 121 |
box2_x2, box2_y2 = box2_x1 + box2_w, box2_y1 + box2_h
|
|
|
|
|
|
|
| 122 |
x_left = max(box1_x1, box2_x1)
|
| 123 |
y_top = max(box1_y1, box2_y1)
|
| 124 |
x_right = min(box1_x2, box2_x2)
|
| 125 |
y_bottom = min(box1_y2, box2_y2)
|
|
|
|
| 126 |
if x_right < x_left or y_bottom < y_top:
|
| 127 |
return 0.0
|
|
|
|
| 128 |
intersection_area = (x_right - x_left) * (y_bottom - y_top)
|
|
|
|
|
|
|
| 129 |
box1_area = box1_w * box1_h
|
| 130 |
box2_area = box2_w * box2_h
|
| 131 |
+
return intersection_area / float(box1_area + box2_area - intersection_area)
|
|
|
|
|
|
|
|
|
|
| 132 |
|
| 133 |
def track_faces(self, faces):
|
| 134 |
matched_faces = []
|
| 135 |
unmatched_detections = list(range(len(faces)))
|
|
|
|
| 136 |
if not self.face_history:
|
| 137 |
for face in faces:
|
| 138 |
face_id = self.next_face_id
|
|
|
|
| 181 |
'result_history': deque(maxlen=self.result_buffer_size)
|
| 182 |
}
|
| 183 |
matched_faces.append((face_id, faces[i]))
|
|
|
|
| 184 |
return matched_faces
|
| 185 |
|
| 186 |
def smooth_face_position(self, face_id):
|
|
|
|
| 187 |
positions = self.face_history[face_id]['positions']
|
|
|
|
| 188 |
if len(positions) == 1:
|
| 189 |
return positions[0]
|
|
|
|
|
|
|
| 190 |
total_weight = 0
|
| 191 |
x, y, w, h = 0, 0, 0, 0
|
|
|
|
| 192 |
for i, pos in enumerate(positions):
|
| 193 |
+
weight = 2 ** i
|
|
|
|
| 194 |
total_weight += weight
|
|
|
|
| 195 |
x += pos[0] * weight
|
| 196 |
y += pos[1] * weight
|
| 197 |
w += pos[2] * weight
|
| 198 |
h += pos[3] * weight
|
| 199 |
+
return (int(x / total_weight), int(y / total_weight), int(w / total_weight), int(h / total_weight))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 200 |
|
| 201 |
def update_face_classification(self, face_id, label, confidence):
|
|
|
|
| 202 |
self.face_history[face_id]['result_history'].append((label, confidence))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 203 |
real_votes = 0
|
| 204 |
fake_votes = 0
|
| 205 |
total_confidence = 0.0
|
|
|
|
| 206 |
for result_label, result_conf in self.face_history[face_id]['result_history']:
|
| 207 |
if result_label == "Real":
|
| 208 |
real_votes += 1
|
|
|
|
| 210 |
elif result_label == "Fake":
|
| 211 |
fake_votes += 1
|
| 212 |
total_confidence += result_conf
|
|
|
|
|
|
|
| 213 |
if real_votes >= fake_votes:
|
| 214 |
smoothed_label = "Real"
|
| 215 |
label_confidence = real_votes / len(self.face_history[face_id]['result_history'])
|
| 216 |
else:
|
| 217 |
smoothed_label = "Fake"
|
| 218 |
label_confidence = fake_votes / len(self.face_history[face_id]['result_history'])
|
|
|
|
|
|
|
| 219 |
avg_confidence = (total_confidence / len(self.face_history[face_id]['result_history'])) * label_confidence
|
|
|
|
|
|
|
| 220 |
self.face_history[face_id]['label'] = smoothed_label
|
| 221 |
self.face_history[face_id]['confidence'] = avg_confidence
|
|
|
|
| 222 |
return smoothed_label, avg_confidence
|
| 223 |
|
| 224 |
def process_video(self, video_path, stframe, status_text, progress_bar, detector_type="dnn", model_type="vit"):
|
|
|
|
| 225 |
use_dnn_current = detector_type == "dnn" and self.use_dnn
|
|
|
|
| 226 |
cap = cv2.VideoCapture(video_path)
|
| 227 |
if not cap.isOpened():
|
| 228 |
st.error(f"Error: Cannot open video source")
|
| 229 |
return
|
|
|
|
|
|
|
| 230 |
frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
| 231 |
frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
| 232 |
fps = cap.get(cv2.CAP_PROP_FPS)
|
| 233 |
total_frames = 250 if video_path != 0 else 0
|
| 234 |
+
if video_path != 0:
|
|
|
|
|
|
|
| 235 |
status_text.text(f"Video Info: {frame_width}x{frame_height}, {fps:.1f} FPS, {total_frames} frames")
|
| 236 |
else:
|
| 237 |
status_text.text(f"Webcam: {frame_width}x{frame_height}")
|
|
|
|
|
|
|
| 238 |
self.face_history = {}
|
| 239 |
self.next_face_id = 0
|
| 240 |
self.processing_times = deque(maxlen=30)
|
|
|
|
| 241 |
frame_count = 0
|
| 242 |
+
process_every_n_frames = 2
|
|
|
|
|
|
|
| 243 |
face_stats = {"Real": 0, "Fake": 0, "Unknown": 0}
|
| 244 |
|
|
|
|
| 245 |
while True:
|
| 246 |
start_time = time.time()
|
|
|
|
| 247 |
ret, frame = cap.read()
|
| 248 |
if not ret:
|
| 249 |
status_text.text("End of video reached")
|
| 250 |
break
|
|
|
|
| 251 |
frame_count += 1
|
|
|
|
| 252 |
if frame_count == 250:
|
| 253 |
st.success("Video Processed Successfully!")
|
| 254 |
break
|
| 255 |
+
if video_path != 0:
|
|
|
|
| 256 |
progress = min(float(frame_count) / float(max(total_frames, 1)), 1.0)
|
| 257 |
progress_bar.progress(progress)
|
|
|
|
| 258 |
process_frame = (frame_count % process_every_n_frames == 0)
|
|
|
|
|
|
|
| 259 |
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
| 260 |
|
| 261 |
if process_frame:
|
| 262 |
+
faces = self.detect_faces_dnn(frame) if use_dnn_current else self.detect_faces_haar(frame)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 263 |
tracked_faces = self.track_faces(faces)
|
| 264 |
+
face_images = []
|
| 265 |
+
face_ids = []
|
| 266 |
for face_id, (x, y, w, h, face_confidence) in tracked_faces:
|
| 267 |
+
if face_id in self.face_history and w > 20 and h > 20:
|
| 268 |
+
sx, sy, sw, sh = self.smooth_face_position(face_id)
|
| 269 |
+
face = frame_rgb[sy:sy+sh, sx:sx+sw]
|
| 270 |
+
if face.size > 0 and face.shape[0] >= 20 and face.shape[1] >= 20:
|
| 271 |
+
face_images.append(face)
|
| 272 |
+
face_ids.append(face_id)
|
| 273 |
+
if face_images:
|
| 274 |
+
if model_type == "vit":
|
| 275 |
+
inputs = self.image_processor(images=face_images, return_tensors="pt")
|
| 276 |
+
with torch.no_grad():
|
| 277 |
+
outputs = self.model(**inputs)
|
| 278 |
+
logits = outputs.logits
|
| 279 |
+
probs = torch.nn.functional.softmax(logits, dim=1)
|
| 280 |
+
preds = torch.argmax(logits, dim=1)
|
| 281 |
+
for i, pred in enumerate(preds):
|
| 282 |
+
label = 'Real' if pred.item() == 1 else 'Fake'
|
| 283 |
+
confidence = probs[i][pred].item()
|
| 284 |
+
self.update_face_classification(face_ids[i], label, confidence)
|
| 285 |
+
elif model_type == "cnn" and self.cnn_detector.model is not None:
|
| 286 |
+
img_arrays = [cv2.resize(face, (128, 128)) / 255.0 for face in face_images]
|
| 287 |
+
img_batch = np.array(img_arrays)
|
| 288 |
+
predictions = self.cnn_detector.model.predict(img_batch)
|
| 289 |
+
for i, prediction in enumerate(predictions):
|
| 290 |
+
confidence = float(prediction[0])
|
| 291 |
+
label = 'Real' if confidence < 0.5 else 'Fake'
|
| 292 |
+
if label == 'Fake':
|
| 293 |
+
confidence = confidence
|
| 294 |
+
else:
|
| 295 |
+
confidence = 1.0 - confidence
|
| 296 |
+
self.update_face_classification(face_ids[i], label, confidence)
|
| 297 |
|
| 298 |
+
for face_id in self.face_history:
|
| 299 |
+
if self.face_history[face_id]['ttl'] > 0:
|
| 300 |
sx, sy, sw, sh = self.smooth_face_position(face_id)
|
|
|
|
| 301 |
cv2.rectangle(frame, (sx, sy), (sx+sw, sy+sh), (0, 255, 255), 2)
|
| 302 |
+
label = self.face_history[face_id]['label'] or "Unknown"
|
| 303 |
+
confidence = self.face_history[face_id]['confidence']
|
| 304 |
+
result_text = f"{label}: {confidence:.2f}"
|
| 305 |
+
text_color = (0, 255, 0) if label == "Real" else (0, 0, 255)
|
| 306 |
+
cv2.rectangle(frame, (sx, sy+sh), (sx+len(result_text)*11, sy+sh+25), (0, 0, 0), -1)
|
| 307 |
+
cv2.putText(frame, result_text, (sx, sy+sh+20),
|
| 308 |
+
cv2.FONT_HERSHEY_SIMPLEX, 0.7, text_color, 2)
|
| 309 |
+
cv2.putText(frame, f"ID:{face_id}", (sx, sy-5),
|
| 310 |
+
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 1)
|
| 311 |
+
if label in face_stats:
|
| 312 |
+
face_stats[label] += 1
|
| 313 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 314 |
process_time = time.time() - start_time
|
| 315 |
self.processing_times.append(process_time)
|
| 316 |
avg_time = sum(self.processing_times) / len(self.processing_times)
|
| 317 |
effective_fps = 1.0 / avg_time if avg_time > 0 else 0
|
| 318 |
|
| 319 |
+
if video_path != 0:
|
|
|
|
| 320 |
progress_percent = (frame_count / total_frames) * 100 if total_frames > 0 else 0
|
| 321 |
cv2.putText(frame, f"Frame: {frame_count}/{total_frames} ({progress_percent:.1f}%)",
|
| 322 |
(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
|
| 323 |
else:
|
| 324 |
cv2.putText(frame, f"Frame: {frame_count}",
|
| 325 |
(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
|
|
|
|
|
|
|
| 326 |
detector_name = "DNN" if use_dnn_current else "Haar Cascade"
|
| 327 |
model_name = "ViT" if model_type == "vit" else "CNN"
|
| 328 |
cv2.putText(frame, f"Detector: {detector_name} | Model: {model_name} | FPS: {effective_fps:.1f}",
|
| 329 |
(10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
|
|
|
|
|
|
|
| 330 |
cv2.putText(frame, f"Tracked faces: {len(self.face_history)}",
|
| 331 |
(10, 90), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
|
|
|
|
|
|
|
| 332 |
stframe.image(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB), channels="RGB")
|
|
|
|
|
|
|
| 333 |
status_text.text(f"Real: {face_stats['Real']} | Fake: {face_stats['Fake']} | FPS: {effective_fps:.1f}")
|
|
|
|
|
|
|
| 334 |
if st.session_state.get('stop_button', False):
|
| 335 |
break
|
|
|
|
|
|
|
| 336 |
cap.release()
|
| 337 |
return face_stats
|
| 338 |
|
|
|
|
| 339 |
def ensure_sample_video():
|
| 340 |
sample_dir = "sample_videos"
|
| 341 |
sample_path = os.path.join(sample_dir, "Sample.mp4")
|
|
|
|
|
|
|
| 342 |
if not os.path.exists(sample_dir):
|
| 343 |
os.makedirs(sample_dir)
|
|
|
|
|
|
|
| 344 |
if not os.path.exists(sample_path):
|
| 345 |
try:
|
| 346 |
with st.spinner("Downloading sample video..."):
|
|
|
|
| 347 |
sample_url = "https://storage.googleapis.com/deepfake-demo/sample_deepfake.mp4"
|
|
|
|
|
|
|
| 348 |
with urllib.request.urlopen(sample_url) as response, open(sample_path, 'wb') as out_file:
|
| 349 |
shutil.copyfileobj(response, out_file)
|
|
|
|
| 350 |
st.success("Sample video downloaded successfully!")
|
| 351 |
except Exception as e:
|
| 352 |
st.error(f"Failed to download sample video: {e}")
|
| 353 |
return None
|
|
|
|
| 354 |
return sample_path
|
| 355 |
|
| 356 |
def main():
|
| 357 |
st.set_page_config(page_title="Deepfake Detector", layout="wide")
|
|
|
|
|
|
|
| 358 |
st.title("Deepfake Detection App")
|
| 359 |
st.markdown("""
|
| 360 |
This app uses computer vision and deep learning to detect deepfake videos.
|
| 361 |
Upload a video or use your webcam to detect if faces are real or manipulated.
|
| 362 |
""")
|
| 363 |
|
|
|
|
| 364 |
if 'detector' not in st.session_state:
|
| 365 |
st.session_state.detector = None
|
|
|
|
| 366 |
if 'stop_button' not in st.session_state:
|
| 367 |
st.session_state.stop_button = False
|
|
|
|
| 368 |
if 'use_sample' not in st.session_state:
|
| 369 |
st.session_state.use_sample = False
|
|
|
|
| 370 |
if 'sample_path' not in st.session_state:
|
| 371 |
st.session_state.sample_path = None
|
| 372 |
|
|
|
|
| 373 |
if st.session_state.detector is None:
|
| 374 |
st.session_state.detector = DeepfakeDetector()
|
| 375 |
|
|
|
|
| 376 |
st.sidebar.title("Options")
|
| 377 |
+
input_option = st.sidebar.radio("Select Input Source", ["Upload Video", "Use Webcam", "Try Sample Video"])
|
| 378 |
+
detector_type = st.sidebar.selectbox("Face Detector", ["DNN (better for close-ups)", "Haar Cascade (faster)"],
|
| 379 |
+
index=0 if st.session_state.detector.use_dnn else 1)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 380 |
detector_option = "dnn" if "DNN" in detector_type else "haar"
|
| 381 |
+
model_type = st.sidebar.selectbox("Deepfake Detection Model", ["Vision Transformer (ViT)", "F3 Net Model"], index=0)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 382 |
model_option = "vit" if "Vision" in model_type else "cnn"
|
| 383 |
|
|
|
|
| 384 |
col1, col2 = st.columns([3, 1])
|
|
|
|
| 385 |
with col1:
|
|
|
|
| 386 |
video_placeholder = st.empty()
|
|
|
|
| 387 |
with col2:
|
|
|
|
| 388 |
status_text = st.empty()
|
| 389 |
progress_bar = st.empty()
|
|
|
|
|
|
|
| 390 |
st.subheader("Results")
|
| 391 |
results_area = st.empty()
|
|
|
|
|
|
|
| 392 |
if st.button("Stop Processing"):
|
| 393 |
st.session_state.stop_button = True
|
| 394 |
|
|
|
|
| 395 |
if input_option == "Upload Video":
|
| 396 |
uploaded_file = st.sidebar.file_uploader("Choose a video file", type=["mp4", "avi", "mov", "mkv"])
|
|
|
|
| 397 |
if uploaded_file is not None:
|
| 398 |
st.session_state.stop_button = False
|
|
|
|
|
|
|
| 399 |
tfile = tempfile.NamedTemporaryFile(delete=False)
|
| 400 |
tfile.write(uploaded_file.read())
|
| 401 |
video_path = tfile.name
|
| 402 |
+
face_stats = st.session_state.detector.process_video(video_path, video_placeholder, status_text,
|
| 403 |
+
progress_bar, detector_option, model_option)
|
| 404 |
+
results_df = {"Category": ["Real Faces", "Fake Faces"], "Count": [face_stats["Real"], face_stats["Fake"]]}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 405 |
results_area.dataframe(results_df)
|
|
|
|
|
|
|
| 406 |
os.unlink(video_path)
|
|
|
|
| 407 |
elif input_option == "Use Webcam":
|
|
|
|
| 408 |
st.session_state.stop_button = False
|
|
|
|
| 409 |
if st.sidebar.button("Start Webcam"):
|
| 410 |
+
face_stats = st.session_state.detector.process_video(0, video_placeholder, status_text, progress_bar,
|
| 411 |
+
detector_option, model_option)
|
| 412 |
+
results_df = {"Category": ["Real Faces", "Fake Faces"], "Count": [face_stats["Real"], face_stats["Fake"]]}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 413 |
results_area.dataframe(results_df)
|
|
|
|
| 414 |
elif input_option == "Try Sample Video":
|
|
|
|
| 415 |
st.session_state.stop_button = False
|
|
|
|
|
|
|
| 416 |
sample_path = ensure_sample_video()
|
| 417 |
+
if sample_path and st.sidebar.button("Process Sample Video"):
|
| 418 |
+
face_stats = st.session_state.detector.process_video(sample_path, video_placeholder, status_text,
|
| 419 |
+
progress_bar, detector_option, model_option)
|
| 420 |
+
results_df = {"Category": ["Real Faces", "Fake Faces"], "Count": [face_stats["Real"], face_stats["Fake"]]}
|
| 421 |
+
results_area.dataframe(results_df)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 422 |
|
| 423 |
if __name__ == "__main__":
|
| 424 |
+
main()
|