File size: 3,628 Bytes
df4a21a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """
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
|