salmanabjam/deepvision-prompt-builder

1"""2Video Processor Module3 4Handles all video processing operations including frame extraction,5validation, and video metadata extraction.6"""7 8import subprocess9from pathlib import Path10from typing import List, Optional, Union, Tuple11import cv212import magic13from loguru import logger14 15from core.config import config16from core.exceptions import (17    VideoProcessingError,18    InvalidFileError,19    FileSizeError,20    UnsupportedFormatError,21    FrameExtractionError,22)23from core.image_processor import ImageProcessor24 25 26class VideoProcessor:27    """28    Process videos for analysis.29    30    Handles validation, frame extraction, and metadata extraction31    for videos before they are analyzed.32    """33    34    def __init__(self):35        """Initialize VideoProcessor."""36        self.max_size = config.MAX_VIDEO_SIZE37        self.allowed_formats = config.ALLOWED_VIDEO_FORMATS38        self.fps_extraction = config.VIDEO_FPS_EXTRACTION39        self.max_frames = config.MAX_FRAMES_PER_VIDEO40        self.image_processor = ImageProcessor()41        logger.info("VideoProcessor initialized")42    43    def validate_video(self, video_path: Path) -> bool:44        """45        Validate video file.46        47        Args:48            video_path: Path to video file49            50        Returns:51            True if valid52            53        Raises:54            FileSizeError: If file too large55            UnsupportedFormatError: If format not supported56            InvalidFileError: If file is corrupted57        """58        # Check file exists59        if not video_path.exists():60            raise InvalidFileError(61                f"Video file not found: {video_path}",62                {"path": str(video_path)}63            )64        65        # Check file size66        file_size = video_path.stat().st_size67        if file_size > self.max_size:68            raise FileSizeError(69                f"Video too large: {file_size / 1024 / 1024:.1f}MB",70                {"max_size": self.max_size, "actual_size": file_size}71            )72        73        # Check file extension74        ext = video_path.suffix.lower()75        if ext not in self.allowed_formats:76            raise UnsupportedFormatError(77                f"Unsupported video format: {ext}",78                {"allowed": self.allowed_formats, "received": ext}79            )80        81        # Check MIME type using magic bytes82        try:83            mime = magic.from_file(str(video_path), mime=True)84            if not mime.startswith("video/"):85                raise InvalidFileError(86                    f"File is not a valid video: {mime}",87                    {"mime_type": mime}88                )89        except Exception as e:90            logger.warning(f"Could not verify MIME type: {e}")91        92        return True93    94    def get_video_info(self, video_path: Union[str, Path]) -> dict:95        """96        Get video metadata using OpenCV.97        98        Args:99            video_path: Path to video file100            101        Returns:102            Dictionary with video information103        """104        video_path = Path(video_path)105        self.validate_video(video_path)106        107        try:108            cap = cv2.VideoCapture(str(video_path))109            110            if not cap.isOpened():111                raise InvalidFileError(112                    "Cannot open video file",113                    {"path": str(video_path)}114                )115            116            # Extract metadata117            fps = cap.get(cv2.CAP_PROP_FPS)118            frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))119            width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))120            height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))121            duration = frame_count / fps if fps > 0 else 0122            123            cap.release()124            125            info = {126                "filename": video_path.name,127                "fps": fps,128                "frame_count": frame_count,129                "width": width,130                "height": height,131                "duration": duration,132                "file_size": video_path.stat().st_size,133            }134            135            logger.info(f"Video info: {video_path.name} - {width}x{height}, "136                       f"{fps:.2f}fps, {duration:.2f}s")137            138            return info139            140        except Exception as e:141            logger.error(f"Failed to get video info: {e}")142            raise VideoProcessingError(143                f"Cannot extract video metadata: {str(e)}",144                {"path": str(video_path), "error": str(e)}145            )146    147    def extract_frames(148        self,149        video_path: Union[str, Path],150        fps: Optional[float] = None,151        max_frames: Optional[int] = None,152        output_dir: Optional[Path] = None153    ) -> List[Path]:154        """155        Extract frames from video at specified FPS.156        157        Args:158            video_path: Path to video file159            fps: Frames per second to extract (default: config.VIDEO_FPS_EXTRACTION)160            max_frames: Maximum number of frames to extract161            output_dir: Directory to save frames (default: cache directory)162            163        Returns:164            List of paths to extracted frames165            166        Raises:167            FrameExtractionError: If frame extraction fails168        """169        video_path = Path(video_path)170        self.validate_video(video_path)171        172        if fps is None:173            fps = self.fps_extraction174        175        if max_frames is None:176            max_frames = self.max_frames177        178        if output_dir is None:179            output_dir = config.CACHE_DIR / "frames" / video_path.stem180        181        output_dir.mkdir(parents=True, exist_ok=True)182        183        try:184            cap = cv2.VideoCapture(str(video_path))185            186            if not cap.isOpened():187                raise FrameExtractionError(188                    "Cannot open video file",189                    {"path": str(video_path)}190                )191            192            video_fps = cap.get(cv2.CAP_PROP_FPS)193            frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))194            195            # Calculate frame interval196            frame_interval = int(video_fps / fps) if fps < video_fps else 1197            198            frames_saved = []199            frame_idx = 0200            saved_count = 0201            202            logger.info(f"Extracting frames from {video_path.name} "203                       f"(fps={fps}, interval={frame_interval})")204            205            while True:206                ret, frame = cap.read()207                208                if not ret:209                    break210                211                # Extract frame at specified interval212                if frame_idx % frame_interval == 0:213                    # Save frame214                    frame_path = output_dir / f"frame_{saved_count:04d}.jpg"215                    cv2.imwrite(str(frame_path), frame)216                    frames_saved.append(frame_path)217                    saved_count += 1218                    219                    # Check if we've reached max frames220                    if saved_count >= max_frames:221                        logger.info(f"Reached max frames limit: {max_frames}")222                        break223                224                frame_idx += 1225            226            cap.release()227            228            logger.info(f"Extracted {len(frames_saved)} frames from {video_path.name}")229            230            return frames_saved231            232        except Exception as e:233            logger.error(f"Frame extraction failed: {e}")234            raise FrameExtractionError(235                f"Failed to extract frames: {str(e)}",236                {"path": str(video_path), "error": str(e)}237            )238    239    def extract_key_frames(240        self,241        video_path: Union[str, Path],242        num_frames: int = 5,243        output_dir: Optional[Path] = None244    ) -> List[Path]:245        """246        Extract evenly distributed key frames from video.247        248        Args:249            video_path: Path to video file250            num_frames: Number of key frames to extract251            output_dir: Directory to save frames252            253        Returns:254            List of paths to extracted frames255        """256        video_path = Path(video_path)257        self.validate_video(video_path)258        259        if output_dir is None:260            output_dir = config.CACHE_DIR / "keyframes" / video_path.stem261        262        output_dir.mkdir(parents=True, exist_ok=True)263        264        try:265            cap = cv2.VideoCapture(str(video_path))266            267            if not cap.isOpened():268                raise FrameExtractionError(269                    "Cannot open video file",270                    {"path": str(video_path)}271                )272            273            frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))274            275            # Calculate frame positions276            positions = [int(i * frame_count / (num_frames + 1)) 277                        for i in range(1, num_frames + 1)]278            279            frames_saved = []280            281            for idx, pos in enumerate(positions):282                cap.set(cv2.CAP_PROP_POS_FRAMES, pos)283                ret, frame = cap.read()284                285                if ret:286                    frame_path = output_dir / f"keyframe_{idx:02d}.jpg"287                    cv2.imwrite(str(frame_path), frame)288                    frames_saved.append(frame_path)289            290            cap.release()291            292            logger.info(f"Extracted {len(frames_saved)} key frames from {video_path.name}")293            294            return frames_saved295            296        except Exception as e:297            logger.error(f"Key frame extraction failed: {e}")298            raise FrameExtractionError(299                f"Failed to extract key frames: {str(e)}",300                {"path": str(video_path), "error": str(e)}301            )302    303    def process(304        self,305        video_path: Union[str, Path],306        extract_method: str = "fps",307        **kwargs308    ) -> List[Path]:309        """310        Complete video processing pipeline.311        312        Args:313            video_path: Path to video file314            extract_method: Method for frame extraction ("fps" or "keyframes")315            **kwargs: Additional arguments for extraction method316            317        Returns:318            List of extracted frame paths319        """320        try:321            if extract_method == "fps":322                return self.extract_frames(video_path, **kwargs)323            elif extract_method == "keyframes":324                return self.extract_key_frames(video_path, **kwargs)325            else:326                raise ValueError(f"Unknown extraction method: {extract_method}")327                328        except Exception as e:329            logger.error(f"Video processing failed: {e}")330            raise VideoProcessingError(331                f"Failed to process video: {str(e)}",332                {"path": str(video_path), "error": str(e)}333            )334