modifymain

main.py

@@ -21,14 +21,13 @@ from fastapi import FastAPI, File, UploadFile, Form, Request, Depends
 from fastapi.responses import HTMLResponse, RedirectResponse
 from fastapi.templating import Jinja2Templates
 from fastapi.staticfiles import StaticFiles
-import requests
-import uvicorn
 
 sys.path.append(os.path.abspath(os.path.dirname(__file__)))
 from models.densenet.preprocess.preprocessingwangchan import get_tokenizer, get_transforms
 from models.densenet.train_densenet_only import DenseNet121Classifier
 from models.densenet.train_text_only import TextClassifier
-
+import requests
+import uvicorn
 
 HF_MODEL_URL = "https://huggingface.co/qqqqqqat/densenet_wangchan/resolve/main/best_fusion_densenet.pth"
 LOCAL_MODEL_PATH = "models/densenet/best_fusion_densenet.pth"
@@ -50,13 +49,10 @@ FUSION_LABELMAP_PATH = "models/densenet/label_map_fusion_densenet.json"
 FUSION_WEIGHTS_PATH  = "models/densenet/best_fusion_densenet.pth"
 with open(FUSION_LABELMAP_PATH, "r", encoding="utf-8") as f:
     label_map = json.load(f)
-
 class_names = [label for label, _ in sorted(label_map.items(), key=lambda x: x[1])]
 NUM_CLASSES = len(class_names)
 device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 print(f"🧠 Using device: {device}")
-
-
 class FusionDenseNetText(nn.Module):
     def __init__(self, num_classes, dropout=0.3):
         super().__init__()
@@ -72,11 +68,9 @@ class FusionDenseNetText(nn.Module):
         fused_in   = torch.cat([logits_img, logits_txt], dim=1)
         fused_out  = self.fusion(fused_in)
         return fused_out, logits_img, logits_txt
-
-
 print("🔄 Loading AI model...")
 fusion_model = FusionDenseNetText(num_classes=NUM_CLASSES).to(device)
-
+# ดาวน์โหลดก่อน
 download_model_if_needed()
 
 fusion_model = FusionDenseNetText(num_classes=NUM_CLASSES).to(device)
@@ -84,55 +78,38 @@ fusion_model.load_state_dict(torch.load(LOCAL_MODEL_PATH, map_location=device))
 fusion_model.eval()
 print("✅ AI Model loaded successfully!")
 fusion_model.eval()
-
 tokenizer = get_tokenizer()
 transform = get_transforms((224, 224))
-
-
 def _find_last_conv2d(mod: torch.nn.Module):
     last = None
     for m in mod.modules():
         if isinstance(m, torch.nn.Conv2d): last = m
     return last
-
-
 def compute_gradcam_overlay(img_pil, image_tensor, target_class_idx):
     img_branch = fusion_model.image_model
     target_layer = _find_last_conv2d(img_branch)
     if target_layer is None: return None
-
     activations, gradients = [], []
-
     def fwd_hook(_m, _i, o): activations.append(o)
     def bwd_hook(_m, gin, gout): gradients.append(gout[0])
-
     h1 = target_layer.register_forward_hook(fwd_hook)
     h2 = target_layer.register_full_backward_hook(bwd_hook)
-
     try:
         img_branch.zero_grad()
         logits_img = img_branch(image_tensor)
         score = logits_img[0, target_class_idx]
         score.backward()
-
         act = activations[-1].detach()[0]
         grad = gradients[-1].detach()[0]
-
         weights = torch.mean(grad, dim=(1, 2))
         cam = torch.relu(torch.sum(weights[:, None, None] * act, dim=0))
-
         cam -= cam.min(); cam /= (cam.max() + 1e-8)
-
         cam_img = Image.fromarray((cam.cpu().numpy() * 255).astype(np.uint8)).resize(img_pil.size, Image.BILINEAR)
