diff --git a/examples/media-gen/.gitattributes b/examples/media-gen/.gitattributes new file mode 100644 index 0000000..f8ff2b5 --- /dev/null +++ b/examples/media-gen/.gitattributes @@ -0,0 +1 @@ +*.mp4 filter=lfs diff=lfs merge=lfs -text diff --git a/examples/media-gen/integration_tests/test_video.mp4 b/examples/media-gen/integration_tests/test_video.mp4 new file mode 100644 index 0000000..9502e09 --- /dev/null +++ b/examples/media-gen/integration_tests/test_video.mp4 @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:ac1bb2cde2a13877b15d6a01bba65e4c0b835c9692e80f62478f5df7a95a0438 +size 6988797 diff --git a/examples/media-gen/integration_tests/test_video_utils.py b/examples/media-gen/integration_tests/test_video_utils.py new file mode 100644 index 0000000..2daa8a8 --- /dev/null +++ b/examples/media-gen/integration_tests/test_video_utils.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +""" +Integration test for video screenshot extraction utilities. + +This script demonstrates video screenshot extraction functionality using the test video. +It extracts screenshots every 10 seconds and saves them to ~/Downloads. + +Usage: + python integration_tests/test_video_utils.py + +Requirements: +- OpenCV (opencv-python) +- Test video file: integration_tests/test_video.mp4 +""" + +import os +import sys + +from pathlib import Path + +# Add parent directory to path for imports +sys.path.append(os.path.join(os.path.dirname(__file__), '..')) + +from utils.video_utils import VideoScreenshotExtractor, extract_screenshots + + +def main(): + """Extract screenshots from the test video every 10 seconds.""" + print("šŸŽ¬ Video Screenshot Extraction Test") + print("=" * 50) + + # Path to test video + test_video_path = Path(__file__).parent / "test_video.mp4" + + if not test_video_path.exists(): + print(f"āŒ Test video not found at: {test_video_path}") + print("Please ensure test_video.mp4 exists in the integration_tests directory") + return + + print(f"āœ… Test video found: {test_video_path}") + print(f"šŸ“ File size: {test_video_path.stat().st_size:,} bytes") + + # Get video properties + with VideoScreenshotExtractor(str(test_video_path)) as extractor: + print(f"šŸŽ„ Video properties:") + print(f" - FPS: {extractor.fps:.2f}") + print(f" - Duration: {extractor.duration:.2f} seconds") + print(f" - Frame count: {extractor.frame_count:,}") + print() + + # Extract screenshots every 5 seconds, starting from 1 second + print("šŸ”„ Extracting screenshots every 5 seconds (starting from 1s)...") + print("-" * 50) + + try: + screenshots = extract_screenshots( + video_path=str(test_video_path), + interval_seconds=5.0, + start_time=1.0, # Start from 1 second instead of 0 + output_dir="~/Downloads/polymind_video_screenshots", + filename_prefix="test_video_5s" + ) + + print(f"āœ… Extracted {len(screenshots)} screenshots") + for i, ss in enumerate(screenshots): + print(f" {i+1}. Frame {ss.frame_number} at {ss.timestamp_str}") + print() + + except Exception as e: + print(f"āŒ Screenshot extraction failed: {e}") + return + + # Summary + print("šŸ“Š Summary") + print("=" * 50) + downloads_dir = Path.home() / "Downloads" / "polymind_video_screenshots" + + if downloads_dir.exists(): + screenshot_files = list(downloads_dir.glob("test_video_10s_*.jpg")) + print(f"āœ… Screenshots saved to: {downloads_dir}") + print(f"šŸ“ Files created: {len(screenshot_files)}") + + if screenshot_files: + print(f"šŸ“„ File pattern: test_video_10s_*.jpg") + print(f"šŸ“ Sample file size: {screenshot_files[0].stat().st_size:,} bytes") + else: + print("āŒ No screenshots were created") + + print("\nšŸŽÆ Test completed!") + print("šŸ’” You can now view the extracted screenshots in your Downloads folder") + + +if __name__ == "__main__": + main() diff --git a/examples/media-gen/requirements.txt b/examples/media-gen/requirements.txt index 2d0a496..37f7fa9 100644 --- a/examples/media-gen/requirements.txt +++ b/examples/media-gen/requirements.txt @@ -15,4 +15,8 @@ python-dotenv # For environment variable management anthropic # For Claude integration openai # For OpenAI integration replicate # For Replicate integration -requests # For downloading images from URLs \ No newline at end of file +requests # For downloading images from URLs + +# Video processing dependencies +opencv-python==4.8.1.78 # For video screenshot extraction (compatible with numpy 1.26.0) +numpy==1.26.0 # Required by polymind and compatible with opencv-python 4.8.1.78 diff --git a/examples/media-gen/utils/__init__.py b/examples/media-gen/utils/__init__.py new file mode 100644 index 0000000..fe1a093 --- /dev/null +++ b/examples/media-gen/utils/__init__.py @@ -0,0 +1,10 @@ +""" +Utility functions for media generation and processing. + +This package contains helper functions for video processing, image manipulation, +and other media-related utilities. +""" + +from .video_utils import extract_screenshots, VideoScreenshotExtractor + +__all__ = ['extract_screenshots', 'VideoScreenshotExtractor'] \ No newline at end of file diff --git a/examples/media-gen/utils/video_utils.py b/examples/media-gen/utils/video_utils.py new file mode 100644 index 0000000..ec9e7e0 --- /dev/null +++ b/examples/media-gen/utils/video_utils.py @@ -0,0 +1,188 @@ +""" +Simple video screenshot utility. + +Extract screenshots from video files at regular time intervals. +""" + +import logging +import os +from dataclasses import dataclass +from datetime import timedelta +from typing import List, Optional + +import cv2 +from pathlib import Path + +logger = logging.getLogger(__name__) + + +def _expand_path(path: str) -> Path: + """ + Expand a path, handling tilde (~) for home directory. + + Args: + path: Path string that may contain tilde + + Returns: + Expanded Path object + """ + return Path(os.path.expanduser(path)) + + +@dataclass +class ScreenshotInfo: + """Information about an extracted screenshot.""" + frame_number: int + timestamp: float # seconds + timestamp_str: str # formatted as HH:MM:SS + file_path: str + + +class VideoScreenshotExtractor: + """ + Extract screenshots from video files at regular intervals. + + Simple utility to take screenshots every X seconds from a video file. + """ + + def __init__(self, video_path: str): + """ + Initialize the screenshot extractor. + + Args: + video_path: Path to the video file + """ + self.video_path = Path(video_path) + if not self.video_path.exists(): + raise FileNotFoundError(f"Video file not found: {video_path}") + + self.cap = cv2.VideoCapture(str(self.video_path)) + if not self.cap.isOpened(): + raise ValueError(f"Could not open video file: {video_path}") + + # Get video properties + self.fps = self.cap.get(cv2.CAP_PROP_FPS) + self.frame_count = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT)) + self.duration = self.frame_count / self.fps if self.fps > 0 else 0 + + logger.info( + f"Video loaded: {self.fps:.2f} FPS, " + f"{self.frame_count} frames, " + f"{self.duration:.2f}s duration" + ) + + def __enter__(self): + """Context manager entry.""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Context manager exit - release video capture.""" + self.release() + + def release(self) -> None: + """Release the video capture object.""" + if self.cap: + self.cap.release() + + def _format_timestamp(self, seconds: float) -> str: + """Format seconds into HH:MM:SS string.""" + return str(timedelta(seconds=int(seconds))) + + def extract_screenshots( + self, + interval_seconds: float = 2.0, + start_time: float = 0.0, + output_dir: Optional[str] = "~/Downloads", + filename_prefix: str = "screenshot" + ) -> List[ScreenshotInfo]: + """ + Extract screenshots at regular time intervals. + + Args: + interval_seconds: Time interval between screenshots (default: 2.0) + start_time: Start time in seconds (default: 0.0) + output_dir: Directory to save screenshots (default: ~/Downloads) + filename_prefix: Prefix for saved files + + Returns: + List of ScreenshotInfo objects + """ + if interval_seconds <= 0: + raise ValueError("Interval must be positive") + if start_time < 0: + raise ValueError("Start time must be non-negative") + + screenshots = [] + frame_interval = int(self.fps * interval_seconds) + start_frame = int(start_time * self.fps) + + # Handle output directory with tilde expansion + output_path = _expand_path(output_dir) if output_dir else _expand_path("~/Downloads") + output_path.mkdir(parents=True, exist_ok=True) + + for frame_num in range(start_frame, self.frame_count, frame_interval): + self.cap.set(cv2.CAP_PROP_POS_FRAMES, frame_num) + ret, frame = self.cap.read() + + if not ret: + break + + timestamp = frame_num / self.fps + timestamp_str = self._format_timestamp(timestamp) + + screenshot_info = ScreenshotInfo( + frame_number=frame_num, + timestamp=timestamp, + timestamp_str=timestamp_str, + file_path="" + ) + + # Always save screenshots since we have a default output directory + # Use timestamp-based filename for chronological sorting + timestamp_seconds = int(timestamp) + filename = (f"{filename_prefix}_{timestamp_seconds:06d}s_" + f"{timestamp_str.replace(':', '-')}.jpg") + file_path = output_path / filename + cv2.imwrite(str(file_path), frame) + screenshot_info.file_path = str(file_path) + + screenshots.append(screenshot_info) + logger.debug( + f"Extracted screenshot at {timestamp_str} " + f"(frame {frame_num})" + ) + + logger.info( + f"Extracted {len(screenshots)} screenshots " + f"at {interval_seconds}s intervals" + ) + return screenshots + + +def extract_screenshots( + video_path: str, + interval_seconds: float = 2.0, + start_time: float = 0.0, + output_dir: Optional[str] = "~/Downloads", + filename_prefix: str = "screenshot" +) -> List[ScreenshotInfo]: + """ + Convenience function to extract screenshots from a video file. + + Args: + video_path: Path to the video file + interval_seconds: Time interval between screenshots (default: 2.0) + start_time: Start time in seconds (default: 0.0) + output_dir: Directory to save screenshots (default: ~/Downloads) + filename_prefix: Prefix for saved files + + Returns: + List of ScreenshotInfo objects + """ + with VideoScreenshotExtractor(video_path) as extractor: + return extractor.extract_screenshots( + interval_seconds=interval_seconds, + start_time=start_time, + output_dir=output_dir, + filename_prefix=filename_prefix + )