Update crop faces with mtcnn

01-crop_faces_with_mtcnn.py

@@ -12,67 +12,88 @@ print(physical_devices)
 if physical_devices:
     tf.config.experimental.set_memory_growth(physical_devices[0], True)
 
-base_path = '.\\train_sample_videos\\'
-videos_path = os.path.join(base_path, 'Deepfakes')
+base_path = '.\\train_sample_videos\\FaceForensics++_C23\\'
 
 def get_filename_only(file_path):
     file_basename = os.path.basename(file_path)
     filename_only = file_basename.split('.')[0]
     return filename_only
 
-with open(os.path.join(base_path, 'csv', 'Deepfakes.csv'), newline='', encoding='utf-8') as csvfile:
-    reader = csv.DictReader(csvfile)
-    metadata = {}
-    for row in reader:
-        metadata[row['File Path']] = row['Label'].strip().upper()
-    print(len(metadata))
+# Iterate over all subfolders in base_path (excluding 'csv')
+for folder_name in sorted(os.listdir(base_path)):
+    folder_path = os.path.join(base_path, folder_name)
+    if not os.path.isdir(folder_path) or folder_name == 'csv':
+        continue
 
-for filename in metadata.keys():
-    video_basename = os.path.basename(filename)
-    tmp_path = os.path.join(videos_path, get_filename_only(video_basename))
-    print('Processing Directory: ' + tmp_path)
-    faces_path = os.path.join(tmp_path, 'faces')
-    if os.path.isdir(faces_path) and len(os.listdir(faces_path)) > 0:
-        print('Skipping (faces already exist): ' + faces_path)
+    csv_file = os.path.join(base_path, 'csv', folder_name + '.csv')
+    if not os.path.isfile(csv_file):
+        print(f'CSV not found for {folder_name}, skipping: {csv_file}')
         continue
-    frame_images = [x for x in os.listdir(tmp_path) if os.path.isfile(os.path.join(tmp_path, x))]
-    print('Creating Directory: ' + faces_path)
-    os.makedirs(faces_path, exist_ok=True)
-    print('Cropping Faces from Images...')
 
-    for frame in frame_images:
-        print('Processing ', frame)
-        detector = MTCNN()
-        image = cv2.cvtColor(cv2.imread(os.path.join(tmp_path, frame)), cv2.COLOR_BGR2RGB)
-        results = detector.detect_faces(image)
-        print('Face Detected: ', len(results))
-        count = 0
-        
-        for result in results:
-            bounding_box = result['box']
-            print(bounding_box)
-            confidence = result['confidence']
-            print(confidence)
-            if len(results) < 2 or confidence > 0.95:
-                margin_x = bounding_box[2] * 0.3  # 30% as the margin
-                margin_y = bounding_box[3] * 0.3  # 30% as the margin
-                x1 = int(bounding_box[0] - margin_x)
-                if x1 < 0:
-                    x1 = 0
-                x2 = int(bounding_box[0] + bounding_box[2] + margin_x)
-                if x2 > image.shape[1]:
-                    x2 = image.shape[1]
-                y1 = int(bounding_box[1] - margin_y)
-                if y1 < 0:
-                    y1 = 0
-                y2 = int(bounding_box[1] + bounding_box[3] + margin_y)
-                if y2 > image.shape[0]:
-                    y2 = image.shape[0]
-                print(x1, y1, x2, y2)
-                crop_image = image[y1:y2, x1:x2]
-                new_filename = '{}-{:02d}.png'.format(os.path.join(faces_path, get_filename_only(frame)), count)
-                count = count + 1
-                cv2.imwrite(new_filename, cv2.cvtColor(crop_image, cv2.COLOR_RGB2BGR))
-            else:
-                print('Skipped a face..')
+    print(f'\n{"="*60}')
+    print(f'Processing folder: {folder_name}')
+    print(f'{"="*60}')
+
+    with open(csv_file, newline='', encoding='utf-8') as csvfile:
+        reader = csv.DictReader(csvfile)
+        metadata = {}
+        for row in reader:
+            metadata[row['File Path']] = row['Label'].strip().upper()
+        print(f'{folder_name}: {len(metadata)} entries')
+
+    for filename in metadata.keys():
+        video_basename = os.path.basename(filename)
+        tmp_path = os.path.join(folder_path, get_filename_only(video_basename))
+        print('Processing Directory: ' + tmp_path)
+        faces_path = os.path.join(tmp_path, 'faces')
+        if os.path.isdir(faces_path) and len(os.listdir(faces_path)) > 0:
+            print('Skipping (faces already exist): ' + faces_path)
+            continue
+        if not os.path.isdir(tmp_path):
+            print('Directory not found, skipping: ' + tmp_path)
+            continue
+        frame_images = [x for x in os.listdir(tmp_path) if os.path.isfile(os.path.join(tmp_path, x))]
+        print('Creating Directory: ' + faces_path)
+        os.makedirs(faces_path, exist_ok=True)
+        print('Cropping Faces from Images...')
+
+        for frame in frame_images:
+            print('Processing ', frame)
+            try:
+                detector = MTCNN()
+                image = cv2.cvtColor(cv2.imread(os.path.join(tmp_path, frame)), cv2.COLOR_BGR2RGB)
+                results = detector.detect_faces(image)
+            except Exception as e:
+                print(f'Error detecting faces in {frame}: {e}')
+                continue
+            print('Face Detected: ', len(results))
+            count = 0
+            
+            for result in results:
+                bounding_box = result['box']
+                print(bounding_box)
+                confidence = result['confidence']
+                print(confidence)
+                if len(results) < 2 or confidence > 0.95:
+                    margin_x = bounding_box[2] * 0.3  # 30% as the margin
+                    margin_y = bounding_box[3] * 0.3  # 30% as the margin
+                    x1 = int(bounding_box[0] - margin_x)
+                    if x1 < 0:
+                        x1 = 0
+                    x2 = int(bounding_box[0] + bounding_box[2] + margin_x)
+                    if x2 > image.shape[1]:
+                        x2 = image.shape[1]
+                    y1 = int(bounding_box[1] - margin_y)
+                    if y1 < 0:
+                        y1 = 0
+                    y2 = int(bounding_box[1] + bounding_box[3] + margin_y)
+                    if y2 > image.shape[0]:
+                        y2 = image.shape[0]
+                    print(x1, y1, x2, y2)
+                    crop_image = image[y1:y2, x1:x2]
+                    new_filename = '{}-{:02d}.png'.format(os.path.join(faces_path, get_filename_only(frame)), count)
+                    count = count + 1
+                    cv2.imwrite(new_filename, cv2.cvtColor(crop_image, cv2.COLOR_RGB2BGR))
+                else:
+                    print('Skipped a face..')
     