-
         cam_np = np.asarray(cam_img).astype(np.float32) / 255.0
         heatmap = cm.get_cmap("jet")(cam_np)[:, :, :3]
-
         img_np = np.asarray(img_pil.convert("RGB")).astype(np.float32) / 255.0
         overlay = (0.6 * img_np + 0.4 * heatmap)
-
         return np.clip(overlay * 255, 0, 255).astype(np.uint8)
-    
     finally:
         h1.remove(); h2.remove(); img_branch.zero_grad()
 
@@ -142,191 +119,129 @@ app.mount("/static", StaticFiles(directory="static"), name="static")
 templates = Jinja2Templates(directory="templates")
 os.makedirs("uploads", exist_ok=True)
 
-
-# ===== Cache Cleanup Thread =====
-EXPIRATION_MINUTES = 10
-results_cache = {}
-cache_lock = threading.Lock()
-
+EXPIRATION_MINUTES = 10 
+results_cache = {}       
+cache_lock = threading.Lock() 
 
 def cleanup_expired_cache():
+    """
+    ฟังก์ชันนี้จะทำงานใน Background Thread เพื่อตรวจสอบและลบ Cache ที่หมดอายุ
+    """
     while True:
-        with cache_lock:
+        with cache_lock: # ล็อคเพื่อความปลอดภัยในการเข้าถึง cache
+            # สร้าง list ของ key ที่จะลบ เพื่อไม่ให้แก้ไข dict ขณะวน loop
             expired_keys = []
             current_time = time.time()
             for key, value in results_cache.items():
                 if current_time - value["created_at"] > EXPIRATION_MINUTES * 60:
                     expired_keys.append(key)
-
+            
+            # ลบ key ที่หมดอายุ
             for key in expired_keys:
                 del results_cache[key]
                 print(f"🧹 Cache expired and removed for key: {key}")
-
-        time.sleep(60)
-
+        
+        time.sleep(60) # ตรวจสอบทุกๆ 60 วินาที
 
 @app.on_event("startup")
 async def startup_event():
+    """
+    เริ่ม Background Thread สำหรับทำความสะอาด Cache เมื่อแอปเริ่มทำงาน
+    """
     cleanup_thread = threading.Thread(target=cleanup_expired_cache, daemon=True)
     cleanup_thread.start()
     print("🗑️  Cache cleanup task started.")
 
-
-# ===============================
-#      >>> TURNSTILE VERIFY <<<
-# ===============================
-
-TURNSTILE_SECRET = "0x4AAAAAACEfyIeAjGlYCXeasGsMxTuTlHU"   ### (เพิ่ม)
+SYMPTOM_MAP = {
+    "noSymptoms": "ไม่มีอาการ", "drinkAlcohol": "ดื่มเหล้า", "smoking": "สูบบุหรี่",
+    "chewBetelNut": "เคี้ยวหมาก", "eatSpicyFood": "กินเผ็ดแสบ", "wipeOff": "เช็ดออกได้",
+    "alwaysHurts": "เจ็บเมื่อโดนแผล"
+}
 def process_with_ai_model(image_path: str, prompt_text: str):
