# import numpy as np # import cv2 # from skimage.util import img_as_float # import tensorflow as tf # import matplotlib.pyplot as plt # import time # import scipy.io # from scipy.sparse import spdiags # def preprocess_raw_video(videoFilePath, dim=36): # ######################################################################### # # set up # t = [] # i = 0 # vidObj = cv2.VideoCapture(videoFilePath); # totalFrames = int(vidObj.get(cv2.CAP_PROP_FRAME_COUNT)) # get total frame size # Xsub = np.zeros((totalFrames, dim, dim, 3), dtype = np.float32) # height = vidObj.get(cv2.CAP_PROP_FRAME_HEIGHT) # width = vidObj.get(cv2.CAP_PROP_FRAME_WIDTH) # success, img = vidObj.read() # dims = img.shape # print("Orignal Height", height) # print("Original width", width) # ######################################################################### # # Crop each frame size into dim x dim # while success: # t.append(vidObj.get(cv2.CAP_PROP_POS_MSEC))# current timestamp in milisecond # vidLxL = cv2.resize(img_as_float(img[:, int(width/2)-int(height/2 + 1):int(height/2)+int(width/2), :]), (dim, dim), interpolation = cv2.INTER_AREA) # vidLxL = cv2.rotate(vidLxL, cv2.ROTATE_90_CLOCKWISE) # rotate 90 degree # vidLxL = cv2.cvtColor(vidLxL.astype('float32'), cv2.COLOR_BGR2RGB) # vidLxL[vidLxL > 1] = 1 # vidLxL[vidLxL < (1/255)] = 1/255 # Xsub[i, :, :, :] = vidLxL # success, img = vidObj.read() # read the next one # i = i + 1 # plt.imshow(Xsub[0]) # plt.title('Sample Preprocessed Frame') # plt.show() # ######################################################################### # # Normalized Frames in the motion branch # normalized_len = len(t) - 1 # dXsub = np.zeros((normalized_len, dim, dim, 3), dtype = np.float32) # for j in range(normalized_len - 1): # dXsub[j, :, :, :] = (Xsub[j+1, :, :, :] - Xsub[j, :, :, :]) / (Xsub[j+1, :, :, :] + Xsub[j, :, :, :]) # dXsub = dXsub / np.std(dXsub) # ######################################################################### # # Normalize raw frames in the apperance branch # Xsub = Xsub - np.mean(Xsub) # Xsub = Xsub / np.std(Xsub) # Xsub = Xsub[:totalFrames-1, :, :, :] # ######################################################################### # # Plot an example of data after preprocess # dXsub = np.concatenate((dXsub, Xsub), axis = 3); # return dXsub # def detrend(signal, Lambda): # """detrend(signal, Lambda) -> filtered_signal # This function applies a detrending filter. # This code is based on the following article "An advanced detrending method with application # to HRV analysis". Tarvainen et al., IEEE Trans on Biomedical Engineering, 2002. # *Parameters* # ``signal`` (1d numpy array): # The signal where you want to remove the trend. # ``Lambda`` (int): # The smoothing parameter. # *Returns* # ``filtered_signal`` (1d numpy array): # The detrended signal. # """ # signal_length = signal.shape[0] # # observation matrix # H = np.identity(signal_length) # # second-order difference matrix # ones = np.ones(signal_length) # minus_twos = -2 * np.ones(signal_length) # diags_data = np.array([ones, minus_twos, ones]) # diags_index = np.array([0, 1, 2]) # D = spdiags(diags_data, diags_index, (signal_length - 2), signal_length).toarray() # filtered_signal = np.dot((H - np.linalg.inv(H + (Lambda ** 2) * np.dot(D.T, D))), signal) # return filtered_signal # import numpy as np # import cv2 # import mediapipe as mp # from scipy.sparse import spdiags # import logging # # Set up a logger # logger = logging.getLogger(__name__) # def detrend(signal, Lambda): # """Applies a detrending filter to a signal.""" # signal_length = signal.shape[0] # H = np.identity(signal_length) # ones = np.ones(signal_length) # minus_twos = -2 * np.ones(signal_length) # diags_data = np.array([ones, minus_twos, ones]) # diags_index = np.array([0, 1, 2]) # D = spdiags(diags_data, diags_index, (signal_length - 2), signal_length).toarray() # filtered_signal = np.dot((H - np.linalg.inv(H + (Lambda ** 2) * np.dot(D.T, D))), signal) # return filtered_signal # def preprocess_raw_video(video_path, dim=36): # """ # Preprocesses a video to extract the forehead region using MediaPipe. # This version completely replaces the need for dlib. # """ # # --- 1. SETUP MEDIAPIPE --- # mp_face_mesh = mp.solutions.face_mesh # face_mesh = mp_face_mesh.FaceMesh( # static_image_mode=False, # max_num_faces=1, # min_detection_confidence=0.5 # ) # # --- 2. VIDEO CAPTURE SETUP --- # vidObj = cv2.VideoCapture(video_path) # if not vidObj.isOpened(): # raise IOError(f"Cannot open video file: {video_path}") # total_frames = int(vidObj.get(cv2.CAP_PROP_FRAME_COUNT)) # logger.info(f"Total frames in video: {total_frames}") # # --- 3. FRAME-BY-FRAME PROCESSING --- # all_forehead_rois = [] # frames_processed = 0 # while True: # success, frame = vidObj.read() # if not success: # break # frames_processed += 1 # # Convert the BGR image to RGB. # rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) # # Process the frame and find face landmarks. # results = face_mesh.process(rgb_frame) # if results.multi_face_landmarks: # # Assume only one face. # face_landmarks = results.multi_face_landmarks[0] # # --- 4. EXTRACT FOREHEAD REGION --- # # Using specific MediaPipe landmark indices for the forehead # # Index 10 is roughly the top of the nose bridge. # # We will create a bounding box above this point. # # Get landmark coordinates (normalized 0-1, so we multiply by frame dimensions) # h, w, _ = frame.shape # # We'll define the forehead ROI based on a few key points. # # A common approach is to use points on the eyebrows and above. # # E.g., Landmark 105 (right), 334 (left), 10 (center-ish top) # p1 = face_landmarks.landmark[105] # p2 = face_landmarks.landmark[334] # p_top = face_landmarks.landmark[10] # x1 = int(p1.x * w) # y1 = int(p1.y * h) # x2 = int(p2.x * w) # y2 = int(p_top.y * h) # # Add some padding to get a good region # padding_y = 15 # padding_x = 10 # y_start = max(0, y2 - padding_y) # y_end = min(h, y1 + padding_y) # x_start = min(x1, x2) - padding_x # x_end = max(x1, x2) + padding_x # # Crop the forehead # forehead_roi = frame[y_start:y_end, x_start:x_end] # if forehead_roi.size > 0: # # Resize to the dimension required by the model (36x36) # resized_roi = cv2.resize(forehead_roi, (dim, dim), interpolation=cv2.INTER_AREA) # all_forehead_rois.append(resized_roi) # vidObj.release() # face_mesh.close() # if not all_forehead_rois: # raise ValueError("Could not detect a face or forehead in any frame of the video.") # logger.info(f"Successfully extracted forehead ROI from {len(all_forehead_rois)} frames.") # # --- 5. POST-PROCESSING FOR THE MODEL --- # Xsub = np.array(all_forehead_rois, dtype=np.float32) / 255.0 # # Calculate motion stream (temporal difference) # dXsub = np.diff(Xsub, axis=0) # dXsub = np.concatenate((dXsub, np.zeros((1, dim, dim, 3), dtype=np.float32)), axis=0) # dXsub = (dXsub - np.mean(dXsub)) / np.std(dXsub) # # Normalize appearance stream # Xsub = (Xsub - np.mean(Xsub)) / np.std(Xsub) # # Combine motion and appearance streams as expected by the model # # Note: The original code seems to expect 6 channels, but the model input shapes # # suggest two separate inputs of 3 channels each. We will create the 6-channel # # array as per the original preprocessing logic. # dXsub = np.concatenate((dXsub, Xsub), axis=3) # return dXsub # import cv2 # import numpy as np # import mediapipe as mp # from skimage.transform import resize # import logging # # Set up a logger # logger = logging.getLogger(__name__) # def preprocess_video_for_inference(video_path, target_size=(36, 36)): # """ # Preprocesses a video by detecting and cropping faces using MediaPipe, # then calculates the normalized motion differences for model inference. # This is a direct replacement for the previous dlib-based approach. # """ # # --- 1. SETUP MEDIAPIPE --- # # Use the lightweight and efficient Face Detection model # mp_face_detection = mp.solutions.face_detection # face_detection = mp_face_detection.FaceDetection(model_selection=0, min_detection_confidence=0.5) # # --- 2. VIDEO CAPTURE AND SETUP --- # cap = cv2.VideoCapture(video_path) # if not cap.isOpened(): # raise ValueError(f"Error: Could not open video file: {video_path}") # cropped_frames = [] # total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) # logger.info(f"Total frames to process: {total_frames}") # # --- 3. FRAME-BY-FRAME PROCESSING --- # while cap.isOpened(): # ret, frame = cap.read() # if not ret: # break # # Convert the BGR image to RGB for MediaPipe # rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) # # Process the frame to detect faces # results = face_detection.process(rgb_frame) # if results.detections: # # Get the bounding box of the first detected face # detection = results.detections[0] # bboxC = detection.location_data.relative_bounding_box # ih, iw, _ = frame.shape # # Calculate absolute coordinates # x, y, w, h = int(bboxC.xmin * iw), int(bboxC.ymin * ih), \ # int(bboxC.width * iw), int(bboxC.height * ih) # # Ensure coordinates are valid # x, y, w, h = max(0, x), max(0, y), max(0, w), max(0, h) # face = frame[y:y+h, x:x+w] # if face.size > 0: # # Resize face to the target dimension (e.g., 36x36) # resized_face = resize(face, target_size, anti_aliasing=True) # # Append all three color channels # cropped_frames.append(resized_face) # cap.release() # face_detection.close() # if not cropped_frames: # raise ValueError("Could not detect a face in any frame of the video.") # logger.info(f"Successfully extracted face crops from {len(cropped_frames)} frames.") # # --- 4. CORE POST-PROCESSING LOGIC (UNCHANGED) --- # Xsub = np.array(cropped_frames, dtype=np.float32) # # Normalize Appearance Stream # Xsub = (Xsub - np.mean(Xsub)) / np.std(Xsub) # # Calculate Motion Stream (Temporal Difference) # dXsub = np.diff(Xsub, axis=0) # # Pad the last frame to maintain original length # dXsub = np.concatenate((dXsub, np.zeros((1, target_size[0], target_size[1], 3), dtype=np.float32)), axis=0) # # Normalize Motion Stream # dXsub = (dXsub - np.mean(dXsub)) / np.std(dXsub) # # Combine streams into a 6-channel array as required by the model # final_input = np.concatenate((dXsub, Xsub), axis=3) # return final_input import cv2 import numpy as np import mediapipe as mp from skimage.transform import resize import logging logger = logging.getLogger(__name__) def preprocess_video_for_inference(video_path, target_size=(36, 36)): mp_face_detection = mp.solutions.face_detection face_detection = mp_face_detection.FaceDetection(model_selection=0, min_detection_confidence=0.5) cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise ValueError(f"Error: Could not open video file: {video_path}") cropped_frames = [] while cap.isOpened(): ret, frame = cap.read() if not ret: break rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) results = face_detection.process(rgb_frame) if results.detections: detection = results.detections[0] bboxC = detection.location_data.relative_bounding_box ih, iw, _ = frame.shape x, y, w, h = int(bboxC.xmin * iw), int(bboxC.ymin * ih), \ int(bboxC.width * iw), int(bboxC.height * ih) x, y, w, h = max(0, x), max(0, y), max(0, w), max(0, h) face = frame[y:y+h, x:x+w] if face.size > 0: resized_face = resize(face, target_size, anti_aliasing=True) cropped_frames.append(resized_face) cap.release() face_detection.close() if not cropped_frames: raise ValueError("Could not detect a face in any frame of the video.") Xsub = np.array(cropped_frames, dtype=np.float32) Xsub = (Xsub - np.mean(Xsub)) / np.std(Xsub) dXsub = np.diff(Xsub, axis=0) dXsub = np.concatenate((dXsub, np.zeros((1, target_size[0], target_size[1], 3), dtype=np.float32)), axis=0) dXsub = (dXsub - np.mean(dXsub)) / np.std(dXsub) final_input = np.concatenate((dXsub, Xsub), axis=3) return final_input