Create app.py

app.py

@@ -0,0 +1,183 @@
+import gradio as gr
+import asyncio
+import os
+import glob
+import torch
+from pathlib import Path
+from PIL import Image
+
+# GenAI & ADK Imports
+from google.adk.runners import InMemoryRunner
+from google.genai import types
+
+# Project Imports
+from cellemetry import root_agent
+from cellemetry.config import AnalysisDeps
+from transformers import Sam3Processor, Sam3Model
+
+# --- Global State for Heavy Models ---
+# We load the model once when the app starts to avoid reloading per request.
+MODEL_CACHE = {
+    "model": None,
+    "processor": None,
+    "device": "cpu"
+}
+
+def load_models():
+    """Initialize SAM3 model."""
+    if MODEL_CACHE["model"] is not None:
+        return
+
+    print("--- Loading SAM3 Model ---")
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    MODEL_CACHE["device"] = device
+    
+    try:
+        # Note: Ensure you have access to facebook/sam3 or use a public alternative
+        MODEL_CACHE["model"] = Sam3Model.from_pretrained("facebook/sam3").to(device)
+        MODEL_CACHE["processor"] = Sam3Processor.from_pretrained("facebook/sam3")
+        print(f"✅ SAM3 loaded on {device}")
+    except Exception as e:
+        print(f"⚠️ SAM3 load failed (using mock): {e}")
+        # Allow app to start even if model fails (will fall back to mock logic if implemented)
+
+# Load immediately on startup
+load_models()
+
+# --- Core Analysis Function ---
+async def run_analysis(image_path_str, user_prompt, progress=gr.Progress()):
+    """
+    Async generator that runs the agent and yields updates to the UI.
+    """
+    if not image_path_str:
+        yield "⚠️ Please upload an image first.", None, None
+        return
+
+    # Clean up previous runs
+    for f in glob.glob("/tmp/out_*.png") + glob.glob("/tmp/data_*.npz") + glob.glob("/tmp/*.xlsx"):
+        try:
+            os.remove(f)
+        except:
+            pass
+
+    image_path = Path(image_path_str)
+    
+    # 1. Setup Dependencies
+    deps = AnalysisDeps(
+        sam_model=MODEL_CACHE["model"],
+        sam_processor=MODEL_CACHE["processor"],
+        image_path=image_path,
+        device=MODEL_CACHE["device"],
+        pixel_size_microns=None # Agent will parse this from prompt
+    )
+    
+    # 2. Initialize Runner
+    runner = InMemoryRunner(agent=root_agent, app_name="cellemetry_demo")
+    session = await runner.session_service.create_session(
+        app_name="cellemetry_demo",
+        user_id="demo_user",
+        state=deps.to_state_dict()
+    )
+
+    # 3. Prepare Content
+    image_bytes = image_path.read_bytes()
+    content = types.Content(
+        role="user",
+        parts=[
+            types.Part.from_text(text=user_prompt),
+            types.Part.from_bytes(data=image_bytes, mime_type="image/png"),
+        ]
+    )
+
+    # 4. Stream Execution
+    logs = [f"🚀 Starting analysis on {MODEL_CACHE['device']}..."]
+    log_text = "\n".join(logs)
+    yield log_text, None, None
+
+    async for event in runner.run_async(
+        user_id="demo_user",
+        session_id=session.id,
+        new_message=content,
+    ):
+        author = event.author
+        
+        # Capture Tool Calls
+        if event.get_function_calls():
+            for fc in event.get_function_calls():
+                logs.append(f"🔧 **{author}** calling tool: `{fc.name}`")
+        
+        # Capture Text (Streaming)
+        if event.content and event.content.parts:
+            for part in event.content.parts:
+                if hasattr(part, 'text') and part.text:
+                    if event.partial:
+                        # Update the last log line if it's the same thought
+                        if logs[-1].startswith(f"💬 **{author}**"):
+                            logs[-1] = f"💬 **{author}**: {part.text}..."
+                        else:
+                            logs.append(f"💬 **{author}**: {part.text}...")
+                    else:
+                        logs.append(f"✅ **{author}**: {part.text}")
+        
+        # Yield updated logs immediately
+        yield "\n\n".join(logs), None, None
+
+    # 5. Retrieve Final Results
+    logs.append("\n🏁 **Analysis Complete.** gathering files...")
+    yield "\n\n".join(logs), None, None
+
+    # Collect output images
+    output_images = glob.glob("/tmp/out_*.png")
+    
+    # Collect excel report
+    excel_files = glob.glob("/tmp/*.xlsx")
+    report_file = excel_files[0] if excel_files else None
+
+    logs.append(f"\n📋 Found {len(output_images)} segmentation maps and {1 if report_file else 0} report.")
+    yield "\n\n".join(logs), output_images, report_file
+
+
+# --- Gradio UI Layout ---
+with gr.Blocks(title="Cellemetry Agent", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🔬 Cellemetry: Agentic Microscopy Analysis")
+    gr.Markdown("Upload a microscopy image and ask the agent to identify, segment, and quantify biological structures.")
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            # Input Section
+            img_input = gr.Image(type="filepath", label="Microscopy Image", height=300)
+            
+            prompt_input = gr.Textbox(
+                label="Analysis Request", 
+                lines=3, 
+                value="Identify the green irregular cells and blue round nuclei. Provide a statistical report on morphology and density.",
+                placeholder="E.g., 'Find all red cells and calculate their density.'"
+            )
+            
+            with gr.Accordion("Advanced Settings", open=False):
+                gr.Markdown(f"Running on: **{MODEL_CACHE['device']}**")
+            
+            run_btn = gr.Button("🧪 Run Analysis", variant="primary", size="lg")
+
+        with gr.Column(scale=2):
+            # Output Section
+            with gr.Tabs():
+                with gr.Tab("Live Agent Logs"):
+                    # Markdown component to render bolding/formatting in logs
+                    log_output = gr.Markdown(label="Agent Thought Process", height=500)
+                
+                with gr.Tab("Visual Results"):
+                    gallery = gr.Gallery(label="Segmentation Maps", columns=2)
+                
+                with gr.Tab("Data Report"):
+                    file_output = gr.File(label="Download Excel Report")
+
+    # Connect the Async Function
+    run_btn.click(
+        fn=run_analysis,
+        inputs=[img_input, prompt_input],
+        outputs=[log_output, gallery, file_output]
+    )
+
+if __name__ == "__main__":
+    demo.queue().launch()