Update app.py

app.py

@@ -1,39 +1,145 @@
+import os
+import gc
+import torch
 from transformers import pipeline
 import gradio as gr
 from PIL import Image
+import requests
+from io import BytesIO
+import psutil
+from datetime import datetime
 
-# 建立 pipeline
-pipe = pipeline("image-to-text", model="google/medgemma-4b-it")
+# 設定環境變數，使用臨時目錄避免快速填滿存儲
+os.environ["TRANSFORMERS_CACHE"] = "/tmp/transformers_cache"
+os.environ["HF_HOME"] = "/tmp/hf_home"
+os.environ["TORCH_HOME"] = "/tmp/torch_cache"
 
-def simple_predict(image, question):
-    """
-    最簡單的處理函數 - 用於調試
-    """
+def clear_memory():
+    """清理記憶體和快取"""
+    gc.collect()
+    if torch.cuda.is_available():
+        torch.cuda.empty_cache()
+
+def check_storage():
+    """檢查存儲空間"""
+    try:
+        disk_usage = psutil.disk_usage('/')
+        free_gb = disk_usage.free / (1024**3)
+        used_percent = (disk_usage.used / disk_usage.total) * 100
+        return free_gb, used_percent
+    except:
+        return 0, 100
+
+def load_medgemma_model():
+    """載入 MedGemma 模型，使用優化設定"""
+    try:
+        print("🏥 正在載入 MedGemma-4B 模型...")
+        print(f"⏰ 載入時間: {datetime.now().strftime('%H:%M:%S')}")
+        
+        # 檢查存儲空間
+        free_gb, used_percent = check_storage()
+        print(f"💾 可用空間: {free_gb:.1f}GB, 使用率: {used_percent:.1f}%")
+        
+        if free_gb < 5:  # 如果可用空間少於5GB
+            raise Exception(f"存儲空間不足 ({free_gb:.1f}GB)，建議至少需要 5GB")
+        
+        # 使用優化設定載入模型
+        pipe = pipeline(
+            "image-to-text",
+            model="google/medgemma-4b-it",
+            torch_dtype=torch.float16,  # 使用半精度節省記憶體
+            device_map="auto",
+            low_cpu_mem_usage=True,
+            cache_dir="/tmp/transformers_cache"
+        )
+        
+        print("✅ MedGemma-4B 模型載入成功！")
+        return pipe, "google/medgemma-4b-it"
+        
+    except Exception as e:
+        print(f"❌ MedGemma 載入失敗: {e}")
+        print("🔄 嘗試載入較小的替代模型...")
+        
+        try:
+            # 載入較小的醫療相關模型作為替代
+            pipe = pipeline(
+                "image-to-text",
+                model="Salesforce/blip-image-captioning-base",
+                cache_dir="/tmp/transformers_cache"
+            )
+            print("✅ 已載入 BLIP 模型作為替代")
+            return pipe, "Salesforce/blip-image-captioning-base"
+        except Exception as e2:
+            raise Exception(f"所有模型載入失敗: MedGemma({e}), BLIP({e2})")
+
+def load_image_from_input(image_input):
+    """處理圖片輸入：PIL Image、檔案路徑或 URL"""
     try:
-        print(f"圖片類型: {type(image)}")
-        print(f"問題: {question}")
+        # JPG 檔案上傳（Gradio 返回 PIL Image）
+        if isinstance(image_input, Image.Image):
+            return image_input
         
-        if image is None:
-            return "請上傳圖片"
+        # URL 輸入
+        elif isinstance(image_input, str):
+            if image_input.startswith(("http://", "https://")):
+                print(f"📥 正在下載圖片: {image_input[:50]}...")
+                response = requests.get(image_input, timeout=10)
+                response.raise_for_status()
+                image = Image.open(BytesIO(response.content))
+                print("✅ 圖片下載成功")
+                return image
+            else:
+                # 檔案路徑
+                return Image.open(image_input)
+        else:
+            return Image.open(image_input)
+            
+    except Exception as e:
+        raise Exception(f"無法載入圖片: {e}")
+
+def predict(image_input, question, url_input):
+    """主要預測函數"""
+    try:
+        # 確定圖片來源（優先使用上傳的圖片）
+        if image_input is not None:
+            image_source = image_input
+            source_type = "上傳檔案"
+        elif url_input and url_input.strip():
+            image_source = url_input.strip()
+            source_type = "URL"
+        else:
+            return "❌ 請上傳圖片或輸入圖片 URL"
         
