Spaces:
Runtime error
Runtime error
| """ | |
| Video Preprocessor for Phase 5 Model | |
| =================================== | |
| Handles video loading, resizing, and frame extraction | |
| """ | |
| import cv2 | |
| import numpy as np | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| class VideoPreprocessor: | |
| """Video preprocessing for Phase 5 model""" | |
| def __init__(self, target_fps=8, max_frames=32, img_size=(224, 224)): | |
| self.target_fps = target_fps | |
| self.max_frames = max_frames | |
| self.img_size = img_size | |
| def preprocess_video(self, video_path): | |
| """ | |
| Preprocess video file for model input | |
| Args: | |
| video_path: Path to video file | |
| Returns: | |
| numpy array of shape (channels, frames, height, width) or None if error | |
| """ | |
| try: | |
| cap = cv2.VideoCapture(video_path) | |
| if not cap.isOpened(): | |
| logger.error(f"Could not open video: {video_path}") | |
| return None | |
| # Get video properties | |
| fps = cap.get(cv2.CAP_PROP_FPS) | |
| total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) | |
| if fps <= 0 or total_frames <= 0: | |
| logger.error(f"Invalid video properties: fps={fps}, frames={total_frames}") | |
| cap.release() | |
| return None | |
| # Calculate frame sampling rate | |
| frame_step = max(1, int(fps / self.target_fps)) | |
| frames = [] | |
| frame_count = 0 | |
| while len(frames) < self.max_frames: | |
| ret, frame = cap.read() | |
| if not ret: | |
| break | |
| # Sample frames based on target FPS | |
| if frame_count % frame_step == 0: | |
| # Resize frame | |
| frame = cv2.resize(frame, self.img_size) | |
| # Convert BGR to RGB | |
| frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) | |
| # Normalize to [0, 1] | |
| frame = frame.astype(np.float32) / 255.0 | |
| frames.append(frame) | |
| frame_count += 1 | |
| cap.release() | |
| if len(frames) == 0: | |
| logger.error("No frames extracted from video") | |
| return None | |
| # Pad or truncate to max_frames | |
| if len(frames) < self.max_frames: | |
| # Repeat last frame to reach max_frames | |
| last_frame = frames[-1] | |
| while len(frames) < self.max_frames: | |
| frames.append(last_frame.copy()) | |
| elif len(frames) > self.max_frames: | |
| frames = frames[:self.max_frames] | |
| # Convert to numpy array and transpose to (C, T, H, W) | |
| video_array = np.array(frames) # (T, H, W, C) | |
| video_array = np.transpose(video_array, (3, 0, 1, 2)) # (C, T, H, W) | |
| logger.info(f"Processed video: {video_array.shape}") | |
| return video_array | |
| except Exception as e: | |
| logger.error(f"Error preprocessing video: {e}") | |
| return None |