import io import ast import json import base64 import spaces import requests import numpy as np import gradio as gr from PIL import Image from io import BytesIO import face_recognition from turtle import title from openai import OpenAI from collections import Counter from transformers import pipeline import urllib.request from transformers import YolosImageProcessor, YolosForObjectDetection import torch import matplotlib.pyplot as plt from torchvision.transforms import ToTensor, ToPILImage client = OpenAI() pipe = pipeline("zero-shot-image-classification", model="patrickjohncyh/fashion-clip") color_file_path = 'color_config.json' attributes_file_path = 'attributes_config.json' import os OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") # Open and read the COLOR JSON file with open(color_file_path, 'r') as file: color_data = json.load(file) # Open and read the ATTRIBUTES JSON file with open(attributes_file_path, 'r') as file: attributes_data = json.load(file) COLOURS_DICT = color_data['color_mapping'] ATTRIBUTES_DICT = attributes_data['attribute_mapping'] def shot(input, category, level): output_dict = {} if level == 'variant': subColour, mainColour, score = get_colour(ast.literal_eval(str(input)), category) openai_parsed_response = get_openAI_tags(ast.literal_eval(str(input))) face_embeddings = get_face_embeddings(ast.literal_eval(str(input))) cropped_images = get_cropped_images(ast.literal_eval(str(input)), category) # Ensure all outputs are JSON serializable output_dict['colors'] = { "main": mainColour, "sub": subColour, "score": score } output_dict['image_mapping'] = openai_parsed_response output_dict['face_embeddings'] = face_embeddings output_dict['cropped_images'] = cropped_images if level == 'product': common_result = get_predicted_attributes(ast.literal_eval(str(input)), category) output_dict['attributes'] = common_result output_dict['subcategory'] = category # # Convert the dictionary to a JSON-serializable format # try: # serialized_output = json.dumps(output_dict) # except TypeError as e: # print(f"Serialization Error: {e}") # return {"error": "Serialization failed"} return json.dumps(output_dict) # @spaces.GPU # def get_colour(image_urls, category): # colourLabels = list(COLOURS_DICT.keys()) # for i in range(len(colourLabels)): # colourLabels[i] = colourLabels[i] + " clothing: " + category # responses = pipe(image_urls, candidate_labels=colourLabels) # # Get the most common colour # mainColour = responses[0][0]['label'].split(" clothing:")[0] # if mainColour not in COLOURS_DICT: # return None, None, None # # Add category to the end of each label # labels = COLOURS_DICT[mainColour] # for i in range(len(labels)): # labels[i] = labels[i] + " clothing: " + category # # Run pipeline in one go # responses = pipe(image_urls, candidate_labels=labels) # subColour = responses[0][0]['label'].split(" clothing:")[0] # return subColour, mainColour, responses[0][0]['score'] @spaces.GPU def get_colour(image_urls, category): # Prepare color labels colourLabels = [f"{color} clothing: {category}" for color in COLOURS_DICT.keys()] print("Colour Labels:", colourLabels) # Debug: Print colour labels print("Image URLs:", image_urls) # Debug: Print image URLs # Split labels into two batches mid_index = len(colourLabels) // 2 first_batch = colourLabels[:mid_index] second_batch = colourLabels[mid_index:] # Process the first batch responses_first_batch = pipe(image_urls, candidate_labels=first_batch) # Get the top 3 from the first batch top3_first_batch = sorted(responses_first_batch[0], key=lambda x: x['score'], reverse=True)[:3] # Process the second batch responses_second_batch = pipe(image_urls, candidate_labels=second_batch) # Get the top 3 from the second batch top3_second_batch = sorted(responses_second_batch[0], key=lambda x: x['score'], reverse=True)[:3] # Combine the top 3 from each batch combined_top6 = top3_first_batch + top3_second_batch # Get the final top 3 from the combined list final_top3 = sorted(combined_top6, key=lambda x: x['score'], reverse=True)[:3] mainColour = final_top3[0]['label'].split(" clothing:")[0] if mainColour not in COLOURS_DICT: return None, None, None # Get sub-colors for the main color labels = [f"{label} clothing: {category}" for label in COLOURS_DICT[mainColour]] print("Labels for pipe:", labels) # Debug: Confirm labels are correct responses = pipe(image_urls, candidate_labels=labels) subColour = responses[0][0]['label'].split(" clothing:")[0] return subColour, mainColour, responses[0][0]['score'] @spaces.GPU def get_predicted_attributes(image_urls, category): # Assuming ATTRIBUTES_DICT and pipe are defined outside this function attributes = list(ATTRIBUTES_DICT.get(category, {}).keys()) # Mapping of possible values per attribute common_result = [] for attribute in attributes: values = ATTRIBUTES_DICT.get(category, {}).get(attribute, []) if len(values) == 0: continue # Adjust labels for the pipeline to be in format: "{attr}: {value}, clothing: {category}" attribute_formatted = attribute.replace("colartype", "collar").replace("sleevelength", "sleeve length").replace("fabricstyle", "fabric") values_formatted = [f"{attribute_formatted}: {value}, clothing: {category}" for value in values] # Get the predicted values for the attribute responses = pipe(image_urls, candidate_labels=values_formatted) result = [response[0]['label'].split(", clothing:")[0] for response in responses] # If attribute is details, then get the top 2 most common labels if attribute_formatted == "details": result += [response[1]['label'].split(", clothing:")[0] for response in responses] common_result.append(Counter(result).most_common(2)) else: common_result.append(Counter(result).most_common(1)) # Clean up the results into one long string for i, result in enumerate(common_result): common_result[i] = ", ".join([f"{x[0]}" for x in result]) result = {} # Iterate through the list and split each item into key and value for item in common_result: # Split by ': ' to separate the key and value key, value = item.split(': ', 1) if key == "details": details_split = value.split(" , ") if len(details_split) == 2: result["details_1"] = details_split[0] result["details_2"] = details_split[1] else: result["details_1"] = value # If there's only one detail, assign it to details 1 else: result[key] = value return result def get_openAI_tags(image_urls): # Create list containing JSONs of each image URL imageList = [] for image in image_urls: imageList.append({"type": "image_url", "image_url": {"url": image}}) openai_response = client.chat.completions.create( model="gpt-4o", messages=[ { "role": "system", "content": [ { "type": "text", "text": "You're a tagging assistant, you will help label and tag product pictures for my online e-commerce platform. Your tasks will be to return which angle the product images were taken from. You will have to choose from 'full-body', 'half-body', 'side', 'back', or 'zoomed' angles. You should label each of the images with one of these labels depending on which you think fits best (ideally, every label should be used at least once, but only if there are 5 or more images), and should respond with an unformatted dictionary where the key is a string representation of the url index of the url and the value is the assigned label." } ] }, { "role": "user", "content": imageList }, ], temperature=1, max_tokens=500, top_p=1, frequency_penalty=0, presence_penalty=0 ) response = json.loads(openai_response.choices[0].message.content) return response @spaces.GPU def get_face_embeddings(image_urls): # Initialize a dictionary to store the face encodings or errors results = {} # Loop through each image URL for index, url in enumerate(image_urls): try: # Try to download the image from the URL response = requests.get(url) # Raise an exception if the response is not successful response.raise_for_status() # Load the image using face_recognition image = face_recognition.load_image_file(BytesIO(response.content)) # Get the face encodings for all faces in the image face_encodings = face_recognition.face_encodings(image) # If no faces are detected, store an empty list if not face_encodings: results[str(index)] = [] else: # Otherwise, store the first face encoding as a list results[str(index)] = face_encodings[0].tolist() except Exception as e: # If any error occurs during the download or processing, store the error message results[str(index)] = f"Error processing image: {str(e)}" return results # new ACCURACY_THRESHOLD = 0.86 def open_image_from_url(url): # Fetch the image from the URL response = requests.get(url, stream=True) response.raise_for_status() # Check if the request was successful # Open the image using PIL image = Image.open(BytesIO(response.content)) return image # Add the main data to the session state main = [['Product Id', 'Sku', 'Color', 'Images', 'Status', 'Category', 'Text']] # This is the order of the categories list. NO NOT CHANGE. Just for visualization purposes cats = ['shirt, blouse', 'top, t-shirt, sweatshirt', 'sweater', 'cardigan', 'jacket', 'vest', 'pants', 'shorts', 'skirt', 'coat', 'dress', 'jumpsuit', 'cape', 'glasses', 'hat', 'headband, head covering, hair accessory', 'tie', 'glove', 'watch', 'belt', 'leg warmer', 'tights, stockings', 'sock', 'shoe', 'bag, wallet', 'scarf', 'umbrella', 'hood', 'collar', 'lapel', 'epaulette', 'sleeve', 'pocket', 'neckline', 'buckle', 'zipper', 'applique', 'bead', 'bow', 'flower', 'fringe', 'ribbon', 'rivet', 'ruffle', 'sequin', 'tassel'] filter = ['dress', 'jumpsuit', 'cape', 'glasses', 'hat', 'headband, head covering, hair accessory', 'tie', 'glove', 'watch', 'belt', 'leg warmer', 'tights, stockings', 'sock', 'shoe', 'scarf', 'umbrella', 'hood', 'collar', 'lapel', 'epaulette', 'sleeve', 'pocket', 'neckline', 'buckle', 'zipper', 'applique', 'bead', 'bow', 'flower', 'fringe', 'ribbon', 'rivet', 'ruffle', 'sequin', 'tassel'] # 0 for full body, 1 for upper body, 2 for lower body, 3 for over body (jacket, coat, etc), 4 for accessories yolo_mapping = { 'shirt, blouse': 3, 'top, t-shirt, sweatshirt' : 1, 'sweater': 1, 'cardigan': 1, 'jacket': 3, 'vest': 1, 'pants': 2, 'shorts': 2, 'skirt': 2, 'coat': 3, 'dress': 0, 'jumpsuit': 0, 'bag, wallet': 4 } # First line full body, second line upper body, third line lower body, fourth line over body, fifth line accessories label_mapping = [ ['women-dress-mini', 'women-dress-dress', 'women-dress-maxi', 'women-dress-midi', 'women-playsuitsjumpsuits-playsuit', 'women-playsuitsjumpsuits-jumpsuit', 'women-coords-coords', 'women-swimwear-onepieces', 'women-swimwear-bikinisets'], ['women-sweatersknits-cardigan', 'women-top-waistcoat', 'women-top-blouse', 'women-sweatersknits-blouse', 'women-sweatersknits-sweater', 'women-top-top', 'women-loungewear-hoodie', 'women-top-camistanks', 'women-top-tshirt', 'women-top-croptop', 'women-loungewear-sweatshirt', 'women-top-body'], ['women-loungewear-joggers', 'women-bottom-trousers', 'women-bottom-leggings', 'women-bottom-jeans', 'women-bottom-shorts', 'women-bottom-skirt', 'women-loungewear-activewear', 'women-bottom-joggers'], ['women-top-shirt', 'women-outwear-coatjacket', 'women-outwear-blazer', 'women-outwear-coatjacket', 'women-outwear-kimonos'], ['women-accessories-bags'] ] MODEL_NAME = "valentinafeve/yolos-fashionpedia" feature_extractor = YolosImageProcessor.from_pretrained('hustvl/yolos-small') model = YolosForObjectDetection.from_pretrained(MODEL_NAME) def get_category_index(category): # Find index of label mapping for i, labels in enumerate(label_mapping): if category in labels: break return i def get_yolo_index(category): # Find index of yolo mapping return yolo_mapping[category] def fix_channels(t): """ Some images may have 4 channels (transparent images) or just 1 channel (black and white images), in order to let the images have only 3 channels. I am going to remove the fourth channel in transparent images and stack the single channel in back and white images. :param t: Tensor-like image :return: Tensor-like image with three channels """ if len(t.shape) == 2: return ToPILImage()(torch.stack([t for i in (0, 0, 0)])) if t.shape[0] == 4: return ToPILImage()(t[:3]) if t.shape[0] == 1: return ToPILImage()(torch.stack([t[0] for i in (0, 0, 0)])) return ToPILImage()(t) def idx_to_text(i): return cats[i] # Random colors used for visualization COLORS = [[0.000, 0.447, 0.741], [0.850, 0.325, 0.098], [0.929, 0.694, 0.125], [0.494, 0.184, 0.556], [0.466, 0.674, 0.188], [0.301, 0.745, 0.933]] # for output bounding box post-processing def box_cxcywh_to_xyxy(x): x_c, y_c, w, h = x.unbind(1) b = [(x_c - 0.5 * w), (y_c - 0.5 * h), (x_c + 0.5 * w), (y_c + 0.5 * h)] return torch.stack(b, dim=1) def rescale_bboxes(out_bbox, size): img_w, img_h = size b = box_cxcywh_to_xyxy(out_bbox) b = b * torch.tensor([img_w, img_h, img_w, img_h], dtype=torch.float32) return b def plot_results(pil_img, prob, boxes): plt.figure(figsize=(16,10)) plt.imshow(pil_img) ax = plt.gca() colors = COLORS * 100 i = 0 crops = [] crop_classes = [] for p, (xmin, ymin, xmax, ymax), c in zip(prob, boxes.tolist(), colors): cl = p.argmax() # Save each box as an image box_img = pil_img.crop((xmin, ymin, xmax, ymax)) crops.append(box_img) crop_classes.append(idx_to_text(cl)) ax.add_patch(plt.Rectangle((xmin, ymin), xmax - xmin, ymax - ymin, fill=False, color=c, linewidth=3)) ax.text(xmin, ymin, idx_to_text(cl), fontsize=10, bbox=dict(facecolor=c, alpha=0.8)) i += 1 # Remove white padding all around the image plt.axis('off') plt.subplots_adjust(left=0, right=1, top=1, bottom=0) output_img = plt.gcf() plt.close() return output_img, crops, crop_classes def visualize_predictions(image, outputs, threshold=0.8): # Keep only predictions with confidence >= threshold probas = outputs.logits.softmax(-1)[0, :, :-1] keep = probas.max(-1).values > threshold # Convert predicted boxes from [0; 1] to image scales bboxes_scaled = rescale_bboxes(outputs.pred_boxes[0, keep].cpu(), image.size) # Get filtered probabilities and boxes based on the filter list filter_set = set(filter) filtered_probas_boxes = [ (proba, box) for proba, box in zip(probas[keep], bboxes_scaled) if idx_to_text(proba.argmax()) not in filter_set ] # If there is a jumpsuit or dress detected, remove them if there are other clothes detected contains_jumpsuit_or_dress = any(idx_to_text(proba.argmax()) in ["jumpsuit", "dress"] for proba, _ in filtered_probas_boxes) if contains_jumpsuit_or_dress and len(filtered_probas_boxes) > 1: filtered_probas_boxes = [ (proba, box) for proba, box in filtered_probas_boxes if idx_to_text(proba.argmax()) not in ["jumpsuit", "dress"] ] # Remove duplicates: Only keep one box per class unique_classes = set() unique_filtered_probas_boxes = [] for proba, box in filtered_probas_boxes: class_text = idx_to_text(proba.argmax()) if class_text not in unique_classes: unique_classes.add(class_text) unique_filtered_probas_boxes.append((proba, box)) # If there are remaining filtered probabilities, plot results output_img = None crops = None crop_classes = None if unique_filtered_probas_boxes: final_probas, final_boxes = zip(*unique_filtered_probas_boxes) output_img, crops, crop_classes = plot_results(image, list(final_probas), torch.stack(final_boxes)) # Return the classes of the detected objects return [proba.argmax().item() for proba, _ in unique_filtered_probas_boxes], output_img, crops, crop_classes @spaces.GPU def get_objects(image, threshold=0.8): class_counts = {} image = fix_channels(ToTensor()(image)) image = image.resize((600, 800)) inputs = feature_extractor(images=image, return_tensors="pt") outputs = model(**inputs) detected_classes, output_img, crops, crop_classes = visualize_predictions(image, outputs, threshold=threshold) for cl in detected_classes: class_name = idx_to_text(cl) if class_name not in class_counts: class_counts[class_name] = 0 class_counts[class_name] += 1 if crop_classes is not None: crop_classes = [get_yolo_index(c) for c in crop_classes] return class_counts, output_img, crops, crop_classes def encode_images_to_base64(cropped_list): base64_images = [] for image in cropped_list: with io.BytesIO() as buffer: image.convert('RGB').save(buffer, format='JPEG') base64_image = base64.b64encode(buffer.getvalue()).decode('utf-8') base64_images.append(base64_image) return base64_images # def get_cropped_images(images,category): # cropped_list = [] # resultsPerCategory = {} # for num, image in enumerate(images): # image = open_image_from_url(image) # class_counts, output_img, cropped_images, cropped_classes = get_objects(image, 0.37) # if not class_counts: # continue # # Get the inverse category as any other mapping label except the current one corresponding category # inverse_category = [label for i, labels in enumerate(label_mapping) for label in labels if i != get_category_index(category) and i != 0] # # If category is a cardigan, we don't recommend category indices 1 and 3 # if category == 'women-sweatersknits-cardigan': # inverse_category = [label for i, labels in enumerate(label_mapping) for label in labels if i != get_category_index(category) and i != 1 and i != 3] # for i, image in enumerate(cropped_images): # cropped_category = cropped_classes[i] # print(cropped_category, cropped_classes[i], get_category_index(category)) # specific_category = label_mapping[cropped_category] # if cropped_category == get_category_index(category): # continue # cropped_list.append(image) # base64_images = encode_images_to_base64(cropped_list) # return base64_images def get_cropped_images(images, category): cropped_list = [] resultsPerCategory = {} for num, image in enumerate(images): image = open_image_from_url(image) class_counts, output_img, cropped_images, cropped_classes = get_objects(image, 0.37) if not class_counts: continue for i, image in enumerate(cropped_images): cropped_list.append(image) # Convert cropped images to base64 strings base64_images = encode_images_to_base64(cropped_list) return base64_images # Define the Gradio interface with the updated components iface = gr.Interface( fn=shot, inputs=[ gr.Textbox(label="Image URLs (starting with http/https) comma seperated "), gr.Textbox(label="Category"), gr.Textbox(label="Level; accepted 'variant' or 'product'") ], outputs="text", examples=[ [['https://d2q1sfov6ca7my.cloudfront.net/eyJidWNrZXQiOiAiaGljY3VwLWltYWdlLWhvc3RpbmciLCAia2V5IjogIlc4MDAwMDAwMTM0LU9SL1c4MDAwMDAwMTM0LU9SLTEuanBnIiwgImVkaXRzIjogeyJyZXNpemUiOiB7IndpZHRoIjogODAwLCAiaGVpZ2h0IjogMTIwMC4wLCAiZml0IjogIm91dHNpZGUifX19', 'https://d2q1sfov6ca7my.cloudfront.net/eyJidWNrZXQiOiAiaGljY3VwLWltYWdlLWhvc3RpbmciLCAia2V5IjogIlc4MDAwMDAwMTM0LU9SL1c4MDAwMDAwMTM0LU9SLTIuanBnIiwgImVkaXRzIjogeyJyZXNpemUiOiB7IndpZHRoIjogODAwLCAiaGVpZ2h0IjogMTIwMC4wLCAiZml0IjogIm91dHNpZGUifX19', 'https://d2q1sfov6ca7my.cloudfront.net/eyJidWNrZXQiOiAiaGljY3VwLWltYWdlLWhvc3RpbmciLCAia2V5IjogIlc4MDAwMDAwMTM0LU9SL1c4MDAwMDAwMTM0LU9SLTMuanBnIiwgImVkaXRzIjogeyJyZXNpemUiOiB7IndpZHRoIjogODAwLCAiaGVpZ2h0IjogMTIwMC4wLCAiZml0IjogIm91dHNpZGUifX19'], "women-top-shirt","variant"]], description="Add an image URL (starting with http/https) or upload a picture, and provide a list of labels separated by commas.", title="Full product flow" ) # Launch the interface iface.launch()