PS C:\Users\heman\Desktop\Notes> & C:/Users/heman/AppData/Local/Programs/Python/Python312/python.exe c:/Users/heman/Desktop/Notes/video_tools/video_list.py
Assertion fctx->async_lock failed at libavcodec/pthread_frame.c:173
PS C:\Users\heman\Desktop\Notes> & C:/Users/heman/AppData/Local/Programs/Python/Python312/python.exe c:/Users/heman/Desktop/Notes/video_tools/video_list.py
Assertion fctx->async_lock failed at libavcodec/pthread_frame.c:173
PS C:\Users\heman\Desktop\Notes> 

from typing import List, Optional, Union, Tuple
from pathlib import Path
import logging
import cv2
import numpy as np
from PIL import Image
import concurrent.futures
from dataclasses import dataclass
import tempfile
import os
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

@dataclass
class VideoProcessingConfig:
    """Configuration settings for video processing."""
    fps: int = 10
    resize_dimensions: Optional[Tuple[int, int]] = None
    gif_duration: int = 100  # milliseconds
    max_workers: int = 4
    chunk_size: int = 100  # frames per chunk for batch processing
    quality: int = 95  # GIF quality (1-100)

class VideoToGifConverter:
    """Convert video files to sequences of GIF images."""
    
    def __init__(
        self,
        config: VideoProcessingConfig = VideoProcessingConfig()
    ) -> None:
        """Initialize the converter with configuration settings."""
        self.config = config
        self._temp_dir = None
    
    def __enter__(self):
        """Context manager entry."""
        self._temp_dir = tempfile.mkdtemp()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit with cleanup."""
        if self._temp_dir and os.path.exists(self._temp_dir):
            try:
                for file in os.listdir(self._temp_dir):
                    os.remove(os.path.join(self._temp_dir, file))
                os.rmdir(self._temp_dir)
            except Exception as e:
                logger.error(f"Error cleaning up temporary directory: {str(e)}")

    def _validate_video_path(self, video_path: Union[str, Path]) -> Path:
        """Validate the video file path."""
        path = Path(video_path)
        if not path.exists():
            raise FileNotFoundError(f"Video file not found: {path}")
        return path

    def _create_output_directory(self, output_dir: Union[str, Path]) -> Path:
        """Create and validate output directory."""
        path = Path(output_dir)
        path.mkdir(parents=True, exist_ok=True)
        return path

    def _extract_frames(
        self,
        video_capture: cv2.VideoCapture,
        start_frame: int,
        end_frame: int
    ) -> List[np.ndarray]:
        """Extract a chunk of frames from the video."""
        frames = []
        video_capture.set(cv2.CAP_PROP_POS_FRAMES, start_frame)
        
        for _ in range(end_frame - start_frame):
            ret, frame = video_capture.read()
            if not ret:
                break
                
            if self.config.resize_dimensions:
                frame = cv2.resize(
                    frame,
                    self.config.resize_dimensions,
                    interpolation=cv2.INTER_LANCZOS4
                )
            
            # Convert BGR to RGB
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            frames.append(frame)
            
        return frames

    def _save_gif(
        self,
        frames: List[np.ndarray],
        output_path: Path,
        index: int
    ) -> Optional[Path]:
        """Convert frames to GIF and save."""
        try:
            if not frames:
                return None
                
            pil_frames = [Image.fromarray(frame) for frame in frames]
            
            gif_path = output_path / f"frame_{index:06d}.gif"
            pil_frames[0].save(
                gif_path,
                save_all=True,
                append_images=pil_frames[1:],
                duration=self.config.gif_duration,
                loop=0,
                optimize=True,
                quality=self.config.quality
            )
            
            return gif_path
            
        except Exception as e:
            logger.error(f"Error saving GIF {index}: {str(e)}")
            return None

    def process_video(
        self,
        video_path: Union[str, Path],
        output_dir: Union[str, Path]
    ) -> List[Path]:
        """
        Process video file and convert to sequence of GIF images.
        
        Args:
            video_path: Path to input video file
            output_dir: Directory to save output GIF files
            
        Returns:
            List of paths to generated GIF files
        """
        try:
            video_path = self._validate_video_path(video_path)
            output_dir = self._create_output_directory(output_dir)
            
            cap = cv2.VideoCapture(str(video_path))
            if not cap.isOpened():
                raise RuntimeError(f"Failed to open video: {video_path}")
            
            total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
            chunks = range(0, total_frames, self.config.chunk_size)
            
            gif_paths: List[Path] = []
            
            with concurrent.futures.ThreadPoolExecutor(
                max_workers=self.config.max_workers
            ) as executor:
                # Process video in chunks
                future_to_chunk = {
                    executor.submit(
                        self._extract_frames,
                        cap,
                        chunk_start,
                        min(chunk_start + self.config.chunk_size, total_frames)
                    ): i
                    for i, chunk_start in enumerate(chunks)
                }
                
                # Save GIFs as chunks are processed
                for future in concurrent.futures.as_completed(future_to_chunk):
                    chunk_index = future_to_chunk[future]
                    try:
                        frames = future.result()
                        gif_path = self._save_gif(
                            frames,
                            output_dir,
                            chunk_index
                        )
                        if gif_path:
                            gif_paths.append(gif_path)
                    except Exception as e:
                        logger.error(
                            f"Error processing chunk {chunk_index}: {str(e)}"
                        )
            
            cap.release()
            return sorted(gif_paths)
            
        except Exception as e:
            logger.error(f"Error processing video: {str(e)}")
            raise

def main():
    """Example usage of VideoToGifConverter."""
    try:
        # Example configuration
        config = VideoProcessingConfig(
            fps=10,
            resize_dimensions=(480, 270),
            gif_duration=100,
            max_workers=4,
            chunk_size=50,
            quality=95
        )
        
        # Create output directory with timestamp
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_dir = Path(f"output_gifs_{timestamp}")
        
        # Process video
        with VideoToGifConverter(config) as converter:
            video_path = r"C:\Users\heman\Desktop\traffic.mp4"
            gif_paths = converter.process_video(video_path, output_dir)
            
            logger.info(f"Generated {len(gif_paths)} GIF files")
            logger.info(f"Output directory: {output_dir}")
            
    except Exception as e:
        logger.error(f"Main execution error: {str(e)}")
        raise

if __name__ == "__main__":
    main()