Create app.py

app.py

@@ -0,0 +1,115 @@
+import os
+import io
+import base64
+import time
+import logging
+from typing import Optional
+
+import gradio as gr
+from gradio_client import Client
+from PIL import Image
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+HF_TOKEN = os.getenv("HF_TOKEN")
+if not HF_TOKEN:
+    raise ValueError("HF_TOKEN environment variable is required")
+
+backend_status = {
+    "client": None,
+    "connected": False,
+    "last_check": None,
+    "error_message": ""
+}
+
+def check_backend_connection():
+    """Check connection to SAKS backend"""
+    try:
+        test_client = Client("SnapwearAI/SAKS-backend", hf_token=HF_TOKEN)
+        backend_status.update({
+            "client": test_client,
+            "connected": True,
+            "error_message": "",
+            "last_check": time.time(),
+        })
+        logger.info("✅ SAKS Backend connection established")
+        return True, "🟢 Backend is ready"
+    except Exception as e:
+        backend_status.update({
+            "client": None,
+            "connected": False,
+            "last_check": time.time(),
+            "error_message": str(e),
+        })
+        err = str(e).lower()
+        if "timeout" in err or "read operation timed out" in err:
+            return False, "🟡 Backend starting up..."
+        return False, f"🔴 Backend error: {e}"
+
+check_backend_connection()
+
+def image_to_base64(image: Image.Image) -> str:
+    if image is None:
+        return ""
+    if image.mode != "RGB":
+        image = image.convert("RGB")
+    buf = io.BytesIO()
+    image.save(buf, format="PNG")
+    return base64.b64encode(buf.getvalue()).decode()
+
+def base64_to_image(b64: str) -> Optional[Image.Image]:
+    if not b64:
+        return None
+    try:
+        return Image.open(io.BytesIO(base64.b64decode(b64))).convert("RGB")
+    except Exception as e:
+        logger.error(f"Failed to decode base64: {e}")
+        return None
+
+def process_image(input_image: Image.Image):
+    """Process image through SAKS backend"""
+    if input_image is None:
+        return None
+    
+    try:
+        if not backend_status["connected"]:
+            check_backend_connection()
+            if not backend_status["connected"]:
+                return None
+        
+        client = backend_status["client"]
+        img_b64 = image_to_base64(input_image)
+        
+        result = client.predict(
+            img_b64,
+            api_name="/predict",
+        )
+        
+        if not result or len(result) < 1:
+            return None
+        
+        output_b64 = result[0]
+        return base64_to_image(output_b64)
+        
+    except Exception as e:
+        logger.error(f"Processing error: {e}")
+        return None
+
+with gr.Blocks(title="SAKS") as demo:
+    gr.Markdown("# SAKS Jewelry Detection")
+    
+    with gr.Row():
+        input_img = gr.Image(label="Input Image", type="pil")
+        output_img = gr.Image(label="Output Image")
+    
+    process_btn = gr.Button("Process", variant="primary")
+    
+    process_btn.click(
+        fn=process_image,
+        inputs=input_img,
+        outputs=output_img
+    )
+
+if __name__ == "__main__":
+    demo.launch()