# spaces MUST be imported first — before torch or any CUDA-related package import spaces import time import os import sys import gc import io import json import base64 import datetime import gradio as gr import torch import cv2 import numpy as np from PIL import Image, ImageOps # Add paths BASE_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, BASE_DIR) sys.path.insert(0, os.path.join(BASE_DIR, "lib")) import firebase_config import firestore_service import gdown from dotenv import load_dotenv load_dotenv() import requests # --------------------------------------------------------------------------- # GEMINI VISION — worn vs flat check & pant classification # --------------------------------------------------------------------------- _GEMINI_PROMPT = """ You are an expert garment inspector. Inspect this pants/trousers photo. Determine two things: 1. "is_worn_by_person": - Set to true ONLY IF a living human being or model is actively WEARING the pants (a person's legs, body, torso, shoes, or skin are physically inside the pants). - Set to false IF the pants are placed empty or laid flat on the floor, bed, table, tile, ground, or hanger (even if the pant legs are spread open). - If there is NO visible human body inside the pants, is_worn_by_person MUST be false. 2. "pant_type": - Classify into ONE category: "Jeans", "Formal Pants", "Chinos", "Cargo Pants", "Track Pants / Joggers", "Linen / Casual Pants", "Shorts", or "Other". Return ONLY JSON: {"is_worn_by_person": false, "pant_type": "Jeans"} """ def check_pant_with_gemini(img_pil: Image.Image): """ Checks if a pant is worn by a person, and classifies the pant type using Gemini. Non-blocking: if GEMINI_API_KEY is not set or request fails, returns (False, "Pant"). """ api_key = os.getenv("GEMINI_API_KEY") if not api_key: return False, "Pant" try: max_dim = 768 w, h = img_pil.size if max(w, h) > max_dim: s = max_dim / max(w, h) thumb = img_pil.resize((int(w * s), int(h * s)), Image.BICUBIC) else: thumb = img_pil buf = io.BytesIO() thumb.convert("RGB").save(buf, format="JPEG", quality=80) img_b64 = base64.b64encode(buf.getvalue()).decode("utf-8") model_name = os.getenv("GEMINI_MODEL", "gemini-2.5-flash") url = f"https://generativelanguage.googleapis.com/v1beta/models/{model_name}:generateContent?key={api_key}" payload = { "contents": [{ "parts": [ {"text": _GEMINI_PROMPT}, {"inline_data": {"mime_type": "image/jpeg", "data": img_b64}} ] }], "generationConfig": { "response_mime_type": "application/json", "temperature": 0.0 } } resp = requests.post(url, json=payload, timeout=8) if resp.status_code == 200: data = resp.json() candidates = data.get("candidates", []) if candidates: part = candidates[0].get("content", {}).get("parts", [{}])[0] text = part.get("text", "{}") parsed = json.loads(text) is_worn = bool(parsed.get("is_worn_by_person", False)) pant_type = str(parsed.get("pant_type", "Pant")).strip() print(f"[Gemini] is_worn={is_worn}, pant_type='{pant_type}'", flush=True) return is_worn, pant_type else: print(f"[Gemini] API error: {resp.status_code} {resp.text[:150]}", flush=True) except Exception as e: print(f"[Gemini] Error (continuing without blocking): {e}", flush=True) return False, "Pant" # Initialize Firebase try: firebase_config.initialize_firebase() except Exception as e: print(f"Firebase init error: {e}") # Initialize Cloudinary import cloudinary def init_cloudinary(): if not os.getenv("CLOUDINARY_CLOUD_NAME"): return False import cloudinary.uploader cloudinary.config( cloud_name=os.getenv("CLOUDINARY_CLOUD_NAME"), api_key=os.getenv("CLOUDINARY_API_KEY"), api_secret=os.getenv("CLOUDINARY_API_SECRET"), secure=True ) return True init_cloudinary() # Constants WAIST_L, WAIST_C, WAIST_R = 6, 0, 30 OUTER_ANKLE_L, OUTER_ANKLE_R = 14, 16 INNER_ANKLE_L, INNER_ANKLE_R = 9, 21 INSEAM = 27 FALLBACK_INNER_ANKLE_L = 108 # HRNet model paths CFG_FILE = os.path.join(BASE_DIR, "experiments/deepfashion2/hrnet/w48_384x288_adam_lr1e-3.yaml") MODEL_FILE = os.path.join(BASE_DIR, "lib/models/pose_hrnet-w48_384x288-deepfashion2_mAP_0.7017.pth") MODEL_URL = "https://drive.google.com/file/d/1Srj8YJLu00aZm368sCtEIWB8Vkpr7kXA/view?usp=sharing" def ensure_model_exists(): if not os.path.exists(MODEL_FILE): os.makedirs(os.path.dirname(MODEL_FILE), exist_ok=True) print("Downloading HRNet model...", flush=True) gdown.download(MODEL_URL, MODEL_FILE, quiet=True) ensure_model_exists() # Load HRNet model (CPU — GPU assigned dynamically via @spaces.GPU) from config import cfg, update_config import models import argparse args = argparse.Namespace(cfg=CFG_FILE, opts=[], modelDir="", logDir="", dataDir="", prevModelDir="") update_config(cfg, args) cfg.freeze() _hrnet = models.pose_hrnet.get_pose_net(cfg, is_train=False) _hrnet.load_state_dict(torch.load(MODEL_FILE, map_location="cpu"), strict=False) _hrnet.eval() # Load SAM3 model (CPU — GPU assigned dynamically via @spaces.GPU) from huggingface_hub import hf_hub_download _sam3_model = None _sam3_processor = None cumulative_gpu_time = 0.0 def init_sam3(): global _sam3_model, _sam3_processor try: from sam3.model_builder import build_sam3_image_model from sam3.model.sam3_image_processor import Sam3Processor token = os.getenv("HF_TOKEN") if not token: print("WARNING: HF_TOKEN environment variable not set. Downloading SAM3 may fail.") print("Downloading/loading SAM3 model...", flush=True) model_path = hf_hub_download(repo_id="facebook/sam3", filename="sam3.pt", token=token) _sam3_model = build_sam3_image_model(checkpoint_path=model_path) _sam3_model.eval() _sam3_processor = Sam3Processor(_sam3_model) print("SAM3 loaded successfully.", flush=True) except Exception as e: print(f"Failed to load SAM3: {e}") init_sam3() # --------------------------------------------------------------------------- # SINGLE @spaces.GPU function — runs both ISNet BG removal and HRNet # --------------------------------------------------------------------------- @spaces.GPU def run_pipeline_gpu(img_pil: Image.Image, prompt_text="pants"): """ Runs the full GPU pipeline: 1. SAM3 garment segmentation (uses GPU if available) 2. HRNet landmark estimation (uses GPU) Returns: img_np : numpy HWC image array (for drawing) binary_mask : numpy 2-D mask array heatmaps : numpy array of shape (num_joints, H, W) img_h, img_w: original dimensions used for landmark scaling gpu_time : time spent in GPU function this run """ global _hrnet, _sam3_model, _sam3_processor, cumulative_gpu_time t_start = time.time() # -- Step 1: SAM3 garment segmentation -- t1 = time.time() if _sam3_model is not None and _sam3_processor is not None: _sam3_model.to("cuda" if torch.cuda.is_available() else "cpu") try: device = "cuda" if torch.cuda.is_available() else "cpu" # ZeroGPU does not support bfloat16 — use float16 on GPU, float32 on CPU dtype = torch.float16 if torch.cuda.is_available() else torch.float32 with torch.inference_mode(), torch.autocast(device_type=device, dtype=dtype): inference_state = _sam3_processor.set_image(img_pil) output = _sam3_processor.set_text_prompt(state=inference_state, prompt=prompt_text) masks = output["masks"] scores = output["scores"] if torch.is_tensor(scores): scores_np = scores.cpu().numpy() else: scores_np = np.asarray(scores) # Filter masks with confidence > 0.5 (as per the colab notebook) valid_indices = np.where(scores_np > 0.5)[0] if len(valid_indices) == 0: # If nothing > 0.5, fallback to all indices to at least output something valid_indices = np.arange(len(scores_np)) # Find the largest mask by area among the valid ones best_idx = valid_indices[0] max_area = -1 if torch.is_tensor(masks): for idx in valid_indices: area = masks[idx].sum().item() if area > max_area: max_area = area best_idx = idx else: for idx in valid_indices: area = masks[idx].sum() if area > max_area: max_area = area best_idx = idx best_mask = masks[best_idx] if torch.is_tensor(best_mask): best_mask = best_mask.cpu().numpy() # Ensure shape is (H, W) if best_mask.ndim == 3 and best_mask.shape[0] == 1: best_mask = best_mask[0] if best_mask.shape != (img_pil.height, img_pil.width): best_mask = cv2.resize(best_mask.astype(np.uint8), (img_pil.width, img_pil.height), interpolation=cv2.INTER_NEAREST) mask_np = (best_mask > 0).astype(np.uint8) * 255 except Exception as e: print(f"Error during SAM3 inference: {e}", flush=True) mask_np = np.zeros((img_pil.height, img_pil.width), dtype=np.uint8) else: print("SAM3 model not initialized, falling back to empty mask.") mask_np = np.zeros((img_pil.size[1], img_pil.size[0]), dtype=np.uint8) isnet_time = time.time() - t1 print(f"SAM3: {isnet_time:.2f}s", flush=True) # Generate transparent cutout / white background img_rgb = img_pil.convert("RGB") white_bg = Image.new("RGB", img_pil.size, "WHITE") mask_pil = Image.fromarray(mask_np) white_bg.paste(img_rgb, mask=mask_pil) # Portrait orientation helpers def _portrait(arr_img): h, w = arr_img.shape[:2] if w > h: arr_img = cv2.rotate(arr_img, cv2.ROTATE_90_COUNTERCLOCKWISE) return arr_img img_np = _portrait(np.array(white_bg)) binary_mask = _portrait(mask_np) img_np_orig = _portrait(np.array(img_rgb)) # -- Step 2: HRNet landmark estimation -- t2 = time.time() H, W = img_np.shape[:2] inp = cv2.resize(img_np, (288, 384)).astype(np.float32) / 255.0 inp = inp.transpose(2, 0, 1) inp_tensor = torch.from_numpy(inp).unsqueeze(0) device = "cuda" if torch.cuda.is_available() else "cpu" _hrnet = _hrnet.to(device) with torch.inference_mode(): heatmaps = _hrnet(inp_tensor.to(device)).squeeze(0).cpu().numpy() hrnet_time = time.time() - t2 total = time.time() - t_start cumulative_gpu_time += total print(f"HRNet: {hrnet_time:.2f}s", flush=True) print(f"Total GPU function: {total:.2f}s (Cumulative since boot: {cumulative_gpu_time:.2f}s)", flush=True) return img_np_orig, binary_mask, heatmaps, H, W, total # --------------------------------------------------------------------------- # CPU helpers — landmark decoding, refinement, measurement, drawing # --------------------------------------------------------------------------- def decode_landmarks(heatmaps, H, W): num_joints, hm_h, hm_w = heatmaps.shape pts = {} for jid in range(num_joints): y, x = np.unravel_index(np.argmax(heatmaps[jid]), heatmaps[jid].shape) pts[jid] = (int(x * W / hm_w), int(y * H / hm_h)) if FALLBACK_INNER_ANKLE_L in pts and pts[FALLBACK_INNER_ANKLE_L][1] > pts[INNER_ANKLE_L][1]: pts[INNER_ANKLE_L] = pts[FALLBACK_INNER_ANKLE_L] return pts def refine_landmarks(pts, binary_mask, max_snap_dist_ratio=0.08): refined_pts = pts.copy() contours, _ = cv2.findContours(binary_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE) if not contours: return refined_pts largest_contour = max(contours, key=cv2.contourArea) contour_pts = largest_contour.reshape(-1, 2) H, W = binary_mask.shape[:2] max_snap_dist = max_snap_dist_ratio * max(H, W) boundary_ids = [WAIST_L, WAIST_R, OUTER_ANKLE_L, OUTER_ANKLE_R, INNER_ANKLE_L, INNER_ANKLE_R, INSEAM] for lid in boundary_ids: if lid not in pts: continue pt = np.array(pts[lid], dtype=np.float32) dists = np.linalg.norm(contour_pts - pt, axis=1) min_idx = np.argmin(dists) if dists[min_idx] <= max_snap_dist: refined_pts[lid] = (int(contour_pts[min_idx][0]), int(contour_pts[min_idx][1])) return refined_pts def round_half(x): return round(x * 2) / 2 def dist(p, a, b): x1, y1 = p[a]; x2, y2 = p[b] return float(np.hypot(x1 - x2, y1 - y2)) def draw_measurements(img, pts): draw = img.copy() for a, b in [ (WAIST_L, WAIST_C), (WAIST_C, WAIST_R), (WAIST_L, OUTER_ANKLE_L), (WAIST_R, OUTER_ANKLE_R), (INNER_ANKLE_L, OUTER_ANKLE_L), (INNER_ANKLE_R, OUTER_ANKLE_R), (INNER_ANKLE_L, INSEAM), (INNER_ANKLE_R, INSEAM), ]: cv2.line(draw, pts[a], pts[b], (60, 120, 180), 2) return draw def upload_image_to_cloud(image_obj, folder): if not os.getenv("CLOUDINARY_CLOUD_NAME"): return None try: import cloudinary.uploader if isinstance(image_obj, (np.ndarray, np.generic)): pil_img = Image.fromarray(image_obj) buf = io.BytesIO() pil_img.save(buf, format="JPEG", quality=80) buf.seek(0) image_obj = buf return cloudinary.uploader.upload(image_obj, folder=folder, resource_type="image", timeout=120).get("secure_url") except Exception as e: print(f"Image upload failed: {e}") return None def verify_user_from_token(token): if not token: raise ValueError("No token found") from firebase_admin import auth as firebase_auth return firebase_auth.verify_id_token(token)['uid'] # --------------------------------------------------------------------------- # Main Gradio handler # --------------------------------------------------------------------------- def process_fit(name, waist_in, input_image, request: gr.Request): if not name: return None, "Please enter your name.", "", "" if input_image is None: return None, "Please upload an image.", "", "" try: # Auth uid = None if request: token = request.query_params.get("token") if token: try: uid = verify_user_from_token(token) except Exception as e: print(f"Auth failed: {e}") # Preprocess: EXIF transpose + optional resize img_pil = ImageOps.exif_transpose(input_image) orig_w, orig_h = img_pil.size MAX_DIM = 1024 scale_x = scale_y = 1.0 resized = False if max(orig_w, orig_h) > MAX_DIM: scale = MAX_DIM / max(orig_w, orig_h) new_w, new_h = int(orig_w * scale), int(orig_h * scale) scale_x, scale_y = new_w / orig_w, new_h / orig_h resized = True img_cv2 = cv2.cvtColor(np.array(img_pil.convert("RGB")), cv2.COLOR_RGB2BGR) img_pil = Image.fromarray(cv2.cvtColor( cv2.resize(img_cv2, (new_w, new_h), interpolation=cv2.INTER_AREA), cv2.COLOR_BGR2RGB )) # Check if pant is worn by a person using Gemini Vision is_worn, pant_type = check_pant_with_gemini(img_pil) if is_worn: return ( None, "
Find your perfect size in 10 seconds.