import os import sys import json import random import shutil import hashlib import uuid from typing import List import base64 from io import BytesIO import time import threading import numpy as np import torch import torch.nn as nn from PIL import Image, ImageOps from matplotlib import cm import cv2 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 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 torch.manual_seed(42); np.random.seed(42); random.seed(42) 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__() self.image_model = DenseNet121Classifier(num_classes=num_classes) self.text_model = TextClassifier(num_classes=num_classes) self.fusion = nn.Sequential( nn.Linear(num_classes * 2, 128), nn.ReLU(), nn.Dropout(dropout), nn.Linear(128, num_classes) ) def forward(self, image, input_ids, attention_mask): logits_img = self.image_model(image) logits_txt = self.text_model(input_ids, attention_mask) 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) fusion_model.load_state_dict(torch.load(FUSION_WEIGHTS_PATH, map_location=device)) fusion_model.eval() print("✅ AI Model loaded successfully!") 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() app = FastAPI() app.mount("/static", StaticFiles(directory="static"), name="static") templates = Jinja2Templates(directory="templates") os.makedirs("uploads", exist_ok=True) EXPIRATION_MINUTES = 10 results_cache = {} cache_lock = threading.Lock() def cleanup_expired_cache(): """ ฟังก์ชันนี้จะทำงานใน Background Thread เพื่อตรวจสอบและลบ Cache ที่หมดอายุ """ while True: 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) # ตรวจสอบทุกๆ 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.") SYMPTOM_MAP = { "noSymptoms": "ไม่มีอาการ", "drinkAlcohol": "ดื่มเหล้า", "smoking": "สูบบุหรี่", "chewBetelNut": "เคี้ยวหมาก", "eatSpicyFood": "กินเผ็ดแสบ", "wipeOff": "เช็ดออกได้", "alwaysHurts": "เจ็บเมื่อโดนแผล" } def process_with_ai_model(image_path: str, prompt_text: str): 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("") ): temp_filepath = os.path.join("uploads", f"{uuid.uuid4()}_{file.filename}") with open(temp_filepath, "wb") as buffer: shutil.copyfileobj(file.file, buffer) 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, } with cache_lock: results_cache[result_id] = { "data": result_data, "created_at": time.time() } 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: del results_cache[result_id] return RedirectResponse(url="/detect") context = {"request": request, **cached_item["data"]} return templates.TemplateResponse("detect.html", context) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)