Refactor image processing utilities with improved error handling

Problem

The current implementation of image processing utilities in src/utils/upscaler.py and src/utils/watermark.py has several issues:

1. Inconsistent error handling patterns
2. Duplicated code between modules
3. Limited error recovery mechanisms
4. Poor reporting of image processing failures to the user interface
5. No fallback options when operations fail

Proposed Solution

Refactor both modules to:

1. Create shared base utility classes for image processing
2. Implement consistent error handling with detailed diagnostics
3. Add fallback options for failed operations
4. Create better progress reporting for long-running tasks
5. Optimize image processing for better performance

Implementation Approach

1. Create a base image processor class:

# src/utils/image_processor_base.py
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Optional, Union, List, Dict
from PIL import Image
import logging

logger = logging.getLogger(__name__)

class ImageProcessingError(Exception):
    """Base exception for all image processing errors"""
    pass

class ImageProcessor(ABC):
    """Base class for image processing utilities"""
    
    @abstractmethod
    def process_image(self, image: Image.Image, **kwargs) -> Image.Image:
        """Process a single image"""
        pass
        
    def process_batch(self, 
                     images: List[Union[Image.Image, str, Path]], 
                     **kwargs) -> List[Optional[Image.Image]]:
        """Process multiple images with error handling"""
        results = []
        for img_source in images:
            try:
                # Load image if path is provided
                if isinstance(img_source, (str, Path)):
                    img = self._load_image(img_source)
                else:
                    img = img_source
                
                # Process image
                processed = self.process_image(img, **kwargs)
                results.append(processed)
                
            except Exception as e:
                logger.error(f"Error processing image: {e}")
                results.append(None)
                
        return results
    
    def _load_image(self, path: Union[str, Path]) -> Image.Image:
        """Load an image with proper error handling"""
        try:
            path = Path(path)
            if not path.exists():
                raise ImageProcessingError(f"Image file not found: {path}")
                
            return Image.open(path)
        except Exception as e:
            raise ImageProcessingError(f"Failed to load image {path}: {str(e)}")

2. Refactor Upscaler to extend the base class:

# src/utils/upscaler.py
from src.utils.image_processor_base import ImageProcessor, ImageProcessingError
from PIL import Image

class UpscalerError(ImageProcessingError):
    """Specific exception for upscaling errors"""
    pass

class Upscaler(ImageProcessor):
    def __init__(self):
        super().__init__()
        
    def process_image(self, image: Image.Image, scale_factor=2.0, method=Image.Resampling.LANCZOS) -> Image.Image:
        """Upscale an image with specified parameters"""
        try:
            # Calculate new dimensions
            new_width = int(image.width * scale_factor)
            new_height = int(image.height * scale_factor)
            
            # Perform upscaling
            return image.resize((new_width, new_height), method)
            
        except Exception as e:
            raise UpscalerError(f"Failed to upscale image: {str(e)}")

3. Update Watermarker similarly:

# src/utils/watermark.py
from src.utils.image_processor_base import ImageProcessor, ImageProcessingError
from PIL import Image, ImageDraw, ImageFont

class WatermarkError(ImageProcessingError):
    """Specific exception for watermarking errors"""
    pass

class Watermarker(ImageProcessor):
    def __init__(self, font_path=None):
        super().__init__()
        self.font_path = font_path
        self._initialize_font()
    
    def process_image(self, image: Image.Image, text=None, watermark_image=None, position="bottom-right", opacity=0.3) -> Image.Image:
        """Add watermark to an image"""
        if text:
            return self._add_text_watermark(image, text, position, opacity)
        elif watermark_image:
            return self._add_image_watermark(image, watermark_image, position, opacity)
        else:
            raise WatermarkError("Either text or watermark_image must be provided")

4. Improve error handling in the UI layer:

# In GradioInterface or appropriate controller class:

async def upscale_image_with_error_handling(self, image, scale_factor=2.0):
    """Upscale image with proper error handling and recovery"""
    if image is None:
        return None, "No image provided for upscaling"
        
    try:
        upscaler = Upscaler()
        result = await asyncio.to_thread(
            upscaler.process_image, 
            image, 
            scale_factor=float(scale_factor)
        )
        return result, f"Image upscaled successfully ({scale_factor}x)"
    
    except UpscalerError as e:
        logger.error(f"Upscaling error: {e}")
        # Return original image with error message
        return image, f"Failed to upscale image: {str(e)}"
        
    except Exception as e:
        logger.error(f"Unexpected error during upscaling: {e}", exc_info=True)
        return image, "An unexpected error occurred during upscaling"

Benefits

• More consistent and predictable error handling
• Better user experience with helpful error messages
• Reduced code duplication
• Easier maintenance and extension of image processing utilities
• Improved recovery from errors (returning original image rather than failing completely)