Create exceptions.py

exceptions.py

@@ -0,0 +1,287 @@
+"""
+Custom Exception Classes
+Provides specific exception types for better error handling and debugging
+"""
+
+from typing import Optional, Any, Dict
+
+class VideoProcessingError(Exception):
+    """Base exception for video processing errors"""
+    
+    def __init__(self, message: str, error_code: Optional[str] = None, details: Optional[Dict[str, Any]] = None):
+        super().__init__(message)
+        self.message = message
+        self.error_code = error_code
+        self.details = details or {}
+    
+    def __str__(self):
+        base_msg = self.message
+        if self.error_code:
+            base_msg = f"[{self.error_code}] {base_msg}"
+        if self.details:
+            details_str = ", ".join(f"{k}: {v}" for k, v in self.details.items())
+            base_msg = f"{base_msg} ({details_str})"
+        return base_msg
+
+class ModelLoadingError(VideoProcessingError):
+    """Exception raised when model loading fails"""
+    
+    def __init__(self, model_name: str, message: str, original_error: Optional[Exception] = None):
+        super().__init__(
+            message=f"Failed to load {model_name}: {message}",
+            error_code="MODEL_LOAD_FAILED",
+            details={
+                "model_name": model_name,
+                "original_error": str(original_error) if original_error else None
+            }
+        )
+        self.model_name = model_name
+        self.original_error = original_error
+
+class DeviceError(VideoProcessingError):
+    """Exception raised when device-related operations fail"""
+    
+    def __init__(self, device_type: str, message: str, available_devices: Optional[list] = None):
+        super().__init__(
+            message=f"Device error ({device_type}): {message}",
+            error_code="DEVICE_ERROR",
+            details={
+                "device_type": device_type,
+                "available_devices": available_devices
+            }
+        )
+        self.device_type = device_type
+        self.available_devices = available_devices or []
+
+class MemoryError(VideoProcessingError):
+    """Exception raised when memory operations fail"""
+    
+    def __init__(self, operation: str, message: str, memory_usage: Optional[Dict[str, Any]] = None):
+        super().__init__(
+            message=f"Memory error during {operation}: {message}",
+            error_code="MEMORY_ERROR",
+            details={
+                "operation": operation,
+                "memory_usage": memory_usage
+            }
+        )
+        self.operation = operation
+        self.memory_usage = memory_usage or {}
+
+class VideoFileError(VideoProcessingError):
+    """Exception raised when video file operations fail"""
+    
+    def __init__(self, file_path: str, operation: str, message: str):
+        super().__init__(
+            message=f"Video file error ({operation}): {message}",
+            error_code="VIDEO_FILE_ERROR",
+            details={
+                "file_path": file_path,
+                "operation": operation
+            }
+        )
+        self.file_path = file_path
+        self.operation = operation
+
+class BackgroundProcessingError(VideoProcessingError):
+    """Exception raised when background processing fails"""
+    
+    def __init__(self, background_type: str, message: str, background_path: Optional[str] = None):
+        super().__init__(
+            message=f"Background processing error ({background_type}): {message}",
+            error_code="BACKGROUND_ERROR",
+            details={
+                "background_type": background_type,
+                "background_path": background_path
+            }
+        )
+        self.background_type = background_type
+        self.background_path = background_path
+
+class SegmentationError(VideoProcessingError):
+    """Exception raised when person segmentation fails"""
+    
+    def __init__(self, frame_number: int, message: str, segmentation_method: Optional[str] = None):
+        super().__init__(
+            message=f"Segmentation error at frame {frame_number}: {message}",
+            error_code="SEGMENTATION_ERROR",
+            details={
+                "frame_number": frame_number,
+                "segmentation_method": segmentation_method
+            }
+        )
+        self.frame_number = frame_number
+        self.segmentation_method = segmentation_method
+
+class AudioProcessingError(VideoProcessingError):
+    """Exception raised when audio processing fails"""
+    
+    def __init__(self, operation: str, message: str, input_file: Optional[str] = None, output_file: Optional[str] = None):
+        super().__init__(
+            message=f"Audio processing error ({operation}): {message}",
+            error_code="AUDIO_ERROR",
+            details={
+                "operation": operation,
+                "input_file": input_file,
+                "output_file": output_file
+            }
+        )
+        self.operation = operation
+        self.input_file = input_file
+        self.output_file = output_file
+
+class ConfigurationError(VideoProcessingError):
+    """Exception raised when configuration is invalid"""
+    
+    def __init__(self, config_key: str, value: Any, expected: str):
+        super().__init__(
+            message=f"Invalid configuration: {config_key} = {value}, expected {expected}",
+            error_code="CONFIG_ERROR",
+            details={
+                "config_key": config_key,
+                "value": value,
+                "expected": expected
+            }
+        )
+        self.config_key = config_key
+        self.value = value
+        self.expected = expected
+
+class ProcessingCancelledError(VideoProcessingError):
+    """Exception raised when processing is cancelled by user"""
+    
+    def __init__(self, stage: str, processed_frames: int = 0, total_frames: int = 0):
+        super().__init__(
+            message=f"Processing cancelled at {stage}",
+            error_code="PROCESSING_CANCELLED",
+            details={
+                "stage": stage,
+                "processed_frames": processed_frames,
+                "total_frames": total_frames,
+                "completion_percentage": (processed_frames / total_frames * 100) if total_frames > 0 else 0
+            }
+        )
+        self.stage = stage
+        self.processed_frames = processed_frames
+        self.total_frames = total_frames
+
+class TwoStageProcessingError(VideoProcessingError):
+    """Exception raised during two-stage processing"""
+    
+    def __init__(self, stage: str, message: str, chroma_preset: Optional[str] = None):
+        super().__init__(
+            message=f"Two-stage processing error ({stage}): {message}",
+            error_code="TWO_STAGE_ERROR",
+            details={
+                "stage": stage,
+                "chroma_preset": chroma_preset
+            }
+        )
+        self.stage = stage
+        self.chroma_preset = chroma_preset
+
+class ResourceExhaustionError(VideoProcessingError):
+    """Exception raised when system resources are exhausted"""
+    
+    def __init__(self, resource_type: str, current_usage: float, limit: float, unit: str = ""):
+        super().__init__(
+            message=f"Resource exhaustion: {resource_type} usage {current_usage}{unit} exceeds limit {limit}{unit}",
+            error_code="RESOURCE_EXHAUSTION",
+            details={
+                "resource_type": resource_type,
+                "current_usage": current_usage,
+                "limit": limit,
+                "unit": unit
+            }
+        )
+        self.resource_type = resource_type
+        self.current_usage = current_usage
+        self.limit = limit
+        self.unit = unit
+
+class ValidationError(VideoProcessingError):
+    """Exception raised when input validation fails"""
+    
+    def __init__(self, validation_type: str, message: str, invalid_value: Any = None):
+        super().__init__(
+            message=f"Validation error ({validation_type}): {message}",
+            error_code="VALIDATION_ERROR",
+            details={
+                "validation_type": validation_type,
+                "invalid_value": invalid_value
+            }
+        )
+        self.validation_type = validation_type
+        self.invalid_value = invalid_value
+
+# Utility functions for error handling
+
+def handle_processing_error(func):
+    """Decorator to handle common processing errors"""
+    def wrapper(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except VideoProcessingError:
+            # Re-raise our custom exceptions
+            raise
+        except MemoryError as e:
+            raise MemoryError("memory_operation", str(e)) from e
+        except FileNotFoundError as e:
+            raise VideoFileError(str(e.filename) if e.filename else "unknown", "file_access", str(e)) from e
+        except PermissionError as e:
+            raise VideoFileError(str(e.filename) if e.filename else "unknown", "file_permission", str(e)) from e
+        except Exception as e:
+            raise VideoProcessingError(f"Unexpected error in {func.__name__}: {str(e)}") from e
+    return wrapper
+
+def create_error_context(operation: str, **context) -> Dict[str, Any]:
+    """Create error context dictionary"""
+    return {
+        "operation": operation,
+        "timestamp": __import__('time').time(),
+        **context
+    }
+
+def log_error_with_context(logger, error: VideoProcessingError, additional_context: Optional[Dict[str, Any]] = None):
+    """Log error with full context information"""
+    context = error.details.copy()
+    if additional_context:
+        context.update(additional_context)
+    
+    logger.error(f"{error} | Context: {context}")
+    
+    # Log original error if available
+    if hasattr(error, 'original_error') and error.original_error:
+        logger.error(f"Original error: {error.original_error}")
+
+def is_recoverable_error(error: Exception) -> bool:
+    """Determine if an error is potentially recoverable"""
+    recoverable_errors = (
+        MemoryError,
+        ResourceExhaustionError,
+        DeviceError
+    )
+    
+    # Check if it's a recoverable error type
+    if isinstance(error, recoverable_errors):
+        return True
+    
+    # Check if it's a processing error with specific codes that might be recoverable
+    if isinstance(error, VideoProcessingError):
+        recoverable_codes = ["MEMORY_ERROR", "DEVICE_ERROR", "RESOURCE_EXHAUSTION"]
+        return error.error_code in recoverable_codes
+    
+    return False
+
+def get_error_severity(error: Exception) -> str:
+    """Get error severity level"""
+    if isinstance(error, ProcessingCancelledError):
+        return "INFO"
+    elif isinstance(error, ValidationError):
+        return "WARNING"
+    elif isinstance(error, (ModelLoadingError, DeviceError)):
+        return "CRITICAL"
+    elif isinstance(error, VideoProcessingError):
+        return "ERROR"
+    else:
+        return "UNKNOWN"