-    """ประมวลผลภาพด้วย DenseNet + Text model และสร้างผลลัพธ์"""
-
-    # --- Load image ---
-    img_pil = Image.open(image_path).convert("RGB")
-    img_tensor = transform(img_pil).unsqueeze(0).to(device)
-
-    # --- Tokenize text prompt ---
-    encoding = tokenizer(
-        prompt_text,
-        padding="max_length",
-        truncation=True,
-        max_length=128,
-        return_tensors="pt"
-    )
-    input_ids = encoding["input_ids"].to(device)
-    attention_mask = encoding["attention_mask"].to(device)
-
-    # --- Forward pass ---
-    with torch.no_grad():
-        fused_out, logits_img, logits_txt = fusion_model(
-            img_tensor, input_ids, attention_mask
-        )
-        probs = torch.softmax(fused_out, dim=1)[0]
-        pred_idx = torch.argmax(probs).item()
-
-    predicted_label = class_names[pred_idx]
-    confidence = probs[pred_idx].item()
-
-    name_out = f"{predicted_label} ({confidence:.2f})"
-
-    # --- Compute GradCAM ---
-    overlay_np = compute_gradcam_overlay(img_pil, img_tensor, pred_idx)
-    if overlay_np is not None:
-        overlay_img = Image.fromarray(overlay_np)
-        buffered = BytesIO()
-        overlay_img.save(buffered, format="PNG")
-        gradcam_b64 = base64.b64encode(buffered.getvalue()).decode()
-    else:
-        gradcam_b64 = None
-
-    # --- Encode original image ---
-    buffer2 = BytesIO()
-    img_pil.save(buffer2, format="PNG")
-    image_b64 = base64.b64encode(buffer2.getvalue()).decode()
-
-    return image_b64, gradcam_b64, predicted_label, confidence
-
+    try:
+        image_pil = Image.open(image_path)
+        image_pil = ImageOps.exif_transpose(image_pil)
+        image_pil = image_pil.convert("RGB")
+        image_tensor = transform(image_pil).unsqueeze(0).to(device)
+        enc = tokenizer(prompt_text, return_tensors="pt", padding="max_length",
+                        truncation=True, max_length=128)
+        ids, mask = enc["input_ids"].to(device), enc["attention_mask"].to(device)
+        with torch.no_grad():
+            fused_logits, _, _ = fusion_model(image_tensor, ids, mask)
+            probs_fused = torch.softmax(fused_logits, dim=1)[0].cpu().numpy()
+        pred_idx   = int(np.argmax(probs_fused))
+        pred_label = class_names[pred_idx]
+        confidence = float(probs_fused[pred_idx]) * 100
+        gradcam_overlay_np = compute_gradcam_overlay(image_pil, image_tensor, pred_idx)
+        def image_to_base64(img):
+            buffered = BytesIO()
+            img.save(buffered, format="JPEG")
+            return base64.b64encode(buffered.getvalue()).decode('utf-8')
+        original_b64 = image_to_base64(image_pil)
+        if gradcam_overlay_np is not None:
+            gradcam_pil = Image.fromarray(gradcam_overlay_np)
+            gradcam_b64 = image_to_base64(gradcam_pil)
+        else:
+            gradcam_b64 = original_b64
+        return original_b64, gradcam_b64, pred_label, f"{confidence:.2f}"
+    except Exception as e:
+        print(f"❌ Error during AI processing: {e}")
+        return None, None, "Error", "0.00"
+
+@app.get("/", response_class=RedirectResponse)
+async def root():
+    return RedirectResponse(url="/detect")
+@app.get("/detect", response_class=HTMLResponse)
+async def show_upload_form(request: Request):
+    return templates.TemplateResponse("detect.html", {"request": request})
 
 @app.post("/uploaded")
 async def handle_upload(
     request: Request,
     file: UploadFile = File(...),
     checkboxes: List[str] = Form([]),
-    symptom_text: str = Form(""),
-    cf_turnstile_response: str = Form("")   ### (เพิ่ม)
+    symptom_text: str = Form("")
 ):
-
-    # ---------- (เพิ่ม) VERIFY TURNSTILE ---------
-    
-    if not cf_turnstile_response:
-        return templates.TemplateResponse(
-            "detect.html",
-            {"request": request, "error": "Turnstile token missing"}
-        )
-
-    verify_resp = requests.post(
-        "https://challenges.cloudflare.com/turnstile/v0/siteverify",
-        data={
-            "secret": TURNSTILE_SECRET,
-            "response": cf_turnstile_response
-        }
-    )
-
-    cf_result = verify_resp.json()
-
-    if not cf_result.get("success"):
-        return templates.TemplateResponse(
-            "detect.html",
-            {"request": request, "error": "การยืนยันความปลอดภัยไม่สำเร็จ กรุณาลองใหม่อีกครั้ง"}
-        )
-
-    # ------------------------------------------------
-
-
-    # >>> โค้ดเดิมของคุณทั้งหมดด้านล่าง (ไม่แก้) <<<
-
     temp_filepath = os.path.join("uploads", f"{uuid.uuid4()}_{file.filename}")
     with open(temp_filepath, "wb") as buffer:
         shutil.copyfileobj(file.file, buffer)
