Garima1030/mediaforensics-platform

1"""2Frame extraction from video files using OpenCV.3Samples frames at regular intervals for analysis.4"""5import cv26import numpy as np7from typing import List, Tuple, Dict8from pathlib import Path9import logging10 11logger = logging.getLogger(__name__)12 13 14class FrameExtractor:15    def __init__(self, sample_rate: int = 10, max_frames: int = 100):16        """17        Args:18            sample_rate: Extract every Nth frame19            max_frames: Maximum number of frames to extract20        """21        self.sample_rate = sample_rate22        self.max_frames = max_frames23 24    def extract_from_video(self, video_path: str) -> Tuple[List[np.ndarray], Dict]:25        """26        Extract frames from a video file.27        Returns (frames, metadata)28        """29        cap = cv2.VideoCapture(video_path)30 31        if not cap.isOpened():32            raise ValueError(f"Could not open video: {video_path}")33 34        # Get video metadata35        fps = cap.get(cv2.CAP_PROP_FPS)36        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))37        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))38        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))39        duration = total_frames / fps if fps > 0 else 040 41        metadata = {42            "fps": round(fps, 2),43            "total_frames": total_frames,44            "width": width,45            "height": height,46            "duration_sec": round(duration, 2),47        }48 49        logger.info(f"[FrameExtractor] Video: {width}x{height} @ {fps}fps, {duration:.1f}s")50 51        frames = []52        frame_indices = []53        frame_idx = 054 55        while cap.isOpened() and len(frames) < self.max_frames:56            ret, frame = cap.read()57            if not ret:58                break59 60            if frame_idx % self.sample_rate == 0:61                # Convert BGR to RGB62                frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)63                frames.append(frame_rgb)64                frame_indices.append(frame_idx)65 66            frame_idx += 167 68        cap.release()69        metadata["frames_extracted"] = len(frames)70        metadata["frame_indices"] = frame_indices71 72        logger.info(f"[FrameExtractor] Extracted {len(frames)} frames")73        return frames, metadata74 75    def load_image(self, image_path: str) -> Tuple[np.ndarray, Dict]:76        """77        Load a single image file.78        Returns (frame, metadata)79        """80        img = cv2.imread(image_path)81        if img is None:82            raise ValueError(f"Could not load image: {image_path}")83 84        img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)85        h, w = img_rgb.shape[:2]86 87        metadata = {88            "width": w,89            "height": h,90            "duration_sec": None,91            "fps": None,92            "frames_extracted": 1,93            "frame_indices": [0],94        }95 96        return img_rgb, metadata97 98    def load_media(self, file_path: str) -> Tuple[List[np.ndarray], Dict]:99        """100        Auto-detect media type and load accordingly.101        Returns (frames_list, metadata)102        """103        ext = Path(file_path).suffix.lower()104        video_exts = {".mp4", ".mov", ".avi", ".mkv", ".webm"}105 106        if ext in video_exts:107            return self.extract_from_video(file_path)108        else:109            frame, meta = self.load_image(file_path)110            return [frame], meta111