# ====================================================== # REMBG + HRNet Fashion Landmark Measurement - Streamlit UI # ====================================================== import os import sys import json import argparse import gc import datetime import numpy as np import streamlit as st from PIL import Image, ImageOps import sys import os # ------------------------------------------------------ # DYNAMIC PATH RESOLUTION & PATH INJECTIONS # ------------------------------------------------------ BASE_DIR = os.path.dirname(os.path.abspath(__file__)) if os.path.exists(os.path.join(BASE_DIR, "experiments")): HRNET_ROOT = BASE_DIR else: parent_dir = os.path.dirname(BASE_DIR) if os.path.exists(os.path.join(parent_dir, "experiments")): HRNET_ROOT = parent_dir else: HRNET_ROOT = BASE_DIR sys.path.insert(0, HRNET_ROOT) sys.path.insert(0, os.path.join(HRNET_ROOT, "lib")) sys.path.insert(0, os.path.join(HRNET_ROOT, "streamlit_app")) try: from components.phone_tilt import pant_upload_ui except ImportError: pant_upload_ui = None import gdown import io import time from dotenv import load_dotenv # Load environment variables at the very beginning load_dotenv() # Diagnostic startup logging print("[DIAGNOSTIC] Application started", flush=True) try: import psutil except ImportError: psutil = None def log_memory(label): """ Log application process memory (RSS in GB) for Community Cloud optimization. """ if psutil is None: print(f"[MEMORY] {label}: psutil not installed", flush=True) return 0.0 try: process = psutil.Process(os.getpid()) rss_gb = process.memory_info().rss / (1024 ** 3) print(f"[MEMORY] {label}: {rss_gb:.3f} GB", flush=True) return rss_gb except Exception as e: print(f"[MEMORY] {label}: (Failed to get process memory: {e})", flush=True) return 0.0 log_memory("Clean Streamlit/Python startup") # Defer heavy imports to optimize startup time torch = None cv2 = None cloudinary = None rembg = None # ------------------------------------------------------ # DYNAMIC PATH RESOLUTION & PATH INJECTIONS # ------------------------------------------------------ BASE_DIR = os.path.dirname(os.path.abspath(__file__)) if os.path.exists(os.path.join(BASE_DIR, "experiments")): HRNET_ROOT = BASE_DIR else: parent_dir = os.path.dirname(BASE_DIR) if os.path.exists(os.path.join(parent_dir, "experiments")): HRNET_ROOT = parent_dir else: HRNET_ROOT = BASE_DIR # Insert paths into sys.path to ensure module resolution works from both root and streamlit_app sys.path.insert(0, HRNET_ROOT) sys.path.insert(0, os.path.join(HRNET_ROOT, "lib")) sys.path.insert(0, os.path.join(HRNET_ROOT, "streamlit_app")) # ------------------------------------------------------ # SERVER-SIDE AUTH GUARD # ------------------------------------------------------ import firebase_config import auth import firestore_service # Initialize Firebase (Server-Side) firebase_config.initialize_firebase() is_embedded = st.query_params.get("embed") == "true" # Auth is optional - try to verify token if provided, but don't block access if "uid" not in st.session_state: token = st.query_params.get("token") if token: try: with st.spinner("Verifying session..."): user_uid = auth.verify_user_from_query() st.session_state["uid"] = user_uid if not is_embedded: st.query_params.clear() st.rerun() except Exception as e: # Token invalid - still allow access, just without Firestore save st.session_state["uid"] = None else: st.session_state["uid"] = None # ------------------------------------------------------ # STREAMLIT CONFIG # ------------------------------------------------------ st.set_page_config( page_title="DASH AI Fit + BG Remover", layout="centered", initial_sidebar_state="collapsed" ) # Fullscreen listener st.html(""" """) # ------------------------------------------------------ # PREMIUM CSS INJECTION # ------------------------------------------------------ CSS_CODE = r""" """ st.markdown(CSS_CODE, unsafe_allow_html=True) # ------------------------------------------------------ # LAZY IMPORTS & RESOURCE CACHING # ------------------------------------------------------ def get_cloudinary(): global cloudinary if cloudinary is None: import cloudinary import cloudinary.uploader return cloudinary def init_cloudinary(): """Configure Cloudinary using environment variables.""" cld = get_cloudinary() try: cld.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 except Exception as e: st.error(f"Failed to initialize Cloudinary: {e}") return False # ------------------------------------------------------ # REMBG MODEL PATHS & SETUP (STRICT SINGLE INSTANCE) # ------------------------------------------------------ @st.cache_resource(show_spinner="Loading background removal model...") def load_rembg_session(model_name="isnet-general-use"): """ Load a single cached rembg session. """ global rembg if rembg is None: import rembg print(f"[{model_name.upper()}] Starting initialization", flush=True) rss_before = log_memory(f"Before {model_name} initialization") try: import onnxruntime as ort so = ort.SessionOptions() so.intra_op_num_threads = 1 so.inter_op_num_threads = 1 so.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL session = rembg.new_session(model_name, sess_opts=so) print(f"[{model_name.upper()}] initialization completed", flush=True) rss_after = log_memory(f"After {model_name} initialization") delta = rss_after - rss_before print(f"[MEMORY] {model_name} initialization delta = {delta:.3f} GB", flush=True) return session, model_name except Exception as e: print(f"[{model_name.upper()}] FAILED: {e}", flush=True) import traceback traceback.print_exc() raise e # ------------------------------------------------------ # SEGMENTATION MASK & LANDMARK REFINEMENT HELPERS # ------------------------------------------------------ def load_mask_portrait(mask_np): """Rotate the mask to match portrait image orientation if landscape.""" global cv2 if cv2 is None: import cv2 h, w = mask_np.shape[:2] if w > h: mask_np = cv2.rotate(mask_np, cv2.ROTATE_90_COUNTERCLOCKWISE) return mask_np def refine_landmarks(pts, binary_mask, max_snap_dist_ratio=0.08): """ Refines HRNet landmarks by snapping them to the closest outer boundary of the pants. Only boundary landmarks are snapped. WAIST_C (inside waistband) is ignored. """ global cv2 if cv2 is None: import cv2 refined_pts = pts.copy() contours, _ = cv2.findContours(binary_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE) if not contours: return refined_pts, 0.0 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_landmark_ids = [ WAIST_L, WAIST_R, OUTER_ANKLE_L, OUTER_ANKLE_R, INNER_ANKLE_L, INNER_ANKLE_R, INSEAM ] snap_distances = [] for lid in boundary_landmark_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) min_dist = dists[min_idx] if min_dist <= max_snap_dist: refined_pts[lid] = (int(contour_pts[min_idx][0]), int(contour_pts[min_idx][1])) snap_distances.append(min_dist) avg_snap_dist = float(np.mean(snap_distances)) if snap_distances else 0.0 return refined_pts, avg_snap_dist def draw_landmarks_debug(img, orig_pts, refined_pts): """Draws original and refined landmarks on the image for visual verification.""" global cv2 if cv2 is None: import cv2 draw = img.copy() boundary_landmark_ids = [ WAIST_L, WAIST_R, OUTER_ANKLE_L, OUTER_ANKLE_R, INNER_ANKLE_L, INNER_ANKLE_R, INSEAM ] # Draw original landmarks in red for lid, pt in orig_pts.items(): color = (255, 0, 0) if lid in boundary_landmark_ids else (255, 165, 0) cv2.circle(draw, pt, 4, color, -1) # Draw refined landmarks in green for lid, pt in refined_pts.items(): if lid in boundary_landmark_ids: cv2.circle(draw, pt, 4, (0, 255, 0), -1) # Draw line showing direction of shift if lid in orig_pts: cv2.line(draw, orig_pts[lid], pt, (0, 0, 255), 1) return draw # ------------------------------------------------------ # HRNET PATHS & SETUP (STRICT SINGLE INSTANCE) # ------------------------------------------------------ CFG_FILE = os.path.join(HRNET_ROOT, "experiments/deepfashion2/hrnet/w48_384x288_adam_lr1e-3.yaml") MODEL_FILE = os.path.join(HRNET_ROOT, "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) with st.spinner("Initializing AI engine..."): gdown.download(MODEL_URL, MODEL_FILE, quiet=True) @st.cache_resource(show_spinner="Loading HRNet Landmark Detection model...") def load_hrnet(): """ Load and cache HRNet model. Uses @st.cache_resource so it stays in RAM between requests (avoids ~5–10 s reload on each analysis). """ global torch, cv2 if torch is None: import torch if cv2 is None: import cv2 # Limit CPU threads before model creation to reduce RSS overhead on cloud VMs torch.set_num_threads(2) print("[HRNET] Starting HRNet initialization", flush=True) rss_before = log_memory("Before HRNet initialization") ensure_model_exists() from config import cfg, update_config import models args = argparse.Namespace(cfg=CFG_FILE, opts=[], modelDir="", logDir="", dataDir="", prevModelDir="") update_config(cfg, args) cfg.freeze() model = models.pose_hrnet.get_pose_net(cfg, is_train=False) model.load_state_dict(torch.load(MODEL_FILE, map_location="cpu"), strict=False) model.eval() rss_after = log_memory("After HRNet initialization") delta = rss_after - rss_before print(f"[MEMORY] HRNet initialization delta = {delta:.3f} GB", flush=True) return model # ------------------------------------------------------ # AI HELPERS # ------------------------------------------------------ def load_image_portrait(pil_img): global cv2 if cv2 is None: import cv2 img = np.array(pil_img.convert("RGB")) h, w = img.shape[:2] if w > h: img = cv2.rotate(img, cv2.ROTATE_90_COUNTERCLOCKWISE) return img def round_half(x): return round(x * 2) / 2 # 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 def run_hrnet(img): """ Run HRNet inference using the cached model instance. Intermediate tensors are freed immediately after inference. """ global torch, cv2 if torch is None: import torch if cv2 is None: import cv2 model = load_hrnet() # Cached — no reload cost after first request H, W = img.shape[:2] inp = cv2.resize(img, (288, 384)).astype(np.float32) / 255.0 inp = inp.transpose(2, 0, 1) inp_tensor = torch.from_numpy(inp).unsqueeze(0) log_memory("Before HRNet inference") with torch.inference_mode(): heatmaps_tensor = model(inp_tensor).squeeze(0) heatmaps = heatmaps_tensor.cpu().numpy() log_memory("After HRNet inference") pts = {} num_joints, hm_h, hm_w = heatmaps.shape 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] # Free intermediate tensors (NOT the model — it stays cached) del inp, inp_tensor, heatmaps_tensor, heatmaps gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() log_memory("After HRNet cleanup") return pts 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): global cv2 if cv2 is None: import cv2 draw = img.copy() connections = [ (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), ] for a, b in connections: cv2.line(draw, pts[a], pts[b], (60,120,180), 2) return draw def resize_and_compress(image_arr, max_dim=1000, quality=75): """Optimizes the image: resizes to max_dim and compresses to JPEG.""" try: pil_img = Image.fromarray(image_arr) width, height = pil_img.size if max(width, height) > max_dim: if width > height: new_width = max_dim new_height = int(height * (max_dim / width)) else: new_height = max_dim new_width = int(width * (max_dim / height)) pil_img = pil_img.resize((new_width, new_height), Image.BICUBIC) buf = io.BytesIO() pil_img.save(buf, format="JPEG", quality=quality, optimize=True) buf.seek(0) return buf except Exception as e: print(f"Optimization failed: {e}") return None def upload_image_to_cloud(image_obj, folder): cld = get_cloudinary() try: 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) obj_to_upload = buf else: obj_to_upload = image_obj response = cld.uploader.upload( obj_to_upload, folder=folder, resource_type="image", timeout=120 ) return response.get("secure_url") except Exception as e: print(f"Image upload failed: {e}") return None # ------------------------------------------------------ # UI HEADER # ------------------------------------------------------ st.markdown("""
Find your perfect size in 10 seconds.
""", unsafe_allow_html=True) # ------------------------------------------------------ # STEP 1: NAME # ------------------------------------------------------ st.markdown('
STEP 1 - ENTER YOUR NAME
', unsafe_allow_html=True) user_name = st.text_input("Enter your name", placeholder="Enter your name", label_visibility="collapsed") # ------------------------------------------------------ # STEP 2: IMAGE UPLOAD # ------------------------------------------------------ st.markdown('
STEP 2 - upload a flat laid pant image
', unsafe_allow_html=True) # Use the custom all-in-one upload/tilt/camera component if pant_upload_ui: _component_result = pant_upload_ui(key="pant_uploader") # Store the image in session_state so it survives the rerun triggered # by the "ANALYZE MY FIT" button click (component value resets on rerun). if _component_result is not None: st.session_state["uploaded_img_pil"] = _component_result uploaded_img_pil = st.session_state.get("uploaded_img_pil", None) # Give user visual feedback on desktop/mobile that their upload succeeded if uploaded_img_pil is not None: st.success("Image uploaded successfully!") st.image(uploaded_img_pil, caption="Your Uploaded Image", width=150) else: uploaded_img_pil = None uploaded = None # fallback for file-based pipeline path # ------------------------------------------------------ # STEP 3: WAIST SIZE (Editable Only) # ------------------------------------------------------ st.markdown('
STEP 3 - ENTER WAIST SIZE
', unsafe_allow_html=True) # Initialize session state for waist if not exists if 'waist_val' not in st.session_state: st.session_state.waist_val = 32 # Callback to sync input with session state def update_waist_from_input(): st.session_state.waist_val = st.session_state.waist_input # Single editable number input st.number_input( "Waist Size", min_value=24, max_value=60, value=st.session_state.waist_val, step=1, key="waist_input", on_change=update_waist_from_input, label_visibility="collapsed" ) # ------------------------------------------------------ # DEBUG MODE & OPTIONS # ------------------------------------------------------ show_debug = False # ------------------------------------------------------ # SOCIAL PROOF & ACTION # ------------------------------------------------------ st.markdown('
Used by 500+ Users
', unsafe_allow_html=True) run_ai = st.button("ANALYZE MY FIT") # ------------------------------------------------------ # VALIDATION & LOGIC # ------------------------------------------------------ if run_ai: if not user_name: st.warning("Please enter your name.") st.stop() if uploaded_img_pil is None and uploaded is None: st.warning("Please upload or capture a pant image.") st.stop() # Init Cloudinary now if needed if not init_cloudinary(): st.warning("Cloudinary config missing.") try: # Use image from the component (PIL) or fall back to uploaded file if uploaded_img_pil is not None: img_pil = ImageOps.exif_transpose(uploaded_img_pil) else: img_pil = Image.open(uploaded) img_pil = ImageOps.exif_transpose(img_pil) # --- Conditional Proportional Resize --- # Only resize if EITHER dimension exceeds MAX_PROCESSING_DIM. # Small images are passed through untouched. MAX_PROCESSING_DIM = 1024 orig_w, orig_h = img_pil.size scale_x = 1.0 scale_y = 1.0 resized = False if orig_w > MAX_PROCESSING_DIM or orig_h > MAX_PROCESSING_DIM: # Compute scale factor so the longer side lands exactly on MAX_PROCESSING_DIM scale = MAX_PROCESSING_DIM / max(orig_w, orig_h) new_w = int(orig_w * scale) new_h = int(orig_h * scale) scale_x = new_w / orig_w scale_y = new_h / orig_h resized = True import cv2 as _cv2 _img_cv2 = np.array(img_pil.convert("RGB")) _img_cv2 = _cv2.cvtColor(_img_cv2, _cv2.COLOR_RGB2BGR) _proc = _cv2.resize(_img_cv2, (new_w, new_h), interpolation=_cv2.INTER_AREA) img_pil = Image.fromarray(_cv2.cvtColor(_proc, _cv2.COLOR_BGR2RGB)) del _proc, _img_cv2 gc.collect() print(f"[RESIZE] Original: {orig_w}x{orig_h} -> Processing: {new_w}x{new_h} (scale={scale:.4f})", flush=True) else: print(f"[RESIZE] Image {orig_w}x{orig_h} is within limit — no resize.", flush=True) waist_in = st.session_state.waist_val if show_debug: st.markdown("### 🔍 Background Removal & Landmark Refinement Debug Mode") # Use cached single ISNet session to avoid loading multiple model weights into memory rembg_session, active_model = load_rembg_session("isnet-general-use") t_start = time.time() import rembg as _rembg log_memory("Before rembg inference (Debug)") removed = _rembg.remove(img_pil, session=rembg_session) log_memory("After rembg inference (Debug)") infer_time = time.time() - t_start if removed.mode == "RGBA": white_bg = Image.new("RGBA", removed.size, "WHITE") white_bg.paste(removed, mask=removed) img_rgb = white_bg.convert("RGB") alpha = np.array(removed.split()[-1]) mask_np = (alpha >= 128).astype(np.uint8) * 255 del alpha, white_bg else: img_rgb = removed.convert("RGB") gray = cv2.cvtColor(np.array(img_rgb), cv2.COLOR_RGB2GRAY) _, mask_np = cv2.threshold(gray, 240, 255, cv2.THRESH_BINARY_INV) del gray img_np = load_image_portrait(img_rgb) binary_mask = load_mask_portrait(mask_np) del mask_np gc.collect() raw_pts = run_hrnet(img_np) pts, avg_snap = refine_landmarks(raw_pts, binary_mask) # Draw on the processing-copy size (correct visual alignment) debug_img = draw_landmarks_debug(img_np, pts, pts) result_img = draw_measurements(img_np, pts) # Map landmark coords back to original space only when a resize occurred if resized: for lid in pts: px, py = pts[lid] pts[lid] = (int(px / scale_x), int(py / scale_y)) del raw_pts, binary_mask gc.collect() del img_np gc.collect() # Early-convert to bytes to free large arrays before display _removed_buf = io.BytesIO() removed.save(_removed_buf, format="PNG") _removed_buf.seek(0) del removed _debug_buf = io.BytesIO() Image.fromarray(debug_img).save(_debug_buf, format="JPEG", quality=88, optimize=True) _debug_buf.seek(0) del debug_img # result_img is needed for the shared measurement block below _result_buf = io.BytesIO() Image.fromarray(result_img).save(_result_buf, format="JPEG", quality=88, optimize=True) _result_buf.seek(0) del result_img # Convert original img_pil to bytes _orig_buf = io.BytesIO() img_pil.save(_orig_buf, format="JPEG", quality=88, optimize=True) _orig_buf.seek(0) del img_pil gc.collect() col1, col2 = st.columns(2) with col1: st.image(_removed_buf, caption="ISNet Cutout", width="stretch") with col2: st.image(_debug_buf, caption="Landmarks Refined", width="stretch") st.markdown(f""" **Model Metrics:** * **Active Model Used:** `{active_model}` * **Inference Time:** `{infer_time:.2f} seconds` * **Average Snapping Distance:** `{avg_snap:.2f} pixels` """) del _debug_buf gc.collect() else: # Step 1: Remove background with st.spinner("⚡ Removing background..."): img_w, img_h = img_pil.size img_mode = img_pil.mode channels = len(img_pil.getbands()) approx_mb = (img_w * img_h * channels) / (1024 ** 2) print(f"[IMAGE] width={img_w} height={img_h} mode={img_mode} approx={approx_mb:.2f}MB", flush=True) rembg_session, active_model = load_rembg_session() # Pre-shrink to 800px max before rembg — rembg internally resizes # to ~1024 anyway, but passing a smaller image reduces the RGBA # output array size and the inference memory spike significantly. REMBG_MAX_DIM = 800 rembg_input = img_pil _rw, _rh = img_pil.size if max(_rw, _rh) > REMBG_MAX_DIM: _rs = REMBG_MAX_DIM / max(_rw, _rh) rembg_input = img_pil.resize( (int(_rw * _rs), int(_rh * _rs)), Image.BICUBIC ) print(f"[REMBG] Resized input to {rembg_input.size} for BG removal", flush=True) import rembg as _rembg log_memory("Before rembg inference") removed_bg_img = _rembg.remove(rembg_input, session=rembg_session) log_memory("After rembg inference") del rembg_input st.info(f"Active segmentation model: **{active_model}**") # Step 2: Convert to RGB with white background with st.spinner("Preparing image for analysis..."): if removed_bg_img.mode == "RGBA": white_bg = Image.new("RGBA", removed_bg_img.size, "WHITE") white_bg.paste(removed_bg_img, mask=removed_bg_img) img_rgb = white_bg.convert("RGB") alpha = np.array(removed_bg_img.split()[-1]) mask_np = (alpha >= 128).astype(np.uint8) * 255 del alpha, white_bg else: img_rgb = removed_bg_img.convert("RGB") gray = cv2.cvtColor(np.array(img_rgb), cv2.COLOR_RGB2GRAY) _, mask_np = cv2.threshold(gray, 240, 255, cv2.THRESH_BINARY_INV) del gray img_np = load_image_portrait(img_rgb) binary_mask = load_mask_portrait(mask_np) del img_rgb, mask_np # Early-save removed_bg_img to bytes then free the large RGBA PIL object # so it is not alive in RAM while HRNet (~0.40 GB) is loading _removed_buf = io.BytesIO() removed_bg_img.save(_removed_buf, format="PNG") _removed_buf.seek(0) del removed_bg_img gc.collect() log_memory("After ISNet cleanup + early RGBA free") # Step 3: Run HRNet landmark estimation & refine positions with st.spinner("Analyzing measurements..."): raw_pts = run_hrnet(img_np) pts, _ = refine_landmarks(raw_pts, binary_mask) result_img = draw_measurements(img_np, pts) # Map landmark coords back to original space only when a resize occurred if resized: for lid in pts: px, py = pts[lid] pts[lid] = (int(px / scale_x), int(py / scale_y)) del raw_pts, binary_mask, img_np gc.collect() # --- Early conversion: free large numpy/PIL objects, keep only compact bytes --- # Convert result_img (numpy HWC array) -> JPEG bytes, then free the array _result_buf = io.BytesIO() Image.fromarray(result_img).save(_result_buf, format="JPEG", quality=88, optimize=True) _result_buf.seek(0) del result_img # NOTE: _removed_buf was already created and removed_bg_img freed earlier # (inside the non-debug branch above) to free RAM before HRNet loads. # Guard: only create _removed_buf here if show_debug path ran instead. if "_removed_buf" not in locals(): _removed_buf = io.BytesIO() # Convert original img_pil -> JPEG bytes, then free PIL object _orig_buf = io.BytesIO() img_pil.save(_orig_buf, format="JPEG", quality=88, optimize=True) _orig_buf.seek(0) del img_pil gc.collect() log_memory("After early buffer conversion") # Calculate Measurements using refined points waist_px = dist(pts, WAIST_L, WAIST_C) + dist(pts, WAIST_C, WAIST_R) outseam_px = (dist(pts, WAIST_L, OUTER_ANKLE_L) + dist(pts, WAIST_R, OUTER_ANKLE_R)) / 2 ankle_px = (dist(pts, INNER_ANKLE_L, OUTER_ANKLE_L) + dist(pts, INNER_ANKLE_R, OUTER_ANKLE_R)) / 2 inseam_px = (dist(pts, INSEAM, INNER_ANKLE_L) + dist(pts, INSEAM, INNER_ANKLE_R)) / 2 half_waist = (waist_in / 2) - 1 outseam_val = round_half(outseam_px / waist_px * half_waist) ankle_val = round_half(ankle_px / waist_px * half_waist) inseam_val = round_half(inseam_px / waist_px * half_waist) log_memory("After measurement calculation") # ------------------------------------------------------ # RESULTS DISPLAY (Styled Card) # Use pre-converted byte buffers — large arrays already freed # ------------------------------------------------------ st.image(_result_buf, caption="AI Detection Result", width="stretch") results_html = f"""
Hello {user_name},
Recommended Size
{int(waist_in)}
Waist
{waist_in}in
Outseam
{outseam_val}in
Inseam
{inseam_val}in
Ankle
{ankle_val}in
""" st.markdown(results_html, unsafe_allow_html=True) # ------------------------------------------------------ # POST MESSAGE TO PARENT (auto-save to localStorage) # ------------------------------------------------------ fit_payload = json.dumps({ "type": "dashFitResult", "name": user_name, "waist": waist_in, "outseam": float(outseam_val), "inseam": float(inseam_val), "ankle": float(ankle_val), "recommendedSize": str(int(waist_in)), "saved_at": datetime.datetime.now(datetime.timezone.utc).isoformat() }) post_msg_js = f""" """ st.html(post_msg_js) st.success("✅ Measurements saved successfully! You can now access them via 'Custom Fit' on the DASH website.") # ------------------------------------------------------ # AUTOMATIC SAVE (Server-Side, only if logged in) # Upload from _result_buf (already a compressed JPEG BytesIO) # ------------------------------------------------------ if st.session_state.get("uid"): with st.spinner("Syncing to cloud profile..."): _result_buf.seek(0) image_url = upload_image_to_cloud(_result_buf, "dashclothing/scans") if image_url: scan_data = { "name": user_name, "imageUrl": image_url, "waist": waist_in, "outseam": outseam_val, "inseam": inseam_val, "ankle": ankle_val, "status": "completed" } try: doc_id = firestore_service.save_prediction(st.session_state["uid"], scan_data) st.success("Result Saved!") except Exception as e: st.error(f"Failed to save to database: {str(e)}") else: st.warning("Image upload skipped.") # Final cleanup of all request-specific variables del _result_buf, _removed_buf, _orig_buf, pts gc.collect() log_memory("After Request Cleanup") # Clear the stored image so the next click requires a fresh upload st.session_state.pop("uploaded_img_pil", None) except Exception as e: st.error(f"An error occurred: {e}") # Clear stale image on error too st.session_state.pop("uploaded_img_pil", None)