-
-    SYMPTOM_MAP = {
-        "noSymptoms": "ไม่มีอาการ", "drinkAlcohol": "ดื่มเหล้า", "smoking": "สูบบุหรี่",
-        "chewBetelNut": "เคี้ยวหมาก", "eatSpicyFood": "กินเผ็ดแสบ", "wipeOff": "เช็ดออกได้",
-        "alwaysHurts": "เจ็บเมื่อโดนแผล"
-    }
-
     final_prompt_parts = []
     selected_symptoms_thai = {SYMPTOM_MAP.get(cb) for cb in checkboxes if SYMPTOM_MAP.get(cb)}
-
     if "ไม่มีอาการ" in selected_symptoms_thai:
         symptoms_group   = {"เจ็บเมื่อโดนแผล", "กินเผ็ดแสบ"}
         lifestyles_group = {"ดื่มเหล้า", "สูบบุหรี่", "เคี้ยวหมาก"}
         patterns_group   = {"เช็ดออกได้"}
         special_group    = {"ไม่มีอาการ"}
-
         final_selected = (selected_symptoms_thai - symptoms_group) | \
                          (selected_symptoms_thai & (lifestyles_group | patterns_group | special_group))
-
         final_prompt_parts.append(" ".join(sorted(list(final_selected))))
-
     elif selected_symptoms_thai:
         final_prompt_parts.append(" ".join(sorted(list(selected_symptoms_thai))))
-
     if symptom_text and symptom_text.strip():
         final_prompt_parts.append(symptom_text.strip())
-
     final_prompt = "; ".join(final_prompt_parts) if final_prompt_parts else "ไม่มีอาการ"
-
     image_b64, gradcam_b64, name_out, eva_output = process_with_ai_model(
         image_path=temp_filepath, prompt_text=final_prompt
     )
-
     os.remove(temp_filepath)
-
     result_id = str(uuid.uuid4())
-
     result_data = {
-        "image_b64_data": image_b64,
-        "gradcam_b64_data": gradcam_b64,
-        "name_out": name_out,
-        "eva_output": eva_output,
+        "image_b64_data": image_b64, "gradcam_b64_data": gradcam_b64,
+        "name_out": name_out, "eva_output": eva_output,
     }
-
     with cache_lock:
         results_cache[result_id] = {
             "data": result_data,
-            "created_at": time.time()
+            "created_at": time.time() 
         }
 
-    results_url = request.url_for("show_results", result_id=result_id)
+    results_url = request.url_for('show_results', result_id=result_id)
     return RedirectResponse(url=results_url, status_code=303)
 
-
-
 @app.get("/results/{result_id}", response_class=HTMLResponse)
 async def show_results(request: Request, result_id: str):
     with cache_lock:
         cached_item = results_cache.get(result_id)
-
     if not cached_item or (time.time() - cached_item["created_at"] > EXPIRATION_MINUTES * 60):
         if cached_item:
             with cache_lock:
@@ -336,8 +251,6 @@ async def show_results(request: Request, result_id: str):
     context = {"request": request, **cached_item["data"]}
     return templates.TemplateResponse("detect.html", context)
 
-
-
 if __name__ == "__main__":
-    port = int(os.environ.get("PORT", 8000))
-    uvicorn.run(app, host="0.0.0.0", port=port)
+    port = int(os.environ.get("PORT", 8000))   # ใช้ PORT ของ hosting ถ้ามี
+    uvicorn.run(app, host="0.0.0.0", port=port)