""" 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"