-        if not question:
-            question = "請描述這張圖片"
+        print(f"📷 處理圖片來源: {source_type}")
         
-        # 確保是 RGB 模式
+        # 載入圖片
+        image = load_image_from_input(image_source)
+        
+        # 圖片預處理
+        original_size = image.size
         if image.mode != 'RGB':
             image = image.convert('RGB')
+            print(f"🔄 轉換圖片格式: {image.mode}")
         
-        # 嘗試最簡單的調用方式
-        print("嘗試直接傳遞圖片...")
-        result = pipe(image)
-        print(f"直接傳遞結果: {result}")
+        # 調整圖片大小以節省記憶體（保持品質）
+        max_size = 768  # MedGemma 建議大小
+        if max(image.size) > max_size:
+            ratio = max_size / max(image.size)
+            new_size = tuple(int(dim * ratio) for dim in image.size)
+            image = image.resize(new_size, Image.Resampling.LANCZOS)
+            print(f"📐 調整圖片大小: {original_size} → {image.size}")
         
-        return str(result)
+        # 處理問題輸入
+        if not question or not question.strip():
+            question = "請詳細分析這張醫療影像，描述你看到的重要特徵、可能的病理變化，以及任何需要注意的異常。"
         
-    except Exception as e:
-        print(f"直接傳遞失敗，嘗試消息格式...")
-        try:
-            # 嘗試消息格式
+        question = question.strip()
+        print(f"❓ 醫療問題: {question[:100]}...")
+        
+        # 根據模型類型選擇輸入格式
+        global model_name
+        if "medgemma" in model_name.lower():
+            # MedGemma 使用對話格式
             messages = [
                 {
                     "role": "user",
@@ -43,23 +149,180 @@ def simple_predict(image, question):
                     ]
                 }
             ]
+            print("🔬 使用 MedGemma 專業醫療分析模式")
             result = pipe(messages)
-            print(f"消息格式結果: {result}")
-            return str(result)
-        except Exception as e2:
-            return f"兩種方式都失敗了:\n直接傳遞: {str(e)}\n消息格式: {str(e2)}"
+        else:
+            # 其他模型直接使用圖片
+            print("🔍 使用通用圖片描述模式")
+            result = pipe(image)
+        
+        # 清理記憶體
+        clear_memory()
+        
+        # 解析結果
+        if isinstance(result, list) and len(result) > 0:
+            if isinstance(result[0], dict):
+                generated_text = result[0].get('generated_text', str(result[0]))
+            else:
+                generated_text = str(result[0])
+        else:
+            generated_text = str(result)
+        
+        # 添加分析資訊
+        analysis_info = f"""
+🏥 **醫療影像分析結果**
+
+📊 **圖片資訊:**
+- 原始尺寸: {original_size}
+- 處理尺寸: {image.size}
+- 來源: {source_type}
+
+🤖 **使用模型:** {model_name}
+
+🔬 **分析結果:**
+{generated_text}
+
+---
+⚠️ **重要提醒:** 此分析僅供參考，不能替代專業醫療診斷。如有疑慮請諮詢專業醫師。
+"""
+        
+        return analysis_info
+        
+    except Exception as e:
+        clear_memory()
+        error_msg = f"❌ 處理錯誤: {str(e)}"
+        print(error_msg)
+        return error_msg
+
+# 載入模型
+try:
+    pipe, model_name = load_medgemma_model()
+    model_status = f"✅ {model_name} 已準備就緒"
+except Exception as e:
+    model_status = f"❌ 模型載入失敗: {e}"
+    pipe = None
+    model_name = "未載入"
 
