"""Video processing — extract frames from uploaded videos.""" import logging import numpy as np log = logging.getLogger("surprise.video") def validate_video(path: str) -> dict: """Open the video and return metadata. Raises ValueError on failure.""" import cv2 cap = cv2.VideoCapture(path) if not cap.isOpened(): raise ValueError("Could not open video file (unsupported format?)") fps = cap.get(cv2.CAP_PROP_FPS) or 30.0 n_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) duration = n_frames / fps if fps > 0 else 0 cap.release() if n_frames < 4: raise ValueError(f"Video too short: only {n_frames} frames (need ≥4)") return { "fps": float(fps), "n_frames": n_frames, "width": width, "height": height, "duration": float(duration), } def extract_frames( path: str, target_size: int = 224, max_frames: int = 240, target_fps: int = 8, ) -> np.ndarray: """ Extract frames from video, resampled to target_fps and resized to target_size. Returns: np.ndarray of shape (N, H, W, 3), dtype uint8, RGB. """ import cv2 cap = cv2.VideoCapture(path) if not cap.isOpened(): raise ValueError("Cannot open video") src_fps = cap.get(cv2.CAP_PROP_FPS) or 30.0 stride = max(1, int(round(src_fps / target_fps))) frames = [] idx = 0 while True: ret, frame = cap.read() if not ret: break if idx % stride == 0: frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) frame = _center_crop_resize(frame, target_size) frames.append(frame) if len(frames) >= max_frames: break idx += 1 cap.release() if len(frames) < 4: raise ValueError(f"Too few frames extracted ({len(frames)})") return np.stack(frames, axis=0) def _center_crop_resize(img: np.ndarray, size: int) -> np.ndarray: """Center-crop to square then resize to (size, size).""" import cv2 h, w = img.shape[:2] s = min(h, w) y0 = (h - s) // 2 x0 = (w - s) // 2 img = img[y0 : y0 + s, x0 : x0 + s] img = cv2.resize(img, (size, size), interpolation=cv2.INTER_AREA) return img