Spaces:
Sleeping
Sleeping
File size: 3,493 Bytes
0116945 |
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 |
import uuid
import requests
from PIL import Image
import numpy as np
import gradio as gr
from encoder import FashionCLIPEncoder
# Constants
REQUESTS_HEADERS = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
BATCH_SIZE = 30 # Define batch size for processing
# Initialize encoder
encoder = FashionCLIPEncoder()
# Helper function to download images
def download_image_as_pil(url: str, timeout: int = 10) -> Image.Image:
try:
response = requests.get(url, stream=True, headers=REQUESTS_HEADERS, timeout=timeout)
if response.status_code == 200:
return Image.open(response.raw).convert("RGB") # Ensure consistent format
return None
except Exception as e:
print(f"Error downloading image: {e}")
return None
# Embedding function for a batch of images
def batch_process_images(image_urls: str):
# Split the input string by commas and strip whitespace
urls = [url.strip() for url in image_urls.split(",") if url.strip()]
if not urls:
return {"error": "No valid image URLs provided."}
results = []
batch_urls, batch_images = [], []
for url in urls:
try:
# Download image
image = download_image_as_pil(url)
if not image:
results.append({"image_url": url, "error": "Failed to download image"})
continue
batch_urls.append(url)
batch_images.append(image)
# Process batch when reaching batch size
if len(batch_images) == BATCH_SIZE:
process_batch(batch_urls, batch_images, results)
batch_urls, batch_images = [], []
except Exception as e:
results.append({"image_url": url, "error": str(e)})
# Process remaining images in the last batch
if batch_images:
process_batch(batch_urls, batch_images, results)
return results
# Helper function to process a batch
def process_batch(batch_urls, batch_images, results):
try:
# Generate embeddings
embeddings = encoder.encode_images(batch_images)
for url, embedding in zip(batch_urls, embeddings):
# Normalize embedding
embedding_normalized = embedding / np.linalg.norm(embedding)
# Append results
results.append({
"image_url": url,
"embedding_preview": embedding_normalized[:5].tolist(), # First 5 values for preview
"success": True
})
except Exception as e:
for url in batch_urls:
results.append({"image_url": url, "error": str(e)})
# Gradio Interface
iface = gr.Interface(
fn=batch_process_images,
inputs=gr.Textbox(
lines=5,
placeholder="Enter image URLs separated by commas",
label="Batch Image URLs",
),
outputs=gr.JSON(label="Embedding Results"),
title="Batch Fashion CLIP Embedding API",
description="Enter multiple image URLs (separated by commas) to generate embeddings for the batch. Each embedding preview includes the first 5 values.",
examples=[
["https://cdn.shopify.com/s/files/1/0522/2239/4534/files/CT21355-22_1024x1024.webp, https://cdn.shopify.com/s/files/1/0522/2239/4534/files/00907857-C6B0-4D2A-8AEA-688BDE1E67D7_1024x1024.jpg"]
],
)
# Launch Gradio App
if __name__ == "__main__":
iface.launch() |