App/app.py

@@ -8,9 +8,7 @@ import subprocess
 import cv2
 import numpy as np
 import imageio_ffmpeg
-import mediapipe as mp
-from mediapipe.tasks.python import BaseOptions
-from mediapipe.tasks.python.vision import FaceDetector, FaceDetectorOptions
+from ultralytics import YOLO
 from flask import Flask, request, render_template, send_from_directory, jsonify
 from werkzeug.utils import secure_filename
 import uuid
@@ -42,14 +40,11 @@ sys.stderr = _stderr
 logger.info('Model loaded successfully')
 INPUT_SIZE = 128
 
-# Initialize MediaPipe face detector
-logger.info('Initializing MediaPipe face detector')
-FACE_MODEL_PATH = os.path.join(os.path.dirname(__file__), 'blaze_face_short_range.tflite')
-face_detector_options = FaceDetectorOptions(
-    base_options=BaseOptions(model_asset_path=FACE_MODEL_PATH),
-    min_detection_confidence=0.5
-)
-logger.info('MediaPipe face detector ready')
+# Initialize YOLO face detector
+logger.info('Initializing YOLO face detector')
+FACE_MODEL_PATH = os.path.join(os.path.dirname(__file__), 'yolov8n-face.pt')
+face_detector = YOLO(FACE_MODEL_PATH)
+logger.info('YOLO face detector ready')
 
 # In-memory job store: job_id -> {status, result, ...}
 jobs = {}
@@ -103,32 +98,30 @@ def extract_faces_from_video(video_path):
         cap.release()
         return faces
 
-    with FaceDetector.create_from_options(face_detector_options) as face_det:
-        while cap.isOpened():
-            frame_id = cap.get(cv2.CAP_PROP_POS_FRAMES)
-            ret, frame = cap.read()
-            if not ret:
-                break
-            if frame_id % math.floor(frame_rate) == 0:
-                image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-                mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=image_rgb)
-                results = face_det.detect(mp_image)
-                for detection in results.detections:
-                    score = detection.categories[0].score
-                    if score > 0.5:
-                        bbox = detection.bounding_box
-                        bx, by, bw, bh = bbox.origin_x, bbox.origin_y, bbox.width, bbox.height
-                        h, w = image_rgb.shape[:2]
-                        margin_x = int(bw * 0.3)
-                        margin_y = int(bh * 0.3)
-                        x1 = max(0, bx - margin_x)
-                        x2 = min(w, bx + bw + margin_x)
-                        y1 = max(0, by - margin_y)
-                        y2 = min(h, by + bh + margin_y)
-                        crop = image_rgb[y1:y2, x1:x2]
-                        if crop.size > 0:
-                            crop_resized = cv2.resize(crop, (INPUT_SIZE, INPUT_SIZE))
-                            faces.append(crop_resized)
+    while cap.isOpened():
+        frame_id = cap.get(cv2.CAP_PROP_POS_FRAMES)
+        ret, frame = cap.read()
+        if not ret:
+            break
+        if frame_id % math.floor(frame_rate) == 0:
+            image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+            h, w = image_rgb.shape[:2]
+            results = face_detector(frame, verbose=False)[0]
+            for box in results.boxes:
+                if box.conf[0] > 0.5:
+                    bx1, by1, bx2, by2 = map(int, box.xyxy[0])
+                    bw = bx2 - bx1
+                    bh = by2 - by1
+                    margin_x = int(bw * 0.3)
+                    margin_y = int(bh * 0.3)
+                    x1 = max(0, bx1 - margin_x)
+                    x2 = min(w, bx2 + margin_x)
+                    y1 = max(0, by1 - margin_y)
+                    y2 = min(h, by2 + margin_y)
+                    crop = image_rgb[y1:y2, x1:x2]
+                    if crop.size > 0:
+                        crop_resized = cv2.resize(crop, (INPUT_SIZE, INPUT_SIZE))
+                        faces.append(crop_resized)
 
     cap.release()
     logger.info('Face extraction complete — %d faces found', len(faces))
