Spaces:
Runtime error
Runtime error
| """ | |
| Image preprocessing service. | |
| """ | |
| from typing import Optional, Dict, Any | |
| from PIL import Image | |
| from app.core.errors import ImageProcessingError | |
| from app.core.logging import get_logger | |
| from app.utils.image import load_image_from_bytes, validate_image_bytes | |
| from app.utils.security import validate_file_size, validate_image_content, MAX_FILE_SIZE_BYTES | |
| logger = get_logger(__name__) | |
| class PreprocessService: | |
| """ | |
| Service for preprocessing images before model inference. | |
| For Milestone 1, this is minimal - just validates and optionally | |
| decodes images. Future milestones will add more preprocessing. | |
| """ | |
| def __init__(self, max_file_size: int = MAX_FILE_SIZE_BYTES): | |
| """ | |
| Initialize the preprocess service. | |
| Args: | |
| max_file_size: Maximum allowed file size in bytes | |
| """ | |
| self.max_file_size = max_file_size | |
| def validate_image(self, image_bytes: bytes) -> Dict[str, Any]: | |
| """ | |
| Validate uploaded image bytes. | |
| Args: | |
| image_bytes: Raw image bytes | |
| Returns: | |
| Dictionary with validation results | |
| Raises: | |
| ImageProcessingError: If validation fails | |
| """ | |
| # Check file size | |
| if not validate_file_size(image_bytes, self.max_file_size): | |
| raise ImageProcessingError( | |
| message=f"File too large. Maximum size is {self.max_file_size // (1024*1024)}MB", | |
| details={"size": len(image_bytes), "max_size": self.max_file_size} | |
| ) | |
| # Check content type via magic bytes | |
| if not validate_image_content(image_bytes): | |
| raise ImageProcessingError( | |
| message="Invalid image format. Supported formats: JPEG, PNG, GIF, WebP, BMP", | |
| details={"size": len(image_bytes)} | |
| ) | |
| return { | |
| "valid": True, | |
| "size_bytes": len(image_bytes) | |
| } | |
| def decode_image(self, image_bytes: bytes) -> Image.Image: | |
| """ | |
| Decode image bytes to PIL Image. | |
| Args: | |
| image_bytes: Raw image bytes | |
| Returns: | |
| PIL Image object | |
| Raises: | |
| ImageProcessingError: If decoding fails | |
| """ | |
| return load_image_from_bytes(image_bytes) | |
| def preprocess( | |
| self, | |
| image_bytes: bytes, | |
| decode: bool = False | |
| ) -> Dict[str, Any]: | |
| """ | |
| Full preprocessing pipeline. | |
| Args: | |
| image_bytes: Raw image bytes | |
| decode: Whether to decode to PIL Image | |
| Returns: | |
| Dictionary with: | |
| - image_bytes: Original or processed bytes | |
| - image: PIL Image if decode=True | |
| - validation: Validation results | |
| """ | |
| # Validate | |
| validation = self.validate_image(image_bytes) | |
| result = { | |
| "image_bytes": image_bytes, | |
| "validation": validation | |
| } | |
| # Optionally decode | |
| if decode: | |
| result["image"] = self.decode_image(image_bytes) | |
| return result | |
| # Global singleton instance | |
| _preprocess_service: Optional[PreprocessService] = None | |
| def get_preprocess_service() -> PreprocessService: | |
| """ | |
| Get the global preprocess service instance. | |
| Returns: | |
| PreprocessService instance | |
| """ | |
| global _preprocess_service | |
| if _preprocess_service is None: | |
| _preprocess_service = PreprocessService() | |
| return _preprocess_service | |