Update app.py

app.py

@@ -1,3 +1,4 @@
+import os
 import cv2
 import numpy as np
 from PIL import Image
@@ -20,10 +21,10 @@ pose = mp_pose.Pose(static_image_mode=True, model_complexity=2)
 
 # Salesforce Configuration (use environment variables in production)
 SF_CONFIG = {
-    "username": "your_salesforce_username@example.com",
-    "password": "your_password",
-    "security_token": "your_security_token",
-    "domain": "login"  # For production: "login" or "test"
+    "username": os.environ.get("SF_USERNAME", "virtualtryonladies493@agentforce.com"),
+    "password": os.environ.get("SF_PASSWORD", "virtual@tryon1"),
+    "security_token": os.environ.get("SF_SECURITY_TOKEN", "your_security_token"),
+    "domain": os.environ.get("SF_DOMAIN", "login")  # For production: "login" or "test"
 }
 
 class SalesforceConnector:
@@ -47,18 +48,18 @@ class SalesforceConnector:
     def is_connected(self):
         return self.sf is not None
     
-    def query_dresses(self):
+    def query_dresses(self, dress_type=None):
         if not self.is_connected():
             self.connect()
             if not self.is_connected():
                 return {}
         
         try:
-            result = self.sf.query(
-                "SELECT Id, Name, Image_URL__c FROM Product2 "
-                "WHERE IsActive = TRUE AND Family = 'Dress' "
-                "LIMIT 20"
-            )
+            query = "SELECT Id, Name, Image_URL__c, Dress_Type__c FROM Product2 WHERE IsActive = TRUE AND Family = 'Dress'"
+            if dress_type:
+                query += f" AND Dress_Type__c = '{dress_type}'"
+            query += " LIMIT 20"
+            result = self.sf.query(query)
             return result.get("records", [])
         except Exception as e:
             logger.error(f"Salesforce query failed: {str(e)}")
@@ -66,10 +67,10 @@ class SalesforceConnector:
 
 sf_connector = SalesforceConnector()
 
-def fetch_dresses():
-    """Fetch dresses with automatic reconnection"""
+def fetch_dresses(dress_type=None):
+    """Fetch dresses with automatic reconnection, filtered by dress type"""
     dresses = {}
-    records = sf_connector.query_dresses()
+    records = sf_connector.query_dresses(dress_type)
     
     for item in records:
         if not item.get('Image_URL__c'):
@@ -77,11 +78,13 @@ def fetch_dresses():
             
         try:
             img_response = requests.get(item['Image_URL__c'], timeout=10)
+            img_response.raise_for_status()
             img = Image.open(BytesIO(img_response.content)).convert("RGBA")
             dresses[item['Id']] = {
                 'name': item['Name'],
                 'image': img,
-                'thumbnail': generate_thumbnail(img)
+                'thumbnail': generate_thumbnail(img),
+                'type': item.get('Dress_Type__c', 'Unknown')
             }
         except Exception as e:
             logger.warning(f"Failed to load dress {item.get('Name')}: {str(e)}")
@@ -96,9 +99,18 @@ def generate_thumbnail(img, size=(150, 200)):
     img.save(buffered, format="PNG")
     return f"data:image/png;base64,{base64.b64encode(buffered.getvalue()).decode()}"
 
-# Rest of your existing functions (get_body_landmarks, warp_and_overlay, etc.)
+# Placeholder for existing functions (replace with your actual implementations)
+def get_body_landmarks(image):
+    """Placeholder for body landmark detection"""
+    # Replace with your actual MediaPipe-based implementation
+    return []  # Return list of landmarks or empty if detection fails
+
+def warp_and_overlay(user_img, dress_img, body_pts):
+    """Placeholder for dress overlay logic"""
+    # Replace with your actual implementation
+    return user_img  # Return processed image
 
-def process_try_on(user_image, dress_id):
+def process_try_on(user_image, dress_id, dress_type):
     """Main processing with enhanced error handling"""
     if user_image is None:
         return None, "Please upload an image"
@@ -115,9 +127,9 @@ def process_try_on(user_image, dress_id):
             return user_image, "Stand facing camera with arms slightly away"
             
         # Get dresses
-        dresses = fetch_dresses()
+        dresses = fetch_dresses(dress_type)
         if not dresses:
-            return user_image, "No dresses available - check Salesforce connection"
+            return user_image, f"No {dress_type} dresses available - check Salesforce connection"
         
         if dress_id not in dresses:
             return user_image, "Selected dress not found"
@@ -138,39 +150,59 @@ with gr.Blocks(title="Virtual Try-On", css=".thumbnail { height: 100px !importan
     with gr.Row():
         with gr.Column():
             input_image = gr.Image(sources=["upload", "webcam"], type="pil", label="Your Photo")
+            dress_type = gr.Dropdown(
+                label="Select Dress Type",
+                choices=["Casual", "Formal"],
+                value="Casual",
+                interactive=True
+            )
             with gr.Row():
                 refresh_btn = gr.Button("🔄 Refresh Dresses")
                 try_btn = gr.Button("👗 Try On Dress", variant="primary")
             dress_dropdown = gr.Dropdown(label="Select Dress", interactive=True)
-            connection_status = gr.Textbox(label="Salesforce Status", 
-                                        value="Connected" if sf_connector.is_connected() else "Disconnected")
+            connection_status = gr.Textbox(
+                label="Salesforce Status",
+                value="Connected" if sf_connector.is_connected() else "Disconnected"
+            )
             
         with gr.Column():
             output_image = gr.Image(label="Your Virtual Try-On", interactive=False)
             status = gr.Textbox(label="Status")
     
-    def update_dress_dropdown():
-        dresses = fetch_dresses()
-        choices = [(f"{d['name']}", id) for id, d in dresses.items()]
+    def update_dress_dropdown(dress_type):
+        dresses = fetch_dresses(dress_type)
+        choices = [(f"{d['name']} ({d['type']})", id) for id, d in dresses.items()]
         status = "Connected" if sf_connector.is_connected() else "Disconnected"
         return (
-            gr.Dropdown(choices=choices, value=choices[0][1] if choices else None),
+            gr.Dropdown.update(
+                choices=choices,
+                value=choices[0][1] if choices else None,
+                label=f"Select Dress ({len(choices)} {dress_type} dresses)"
+            ),
             f"Salesforce: {status} | {len(choices)} dresses loaded"
         )
     
     refresh_btn.click(
-        update_dress_dropdown, 
+        fn=update_dress_dropdown,
+        inputs=dress_type,
+        outputs=[dress_dropdown, connection_status]
+    )
+    
+    dress_type.change(
+        fn=update_dress_dropdown,
+        inputs=dress_type,
         outputs=[dress_dropdown, connection_status]
     )
     
     try_btn.click(
-        process_try_on, 
-        inputs=[input_image, dress_dropdown], 
+        fn=process_try_on,
+        inputs=[input_image, dress_dropdown, dress_type],
         outputs=[output_image, status]
     )
     
     demo.load(
-        update_dress_dropdown, 
+        fn=update_dress_dropdown,
+        inputs=dress_type,
         outputs=[dress_dropdown, connection_status]
     )