core/exceptions.py

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