-# 最簡單的介面
-demo = gr.Interface(
-    fn=simple_predict,
-    inputs=[
-        gr.Image(type="pil", file_types=["jpg", "jpeg"]),
-        gr.Textbox(value="請分析這張醫療影像", lines=2)
-    ],
-    outputs=gr.Textbox(lines=10),
-    title="MedGemma 調試版",
-    description="上傳 JPG 圖片測試 MedGemma 模型"
-)
+# 創建 Gradio 介面
+def create_interface():
+    with gr.Blocks(
+        title="MedGemma 醫療影像分析系統",
+        theme=gr.themes.Soft(),
+        css=".gradio-container {max-width: 1200px; margin: auto;}"
+    ) as demo:
+        
+        gr.Markdown(f"""
+# 🏥 MedGemma 醫療影像分析系統
+        
+**模型狀態:** {model_status}  
+**更新時間:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
+        
+上傳醫療影像（JPG/PNG）或輸入圖片 URL，獲得專業的 AI 醫療影像分析。
+        """)
+        
+        with gr.Row():
+            with gr.Column(scale=1):
+                # 圖片上傳
+                image_input = gr.Image(
+                    label="📤 上傳醫療影像",
+                    type="pil",
+                    file_types=["jpg", "jpeg", "png"],
+                    height=300
+                )
+                
+                # URL 輸入
+                url_input = gr.Textbox(
+                    label="🔗 或輸入圖片 URL",
+                    placeholder="https://example.com/medical-image.jpg",
+                    lines=1
+                )
+                
+                # 問題輸入
+                question_input = gr.Textbox(
+                    label="❓ 醫療問題或分析要求",
+                    placeholder="請分析這張X光片中的異常...",
+                    lines=3,
+                    value="請詳細分析這張醫療影像，包括任何可見的異常或重要特徵。"
+                )
+                
+                # 分析按鈕
+                analyze_btn = gr.Button(
+                    "🔬 開始分析", 
+                    variant="primary",
+                    size="lg"
+                )
+                
+                # 清理按鈕
+                clear_btn = gr.Button("🧹 清理", variant="secondary")
+            
+            with gr.Column(scale=2):
+                # 分析結果
+                output = gr.Textbox(
+                    label="📋 分析結果",
+                    lines=20,
+                    interactive=False,
+                    show_copy_button=True
+                )
+        
+        # 使用說明
+        with gr.Accordion("📖 使用說明", open=False):
+            gr.Markdown("""
+### 如何使用:
+1. **上傳圖片**: 點擊上傳區域選擇 JPG/PNG 醫療影像
+2. **或使用 URL**: 在 URL 欄位貼上圖片連結
+3. **輸入問題**: 描述你想了解的醫療問題
+4. **開始分析**: 點擊分析按鈕獲得結果
+
+### 支援的影像類型:
+- X光片 (X-ray)
+- CT 掃描 (CT Scan) 
+- MRI 影像 (MRI)
+- 超音波影像 (Ultrasound)
+- 病理切片 (Pathology)
+
+### 重要提醒:
+⚠️ 此 AI 分析僅供參考學習，不可作為醫療診斷依據
+⚠️ 如有健康疑慮，請務必諮詢專業醫師
+            """)
+        
+        # 事件綁定
+        analyze_btn.click(
+            fn=predict,
+            inputs=[image_input, question_input, url_input],
+            outputs=output
+        )
+        
+        clear_btn.click(
+            fn=lambda: ("", "", ""),
+            outputs=[image_input, url_input, output]
+        )
+        
+        # 圖片上傳時自動分析
+        image_input.change(
+            fn=lambda img, q, url: predict(img, q, url) if img is not None else "",
+            inputs=[image_input, question_input, url_input],
+            outputs=output
+        )
+    
+    return demo
 
+# 啟動應用
 if __name__ == "__main__":
-    demo.launch(debug=True)
+    if pipe is None:
+        print("❌ 無法啟動：模型載入失敗")
+        exit(1)
+    
+    print("🚀 啟動 MedGemma 醫療影像分析系統...")
+    
+    # 檢查最終狀態
+    free_gb, used_percent = check_storage()
+    print(f"💾 當前存儲狀態: {free_gb:.1f}GB 可用, {used_percent:.1f}% 已使用")
+    
+    demo = create_interface()
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        debug=False,
+        show_error=True,
+        share=False
+    )