Spaces:
Sleeping
Sleeping
| """ | |
| Image processing utilities for KYC POC. | |
| """ | |
| import cv2 | |
| import numpy as np | |
| import base64 | |
| from typing import Optional, Tuple | |
| from fastapi import UploadFile, HTTPException | |
| async def read_image_from_upload(file: UploadFile) -> np.ndarray: | |
| """ | |
| Read uploaded image file into numpy array (OpenCV BGR format). | |
| Args: | |
| file: FastAPI UploadFile object | |
| Returns: | |
| numpy array in BGR format (OpenCV) | |
| Raises: | |
| HTTPException: If image is invalid or cannot be decoded | |
| """ | |
| contents = await file.read() | |
| return decode_image_bytes(contents) | |
| def decode_image_bytes(image_bytes: bytes) -> np.ndarray: | |
| """ | |
| Decode image bytes to numpy array. | |
| Args: | |
| image_bytes: Raw image bytes | |
| Returns: | |
| numpy array in BGR format (OpenCV) | |
| Raises: | |
| HTTPException: If image cannot be decoded | |
| """ | |
| nparr = np.frombuffer(image_bytes, np.uint8) | |
| image = cv2.imdecode(nparr, cv2.IMREAD_COLOR) | |
| if image is None: | |
| raise HTTPException( | |
| status_code=400, | |
| detail={ | |
| "error_code": "IMAGE_INVALID", | |
| "message": "Failed to decode image. Please ensure the file is a valid image." | |
| } | |
| ) | |
| return image | |
| def decode_base64_image(base64_string: str) -> np.ndarray: | |
| """ | |
| Decode base64 encoded image string to numpy array. | |
| Args: | |
| base64_string: Base64 encoded image string (with or without data URI prefix) | |
| Returns: | |
| numpy array in BGR format (OpenCV) | |
| Raises: | |
| HTTPException: If base64 string is invalid or image cannot be decoded | |
| """ | |
| try: | |
| # Remove data URI prefix if present | |
| if "," in base64_string: | |
| base64_string = base64_string.split(",")[1] | |
| # Decode base64 | |
| image_bytes = base64.b64decode(base64_string) | |
| return decode_image_bytes(image_bytes) | |
| except Exception as e: | |
| raise HTTPException( | |
| status_code=400, | |
| detail={ | |
| "error_code": "IMAGE_INVALID", | |
| "message": f"Failed to decode base64 image: {str(e)}" | |
| } | |
| ) | |
| def encode_image_to_base64(image: np.ndarray, format: str = ".jpg") -> str: | |
| """ | |
| Encode numpy array image to base64 string. | |
| Args: | |
| image: numpy array in BGR format | |
| format: Image format (.jpg, .png) | |
| Returns: | |
| Base64 encoded string | |
| """ | |
| _, buffer = cv2.imencode(format, image) | |
| return base64.b64encode(buffer).decode("utf-8") | |
| def resize_image( | |
| image: np.ndarray, | |
| max_size: int = 1024, | |
| keep_aspect: bool = True | |
| ) -> np.ndarray: | |
| """ | |
| Resize image if it exceeds max size. | |
| Args: | |
| image: Input image | |
| max_size: Maximum dimension size | |
| keep_aspect: Whether to keep aspect ratio | |
| Returns: | |
| Resized image | |
| """ | |
| height, width = image.shape[:2] | |
| if max(height, width) <= max_size: | |
| return image | |
| if keep_aspect: | |
| if width > height: | |
| new_width = max_size | |
| new_height = int(height * max_size / width) | |
| else: | |
| new_height = max_size | |
| new_width = int(width * max_size / height) | |
| else: | |
| new_width = max_size | |
| new_height = max_size | |
| return cv2.resize(image, (new_width, new_height), interpolation=cv2.INTER_AREA) | |
| def crop_face_region( | |
| image: np.ndarray, | |
| bbox: Tuple[int, int, int, int], | |
| padding: float = 0.2 | |
| ) -> np.ndarray: | |
| """ | |
| Crop face region from image with padding. | |
| Args: | |
| image: Input image | |
| bbox: Face bounding box (x1, y1, x2, y2) | |
| padding: Padding ratio to add around face | |
| Returns: | |
| Cropped face image | |
| """ | |
| height, width = image.shape[:2] | |
| x1, y1, x2, y2 = bbox | |
| # Calculate padding | |
| face_width = x2 - x1 | |
| face_height = y2 - y1 | |
| pad_x = int(face_width * padding) | |
| pad_y = int(face_height * padding) | |
| # Apply padding with bounds checking | |
| x1 = max(0, x1 - pad_x) | |
| y1 = max(0, y1 - pad_y) | |
| x2 = min(width, x2 + pad_x) | |
| y2 = min(height, y2 + pad_y) | |
| return image[y1:y2, x1:x2] | |
| def validate_image_size(image_bytes: bytes, max_size_bytes: int) -> None: | |
| """ | |
| Validate image size doesn't exceed maximum. | |
| Args: | |
| image_bytes: Image bytes | |
| max_size_bytes: Maximum allowed size in bytes | |
| Raises: | |
| HTTPException: If image exceeds size limit | |
| """ | |
| if len(image_bytes) > max_size_bytes: | |
| max_mb = max_size_bytes / (1024 * 1024) | |
| actual_mb = len(image_bytes) / (1024 * 1024) | |
| raise HTTPException( | |
| status_code=413, | |
| detail={ | |
| "error_code": "IMAGE_TOO_LARGE", | |
| "message": f"Image size ({actual_mb:.2f}MB) exceeds maximum allowed ({max_mb:.2f}MB)" | |
| } | |
| ) | |
| def validate_content_type(content_type: Optional[str], allowed_types: list) -> None: | |
| """ | |
| Validate image content type. | |
| Args: | |
| content_type: MIME type of the file | |
| allowed_types: List of allowed MIME types | |
| Raises: | |
| HTTPException: If content type is not allowed | |
| """ | |
| if content_type not in allowed_types: | |
| raise HTTPException( | |
| status_code=415, | |
| detail={ | |
| "error_code": "UNSUPPORTED_FORMAT", | |
| "message": f"Unsupported image format: {content_type}. Allowed: {allowed_types}" | |
| } | |
| ) | |