@@ -155,45 +148,43 @@ def create_processed_video(video_path, output_path, face_scores=None):
         return
 
     frame_count = 0
-    with FaceDetector.create_from_options(face_detector_options) as face_det:
-        while cap.isOpened():
-            ret, frame = cap.read()
-            if not ret:
-                break
-            image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-            mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=image_rgb)
-            results = face_det.detect(mp_image)
-            for detection in results.detections:
-                det_score = detection.categories[0].score
-                if det_score > 0.5:
-                    bbox = detection.bounding_box
-                    bx, by, bw, bh = bbox.origin_x, bbox.origin_y, bbox.width, bbox.height
-                    x, y = max(0, bx), max(0, by)
-
-                    # Crop and predict this face individually
-                    margin_x = int(bw * 0.3)
-                    margin_y = int(bh * 0.3)
-                    x1 = max(0, bx - margin_x)
-                    x2 = min(w, bx + bw + margin_x)
-                    y1 = max(0, by - margin_y)
-                    y2 = min(h, by + bh + margin_y)
-                    crop = image_rgb[y1:y2, x1:x2]
-                    if crop.size > 0:
-                        crop_resized = cv2.resize(crop, (INPUT_SIZE, INPUT_SIZE))
-                        face_input = np.array([crop_resized], dtype='float32') / 255.0
-                        score = float(model.predict(face_input, verbose=0)[0][0])
-                    else:
-                        score = 0.0
-
-                    is_real = score > 0.5
-                    label = 'REAL' if is_real else 'FAKE'
-                    color = (0, 255, 0) if is_real else (0, 0, 255)
-                    cv2.rectangle(frame, (x, y), (x + bw, y + bh), color, 2)
-                    text = f'{label} {score:.2f}'
-                    cv2.putText(frame, text, (x, y - 10),
-                                cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2)
-            out.write(frame)
-            frame_count += 1
+    while cap.isOpened():
+        ret, frame = cap.read()
+        if not ret:
+            break
+        image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+        results = face_detector(frame, verbose=False)[0]
+        for box in results.boxes:
+            if box.conf[0] > 0.5:
+                bx1, by1, bx2, by2 = map(int, box.xyxy[0])
+                bw = bx2 - bx1
+                bh = by2 - by1
+                x, y = max(0, bx1), max(0, by1)
+
+                # Crop and predict this face individually
+                margin_x = int(bw * 0.3)
+                margin_y = int(bh * 0.3)
+                x1 = max(0, bx1 - margin_x)
+                x2 = min(w, bx2 + margin_x)
+                y1 = max(0, by1 - margin_y)
+                y2 = min(h, by2 + margin_y)
+                crop = image_rgb[y1:y2, x1:x2]
+                if crop.size > 0:
+                    crop_resized = cv2.resize(crop, (INPUT_SIZE, INPUT_SIZE))
+                    face_input = np.array([crop_resized], dtype='float32') / 255.0
+                    score = float(model.predict(face_input, verbose=0)[0][0])
+                else:
+                    score = 0.0
+
+                is_real = score > 0.5
+                label = 'REAL' if is_real else 'FAKE'
+                color = (0, 255, 0) if is_real else (0, 0, 255)
+                cv2.rectangle(frame, (x, y), (bx2, by2), color, 2)
+                text = f'{label} {score:.2f}'
+                cv2.putText(frame, text, (x, y - 10),
+                            cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2)
+        out.write(frame)
+        frame_count += 1
 
     cap.release()
     out.release()

App/yolov8n-face.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d545bf1add5aa736a4febac4f4f9245a6d596cd0fe70d5d57989fe0cb9e626ca
+size 6389512

requirements.txt

@@ -8,6 +8,6 @@ h5py
 split_folders
 flask
 werkzeug
-mediapipe
+ultralytics
 imageio-ffmpeg
 pillow