Delete pipeline/pipeline_utils.py

pipeline/pipeline_utils.py

@@ -1,191 +0,0 @@
-import os
-import cv2
-import numpy as np
-import tempfile
-from pathlib import Path
-from typing import Optional, Union, Callable
-import logging
-from PIL import Image
-
-# Configure logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-class VideoProcessor:
-    def __init__(self, temp_dir: Optional[str] = None):
-        """
-        Initialize the video processor.
-        
-        Args:
-            temp_dir: Directory for temporary files. If None, creates a temp directory.
-        """
-        self.temp_dir = Path(temp_dir) if temp_dir else Path(tempfile.mkdtemp(prefix="bg_replace_"))
-        self.temp_dir.mkdir(parents=True, exist_ok=True)
-        self.device = self._get_device()
-        logger.info(f"Initialized VideoProcessor with device: {self.device}")
-
-    def _get_device(self) -> str:
-        """Check if CUDA is available."""
-        try:
-            import torch
-            return "cuda" if torch.cuda.is_available() else "cpu"
-        except ImportError:
-            return "cpu"
-
-    def _create_static_bg_video(
-        self,
-        bg_image: np.ndarray,
-        reference_video: str,
-        output_path: str
-    ) -> str:
-        """
-        Create a static background video matching the input video's duration.
-        """
-        cap = cv2.VideoCapture(reference_video)
-        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
-        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
-        fps = cap.get(cv2.CAP_PROP_FPS)
-        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
-        cap.release()
-        
-        # Resize background image
-        bg_image = cv2.resize(bg_image, (width, height))
-        
-        # Write video
-        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
-        out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
-        
-        for _ in range(total_frames):
-            out.write(bg_image)
-        
-        out.release()
-        return output_path
-
-    def _process_with_pipeline(
-        self,
-        input_video: str,
-        background: Optional[Union[str, np.ndarray]] = None,
-        bg_type: str = "blur",
-        progress_callback: Optional[Callable] = None
-    ) -> str:
-        """
-        Process video using the two-stage pipeline.
-        """
-        try:
-            # Import the pipeline
-            from integrated_pipeline import TwoStageProcessor
-            
-            # Update progress
-            if progress_callback:
-                progress_callback(0.1, "Initializing pipeline...")
-            
-            # Handle background
-            bg_video_path = ""
-            if bg_type == "image" and background is not None:
-                if isinstance(background, str):
-                    bg_image = cv2.imread(background)
-                else:
-                    bg_image = background
-                
-                bg_video_path = str(self.temp_dir / "background.mp4")
-                self._create_static_bg_video(bg_image, input_video, bg_video_path)
-            
-            # Initialize processor
-            processor = TwoStageProcessor(temp_dir=str(self.temp_dir))
-            
-            # Process video
-            output_path = str(self.temp_dir / "output.mp4")
-            
-            # Mock click points (center of frame)
-            click_points = [[0.5, 0.5]]
-            
-            # Process
-            success = processor.process_video(
-                input_video=input_video,
-                background_video=bg_video_path if bg_type == "image" else "",
-                click_points=click_points,
-                output_path=output_path,
-                use_matanyone=True,
-                progress_callback=progress_callback
-            )
-            
-            if not success:
-                raise RuntimeError("Video processing failed")
-            
-            return output_path
-            
-        except Exception as e:
-            logger.error(f"Error in pipeline: {str(e)}")
-            raise
-
-    def process_video(
-        self,
-        input_path: Union[str, bytes],
-        background: Optional[Union[str, np.ndarray]] = None,
-        bg_type: str = "blur",
-        progress_callback: Optional[Callable] = None
-    ) -> bytes:
-        """
-        Process a video with the given background.
-        
-        Args:
-            input_path: Path to input video or bytes
-            background: Background image path or numpy array
-            bg_type: Type of background ("image", "color", or "blur")
-            progress_callback: Optional callback for progress updates
-            
-        Returns:
-            Processed video as bytes
-        """
-        try:
-            # Save input to temp file if it's bytes
-            if isinstance(input_path, bytes):
-                input_video = str(self.temp_dir / "input.mp4")
-                with open(input_video, "wb") as f:
-                    f.write(input_path)
-            else:
-                input_video = input_path
-            
-            # Process the video
-            output_path = self._process_with_pipeline(
-                input_video,
-                background,
-                bg_type,
-                progress_callback
-            )
-            
-            # Read the output file
-            with open(output_path, "rb") as f:
-                return f.read()
-                
-        except Exception as e:
-            logger.error(f"Error processing video: {str(e)}")
-            raise
-
-# Global instance
-video_processor = VideoProcessor()
-
-def process_video_pipeline(
-    input_data: Union[str, bytes],
-    background: Optional[Union[str, np.ndarray]] = None,
-    bg_type: str = "blur",
-    progress_callback: Optional[Callable] = None
-) -> bytes:
-    """
-    High-level function to process a video.
-    
-    Args:
-        input_data: Input video path or bytes
-        background: Background image path or numpy array
-        bg_type: Type of background ("image", "color", or "blur")
-        progress_callback: Optional progress callback
-        
-    Returns:
-        Processed video as bytes
-    """
-    return video_processor.process_video(
-        input_data,
-        background,
-        bg_type,
-        progress_callback
-    )