Spaces:
Runtime error
Runtime error
File size: 3,244 Bytes
8b9385d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """
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 |