Spaces:
Runtime error
Runtime error
File size: 3,741 Bytes
b91b7e7 e10bda3 b91b7e7 e10bda3 b91b7e7 e10bda3 b91b7e7 | 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 | from datetime import datetime
import io
import uuid
from PIL import ImageOps, Image
from fastapi import HTTPException, UploadFile
import logging
import requests
from services.embed_utils import send_img_to_embed
ALLOWED_FORMATS = ["JPEG", "PNG", "TIFF", "BMP", "WEBP", "JPG"]
def resize_image(img: Image.Image):
"""
Resizes the image while maintaining the aspect ratio and compresses it.
Converts the image to JPEG for better compression.
Parameters:
img (Image.Image): The PIL Image object to be processed.
filename (str): The original filename of the image, used for logging and metadata.
Returns:
dict: The embedding of the processed image and metadata for further storage or indexing.
"""
quality = 85
img = ImageOps.exif_transpose(img)
max_dimensions = (1024, 1024)
img.thumbnail(max_dimensions, Image.Resampling.LANCZOS)
buffer = io.BytesIO()
img.save(buffer, format="JPEG", optimize=True, quality=quality)
buffer.seek(0)
return img
async def process_image_from_url(url: str):
"""
Fetches an image from a URL, validates the format, and processes it.
Parameters:
url (str): The URL of the image to be fetched and processed.
Returns:
dict: A dictionary containing the processed image's embedding and metadata.
"""
try:
logging.info(f"Fetching image from URL: {url}")
response = requests.get(url)
response.raise_for_status()
img = Image.open(io.BytesIO(response.content))
img_format = img.format.upper()
if img_format not in ALLOWED_FORMATS:
logging.error(f"Unsupported image format from URL: {img_format}")
raise HTTPException(
status_code=400, detail=f"Unsupported image format: {img_format}"
)
return await process_image(img, url)
except Exception as e:
logging.error(f"Error processing image from URL: {str(e)}")
raise HTTPException(status_code=500, detail="Failed to process image from URL.")
async def process_image(img: Image.Image, filename: str):
"""
Converts, processes, resizes, and stores an image.Generates an embedding and stores the image in GCP.
Parameters:
img (Image.Image): The PIL Image object to be processed.
filename (str): The original filename of the image, used for logging and metadata.
Returns:
dict: The embedding of the processed image and metadata for further storage or indexing.
"""
try:
# Generate a unique filename for the image
unique_filename = f"{uuid.uuid4().hex}.jpeg"
logging.info(f"Generated unique filename: {unique_filename}")
# Convert RGBA to RGB if necessary
if img.mode == "RGBA":
logging.info("Image has 4 channels (RGBA), converting to 3 channels (RGB).")
img = img.convert("RGB")
# Resize and compress the image
img = resize_image(img)
buffer = io.BytesIO()
img.save(buffer, format="JPEG", optimize=True)
buffer.seek(0)
# Generate the embedding for the image
logging.info("Sending image to embedding service.")
embedding = await send_img_to_embed(img)
logging.info(f"Received embedding for image: {filename}")
# Get the current date in YYYY-MM-DD format
current_date = datetime.now().strftime("%Y-%m-%d")
# Return the vector for upsert to Pinecone, including the original filename and current date in metadata
return embedding
except Exception as e:
logging.error(f"Error processing image: {str(e)}")
raise HTTPException(status_code=500, detail="Failed to process image.")
|