853 lines
34 KiB
Python
853 lines
34 KiB
Python
from fastapi import FastAPI, File, UploadFile, HTTPException, Query, Request, Response
|
||
from pydantic import BaseModel
|
||
from fastapi.responses import JSONResponse, FileResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
import os
|
||
import uuid
|
||
import numpy as np
|
||
from PIL import Image, ImageDraw
|
||
import torch
|
||
import torch.nn as nn
|
||
import torch.nn.functional as F
|
||
import torchvision.models as models
|
||
from torchvision import transforms
|
||
import uvicorn
|
||
from pathlib import Path
|
||
import cv2
|
||
import io
|
||
import base64
|
||
from datetime import datetime
|
||
import re
|
||
|
||
# Import custom models
|
||
import sys
|
||
sys.path.append('.')
|
||
from arch.efficientfeedback import EfficientFeedbackNetwork
|
||
from arch.unet3plus_att import UNet3Plus_Attention
|
||
from pdf_service import generate_medical_report
|
||
|
||
# Configuration
|
||
UPLOAD_FOLDER = 'uploads'
|
||
RESULTS_FOLDER = 'results'
|
||
TEMPLATES_FOLDER = 'templates'
|
||
|
||
# for folder in [UPLOAD_FOLDER, RESULTS_FOLDER, TEMPLATES_FOLDER]:
|
||
# os.makedirs(folder, exist_ok=True)
|
||
os.makedirs(TEMPLATES_FOLDER, exist_ok=True)
|
||
|
||
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
|
||
|
||
# Classes
|
||
ANGLE_CLASSES = ['med-lat', 'post-trans', 'sup-trans-flex', 'sup-up-long']
|
||
SEGMENT_CLASSES_SUPRAPAT = {0: "background", 1: "effusion", 2: "fat", 3: "fat-pat", 4: "femur", 5: "synovium", 6: "tendon"}
|
||
SEGMENT_CLASSES_POST = {0: "background", 1: "fat", 2: "tendon", 3: "muscle", 4: "femur", 5: "artery", 6: "baker's cyst"}
|
||
|
||
# Color map for Suprapat
|
||
COLOR_MAP_SUP = {
|
||
'background': {'color': [0, 0, 0], 'name': 'Nền'},
|
||
'effusion': {'color': [255, 0, 0], 'name': 'Dịch khớp'},
|
||
'fat': {'color': [255, 255, 0], 'name': 'Mỡ'},
|
||
'fat-pat': {'color': [0, 255, 255], 'name': 'Mỡ Hoffa'},
|
||
'femur': {'color': [0, 255, 0], 'name': 'Xương đùi'},
|
||
'synovium': {'color': [255, 0, 255], 'name': 'Màng hoạt dịch'},
|
||
'tendon': {'color': [0, 0, 255], 'name': 'Gân'}
|
||
}
|
||
|
||
# Color map for Post-trans
|
||
COLOR_MAP_POST = {
|
||
'background': {'color': [0, 0, 0], 'name': 'Nền'},
|
||
"baker's cyst": {'color': [255, 0, 0], 'name': "Baker's cyst"},
|
||
'fat': {'color': [255, 255, 0], 'name': 'Mỡ'},
|
||
'muscle': {'color': [0, 255, 255], 'name': 'Cơ bắp'},
|
||
'femur': {'color': [0, 255, 0], 'name': 'Xương đùi'},
|
||
'synovium': {'color': [255, 0, 255], 'name': 'Màng hoạt dịch'},
|
||
'tendon': {'color': [0, 0, 255], 'name': 'Gân'}
|
||
}
|
||
|
||
# Measurement configuration
|
||
DEFAULT_MEASURE_IDS = [1, 5]
|
||
PIXEL_TO_MM = 45.0 / 655.0
|
||
|
||
app = FastAPI(title="Medical Image Analysis API")
|
||
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
# app.mount("/uploads", StaticFiles(directory=UPLOAD_FOLDER), name="uploads")
|
||
# app.mount("/results", StaticFiles(directory=RESULTS_FOLDER), name="results")
|
||
|
||
# ============ MODEL LOADING ============
|
||
|
||
# def load_angle_model(model_name):
|
||
# print(f"📄 Loading angle model: {model_name}")
|
||
|
||
# if model_name == "convnext":
|
||
# model = models.convnext_tiny(weights=None)
|
||
# model.classifier[2] = nn.Linear(model.classifier[2].in_features, 4)
|
||
# checkpoint = torch.load(f"models/best_convnext_tiny.pth", map_location=device, weights_only=False)
|
||
# checkpoint = {k.replace('model.', ''): v for k, v in checkpoint.items()}
|
||
# model.load_state_dict(checkpoint)
|
||
# elif model_name == "densenet":
|
||
# model = models.densenet121(weights=None)
|
||
# model.classifier = nn.Linear(model.classifier.in_features, 4)
|
||
# checkpoint = torch.load(f"models/best_densenet.pth", map_location=device, weights_only=False)
|
||
# checkpoint = {k.replace('model.', ''): v for k, v in checkpoint.items()}
|
||
# model.load_state_dict(checkpoint)
|
||
# elif model_name == "resnet50":
|
||
# model = models.resnet50(weights=None)
|
||
# model.fc = nn.Linear(model.fc.in_features, 4)
|
||
# checkpoint = torch.load(f"models/best_resnet50.pth", map_location=device, weights_only=False)
|
||
# checkpoint = {k.replace('model.', ''): v for k, v in checkpoint.items()}
|
||
# model.load_state_dict(checkpoint)
|
||
# elif model_name == "efficientnet_b2":
|
||
# model = models.efficientnet_b2(weights=None)
|
||
# model.classifier[1] = nn.Linear(model.classifier[1].in_features, 4)
|
||
# checkpoint = torch.load(f"models/best_efficientnet_b2.pth", map_location=device, weights_only=False)
|
||
# checkpoint = {k.replace('model.', ''): v for k, v in checkpoint.items()}
|
||
# model.load_state_dict(checkpoint)
|
||
# elif model_name == "swin":
|
||
# model = models.swin_v2_s(weights=None)
|
||
# model.head = nn.Linear(model.head.in_features, 4)
|
||
# checkpoint = torch.load(f"models/best_swin_v2_s.pth", map_location=device, weights_only=False)
|
||
# checkpoint = {k.replace('model.', ''): v for k, v in checkpoint.items()}
|
||
# model.load_state_dict(checkpoint)
|
||
# else:
|
||
# raise ValueError(f"Unknown angle model: {model_name}")
|
||
|
||
# print(f"✅ Loaded: {model_name}")
|
||
# return model.to(device).eval()
|
||
|
||
# def load_inflammation_model():
|
||
# print("📄 Loading inflammation model")
|
||
# model = models.efficientnet_b0(weights=None)
|
||
# model.classifier[1] = nn.Linear(model.classifier[1].in_features, 2)
|
||
# model.load_state_dict(torch.load("models/efficientnet_b0_ultrasound_2_class.pth", map_location=device, weights_only=False))
|
||
# print("✅ Loaded inflammation model")
|
||
# return model.to(device).eval()
|
||
|
||
# def load_segmentation_model_sup(model_name):
|
||
# print(f"📄 Loading segmentation model SUP: {model_name}")
|
||
|
||
# if model_name == "deeplabv3":
|
||
# model = models.segmentation.deeplabv3_resnet50(weights=None)
|
||
# in_ch = model.classifier[-1].in_channels
|
||
# model.classifier = nn.Sequential(
|
||
# model.classifier[0],
|
||
# nn.Dropout(0.3),
|
||
# nn.Conv2d(in_ch, 7, kernel_size=1)
|
||
# )
|
||
# model.load_state_dict(torch.load("models/best_model_Deeplav3.pth", map_location=device, weights_only=False), strict=False)
|
||
# elif model_name == "unet_resnet101":
|
||
# try:
|
||
# from segmentation_models_pytorch import Unet
|
||
# model = Unet(encoder_name="resnet101", encoder_weights=None, classes=7)
|
||
# model.load_state_dict(torch.load("models/unet_resnet101.pth", map_location=device, weights_only=False))
|
||
# except ImportError:
|
||
# raise ValueError("segmentation_models_pytorch not installed")
|
||
# elif model_name == "efficientfeedback":
|
||
# model = EfficientFeedbackNetwork(in_channels=3, num_class=7)
|
||
# model.load_state_dict(torch.load("models/efficientfeedback.pth", map_location=device, weights_only=False))
|
||
# elif model_name == "unet3plus":
|
||
# model = UNet3Plus_Attention(in_channels=3, num_classes=7)
|
||
# model.load_state_dict(torch.load("models/unet3plus_att.pth", map_location=device, weights_only=False))
|
||
# else:
|
||
# raise ValueError(f"Unknown segmentation model: {model_name}")
|
||
|
||
# print(f"✅ Loaded SUP: {model_name}")
|
||
# return model.to(device).eval()
|
||
|
||
# def load_segmentation_model_post(model_name):
|
||
# print(f"📄 Loading segmentation model POST: {model_name}")
|
||
|
||
# if model_name == "deeplabv3_resnet101":
|
||
# model = models.segmentation.deeplabv3_resnet101(weights=None)
|
||
# in_ch = model.classifier[-1].in_channels
|
||
# model.classifier = nn.Sequential(
|
||
# model.classifier[0],
|
||
# nn.Dropout(0.3),
|
||
# nn.Conv2d(in_ch, 7, kernel_size=1)
|
||
# )
|
||
# model.load_state_dict(torch.load("models/best_model_deeplabv3_resnet101_seed_16.pth", map_location=device, weights_only=False), strict=False)
|
||
# else:
|
||
# raise ValueError(f"Unknown post segmentation model: {model_name}")
|
||
|
||
# print(f"✅ Loaded POST: {model_name}")
|
||
# return model.to(device).eval()
|
||
|
||
# Transforms
|
||
angle_transform = transforms.Compose([
|
||
transforms.Resize((224, 224)),
|
||
transforms.ToTensor(),
|
||
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
|
||
])
|
||
|
||
inflammation_transform = transforms.Compose([
|
||
transforms.Resize((224, 224)),
|
||
transforms.ToTensor(),
|
||
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
|
||
])
|
||
|
||
segmentation_transform = transforms.Compose([
|
||
transforms.Resize((512, 512)),
|
||
transforms.ToTensor()
|
||
])
|
||
|
||
# ============ PREDICTION ============
|
||
|
||
@torch.no_grad()
|
||
def predict_angle(model, image_pil):
|
||
img_tensor = angle_transform(image_pil).unsqueeze(0).to(device)
|
||
output = model(img_tensor)
|
||
probs = torch.softmax(output, dim=1)
|
||
pred_class = torch.argmax(probs, dim=1).item()
|
||
confidence = probs[0][pred_class].item()
|
||
return ANGLE_CLASSES[pred_class], round(confidence * 100, 2)
|
||
|
||
@torch.no_grad()
|
||
def predict_inflammation(model, image_pil):
|
||
img_tensor = inflammation_transform(image_pil).unsqueeze(0).to(device)
|
||
output = model(img_tensor) # TRITON
|
||
probs = torch.softmax(output, dim=1)
|
||
pred_class = torch.argmax(probs, dim=1).item()
|
||
confidence = probs[0][pred_class].item()
|
||
is_inflammation = (pred_class == 1)
|
||
return is_inflammation, round(confidence * 100, 2)
|
||
|
||
@torch.no_grad()
|
||
def segment_image(model, image_pil, model_type, angle_type):
|
||
original_size = image_pil.size
|
||
img_tensor = segmentation_transform(image_pil).unsqueeze(0).to(device)
|
||
|
||
if model_type.startswith("deeplabv3"):
|
||
outputs = model(img_tensor)['out']
|
||
else:
|
||
outputs = model(img_tensor)
|
||
|
||
upsampled = F.interpolate(outputs, size=original_size[::-1], mode='bilinear', align_corners=False)
|
||
preds = upsampled.argmax(dim=1)[0].cpu().numpy()
|
||
|
||
if angle_type == 'sup' and model_type in ["unet3plus", "efficientfeedback"]:
|
||
remap = {0: 0, 1: 2, 2: 6, 3: 1, 4: 4, 5: 5, 6: 3}
|
||
preds = np.vectorize(remap.get)(preds)
|
||
|
||
class_map = SEGMENT_CLASSES_SUPRAPAT if angle_type == 'sup' else SEGMENT_CLASSES_POST
|
||
|
||
masks = {}
|
||
for class_id, class_name in class_map.items():
|
||
mask = (preds == class_id).astype(np.uint8)
|
||
masks[class_name] = mask
|
||
|
||
return preds, masks
|
||
|
||
def get_mask_bounding_box(mask, dist_percent=0.01):
|
||
"""
|
||
Duyệt toàn bộ vùng được mask, loại bỏ nhiễu và trả về khung bao (Bounding Box).
|
||
Áp dụng quy tắc kết hợp:
|
||
1. Giữ lại khối có diện tích lớn nhất (vùng trung tâm).
|
||
2. Giữ lại các khối phụ nếu thỏa mãn một trong hai điều kiện:
|
||
- Diện tích >= 1/5 diện tích khối lớn nhất.
|
||
- Khoảng cách tới khối lớn nhất <= dist_percent * chiều rộng ảnh.
|
||
"""
|
||
if mask is None or np.sum(mask) == 0:
|
||
return None
|
||
|
||
# 1. Chuyển sang uint8
|
||
mask_uint8 = mask.astype(np.uint8)
|
||
if np.max(mask_uint8) == 1:
|
||
mask_uint8 *= 255
|
||
|
||
# Lấy chiều rộng ảnh để tính ngưỡng khoảng cách theo %
|
||
img_width = mask_uint8.shape[1]
|
||
dist_threshold = img_width * dist_percent
|
||
|
||
# 2. Làm sạch mask cơ bản (Morphological Opening)
|
||
kernel = np.ones((5, 5), np.uint8)
|
||
clean_mask = cv2.morphologyEx(mask_uint8, cv2.MORPH_OPEN, kernel)
|
||
|
||
# 3. Tìm các đường bao (các khối tách rời)
|
||
contours, _ = cv2.findContours(clean_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
|
||
|
||
if not contours:
|
||
return None
|
||
|
||
# 4. Tính diện tích từng khối và tìm khối lớn nhất
|
||
contour_info = []
|
||
for cnt in contours:
|
||
contour_info.append({'cnt': cnt, 'area': cv2.contourArea(cnt)})
|
||
|
||
# Sắp xếp theo diện tích giảm dần
|
||
contour_info.sort(key=lambda x: x['area'], reverse=True)
|
||
main_block = contour_info[0]
|
||
max_area = main_block['area']
|
||
|
||
if max_area < 50:
|
||
return None
|
||
|
||
# 5. Chuẩn bị để tính khoảng cách (Distance Transform)
|
||
main_mask = np.zeros_like(mask_uint8)
|
||
cv2.drawContours(main_mask, [main_block['cnt']], -1, 255, -1)
|
||
# dist_map chứa khoảng cách từ mỗi điểm tới biên gần nhất của khối chính
|
||
dist_map = cv2.distanceTransform(255 - main_mask, cv2.DIST_L2, 3)
|
||
|
||
# 6. Lọc các khối
|
||
significant_contours = [main_block['cnt']]
|
||
area_threshold = max_area / 4.0
|
||
|
||
for i in range(1, len(contour_info)):
|
||
other = contour_info[i]
|
||
|
||
# Tạo mask cho khối đang xét để lấy giá trị khoảng cách
|
||
other_mask = np.zeros_like(mask_uint8)
|
||
cv2.drawContours(other_mask, [other['cnt']], -1, 255, -1)
|
||
|
||
# Khoảng cách nhỏ nhất từ khối này tới khối chính
|
||
min_dist = np.min(dist_map[other_mask > 0])
|
||
|
||
# Điều kiện giữ lại: (Diện tích đủ lớn) HOẶC (Ở gần khối chính theo %)
|
||
if other['area'] >= area_threshold or min_dist <= dist_threshold:
|
||
significant_contours.append(other['cnt'])
|
||
|
||
# 7. Tính toán bounding box bao quanh tất cả các vùng được chọn
|
||
all_points = np.concatenate(significant_contours)
|
||
x, y, w, h = cv2.boundingRect(all_points)
|
||
|
||
return x, y, w, h
|
||
|
||
def find_max_continuous_segment(col_array):
|
||
padded = np.concatenate(([0], col_array, [0]))
|
||
diffs = np.diff(padded)
|
||
|
||
starts = np.where(diffs == 1)[0]
|
||
ends = np.where(diffs == -1)[0]
|
||
|
||
if len(starts) == 0:
|
||
return 0, -1, -1
|
||
|
||
lengths = ends - starts
|
||
max_idx = np.argmax(lengths)
|
||
max_len = lengths[max_idx]
|
||
|
||
return max_len, starts[max_idx], ends[max_idx]
|
||
|
||
def measure_thickness_new(masks, image_size, measure_ids=None):
|
||
if measure_ids is None:
|
||
measure_ids = DEFAULT_MEASURE_IDS
|
||
|
||
width, height = image_size
|
||
|
||
mask_all_labels = np.zeros((height, width), dtype=np.uint8)
|
||
mask_measure = np.zeros((height, width), dtype=np.uint8)
|
||
|
||
has_any_label = False
|
||
|
||
if 'fat-pat' in masks:
|
||
class_map = SEGMENT_CLASSES_SUPRAPAT
|
||
else:
|
||
class_map = SEGMENT_CLASSES_POST
|
||
|
||
for class_id, class_name in class_map.items():
|
||
if class_name not in masks or class_name == 'background':
|
||
continue
|
||
mask = masks[class_name]
|
||
if np.sum(mask) > 0:
|
||
has_any_label = True
|
||
mask_all_labels = np.logical_or(mask_all_labels, mask).astype(np.uint8)
|
||
|
||
if class_id in measure_ids:
|
||
mask_measure = np.logical_or(mask_measure, mask).astype(np.uint8)
|
||
|
||
if not has_any_label or np.sum(mask_measure) == 0:
|
||
return None
|
||
|
||
# Đóng khung toàn bộ vùng được mask (xương, màng, dịch, mỡ...)
|
||
bbox_all = get_mask_bounding_box(mask_all_labels)
|
||
if bbox_all is None:
|
||
return None
|
||
|
||
x_all, y_all, w_all, h_all = bbox_all
|
||
|
||
# Từ khung này, xác định vùng quét là 1/3 ở giữa theo chiều ngang
|
||
roi_start = x_all + (w_all // 3)
|
||
roi_end = x_all + (2 * w_all // 3)
|
||
|
||
roi_strip = mask_measure[:, roi_start:roi_end]
|
||
|
||
global_max_len_px = 0
|
||
best_x_rel = 0
|
||
best_y_start = 0
|
||
best_y_end = 0
|
||
|
||
for x in range(roi_strip.shape[1]):
|
||
col = roi_strip[:, x]
|
||
if not np.any(col):
|
||
continue
|
||
|
||
length, y_s, y_e = find_max_continuous_segment(col)
|
||
|
||
if length > global_max_len_px:
|
||
global_max_len_px = length
|
||
best_x_rel = x
|
||
best_y_start = y_s
|
||
best_y_end = y_e
|
||
|
||
if global_max_len_px == 0:
|
||
return None
|
||
|
||
thickness_mm = global_max_len_px * PIXEL_TO_MM
|
||
real_x = roi_start + best_x_rel
|
||
|
||
print(f"📏 Measurement: {thickness_mm:.2f}mm ({global_max_len_px}px) at x={real_x}")
|
||
|
||
return {
|
||
'thickness_px': int(global_max_len_px),
|
||
'thickness_mm': float(round(thickness_mm, 2)),
|
||
'x': int(real_x),
|
||
'y_start': int(best_y_start),
|
||
'y_end': int(best_y_end),
|
||
'roi_start': int(roi_start),
|
||
'roi_end': int(roi_end),
|
||
'bbox': {'x': int(x_all), 'y': int(y_all), 'w': int(w_all), 'h': int(h_all)}
|
||
}
|
||
|
||
def analyze_inflammation_severity(masks, image_size):
|
||
if not masks:
|
||
return None
|
||
|
||
width, height = image_size
|
||
total_pixels = width * height
|
||
|
||
effusion_mask = masks.get('effusion', np.zeros((height, width), dtype=np.uint8))
|
||
effusion_pixels = int(np.sum(effusion_mask))
|
||
effusion_ratio = (effusion_pixels / total_pixels) * 100
|
||
|
||
effusion_thickness = 0
|
||
if effusion_pixels > 0:
|
||
rows_with_effusion = np.any(effusion_mask > 0, axis=1)
|
||
if np.any(rows_with_effusion):
|
||
effusion_thickness = int(np.sum(rows_with_effusion))
|
||
|
||
synovium_mask = masks.get('synovium', np.zeros((height, width), dtype=np.uint8))
|
||
synovium_pixels = int(np.sum(synovium_mask))
|
||
synovium_ratio = (synovium_pixels / total_pixels) * 100
|
||
|
||
effusion_score = min(effusion_thickness / height * 100, 100)
|
||
synovium_score = synovium_ratio
|
||
combined_score = (effusion_score * 0.6 + synovium_score * 0.4)
|
||
|
||
if combined_score > 15:
|
||
level, severity, color = 3, "Nặng", "#dc3545"
|
||
description = f"Dịch khớp dày ({effusion_thickness}px), màng hoạt dịch tăng sinh rõ"
|
||
elif combined_score >= 8:
|
||
level, severity, color = 2, "Trung bình", "#fd7e14"
|
||
description = f"Dịch khớp trung bình ({effusion_thickness}px), màng hoạt dịch tăng sinh vừa"
|
||
elif combined_score >= 3:
|
||
level, severity, color = 1, "Nhẹ", "#ffc107"
|
||
description = f"Dịch khớp mỏng ({effusion_thickness}px), màng hoạt dịch tăng sinh nhẹ"
|
||
else:
|
||
level, severity, color = 0, "Rất nhẹ", "#28a745"
|
||
description = "Lượng dịch và màng hoạt dịch trong giới hạn bình thường"
|
||
|
||
return {
|
||
'level': int(level),
|
||
'severity': severity,
|
||
'color': color,
|
||
'description': description,
|
||
'effusion': {'pixels': effusion_pixels, 'ratio': float(round(effusion_ratio, 2)), 'thickness': effusion_thickness},
|
||
'synovium': {'pixels': synovium_pixels, 'ratio': float(round(synovium_ratio, 2))},
|
||
'combined_score': float(round(combined_score, 2))
|
||
}
|
||
|
||
def create_segmentation_overlay(image_pil, masks, measurement=None, angle_type='sup'):
|
||
if masks is None:
|
||
return image_pil
|
||
|
||
color_map = COLOR_MAP_SUP if angle_type == 'sup' else COLOR_MAP_POST
|
||
|
||
img_array = np.array(image_pil)
|
||
overlay = img_array.copy()
|
||
|
||
for class_name, mask in masks.items():
|
||
if class_name in color_map and np.sum(mask) > 0:
|
||
color = color_map[class_name]['color']
|
||
for i in range(3):
|
||
overlay[:, :, i] = np.where(mask > 0,
|
||
(overlay[:, :, i] * 0.6 + color[i] * 0.4).astype(np.uint8),
|
||
overlay[:, :, i])
|
||
|
||
overlay_pil = Image.fromarray(overlay)
|
||
draw = ImageDraw.Draw(overlay_pil)
|
||
|
||
for class_name in ['effusion', 'synovium']:
|
||
mask = masks.get(class_name)
|
||
if mask is not None and np.sum(mask) > 0:
|
||
mask_uint8 = (mask * 255).astype(np.uint8)
|
||
contours, _ = cv2.findContours(mask_uint8, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
|
||
|
||
for contour in contours:
|
||
points = contour.reshape(-1, 2).tolist()
|
||
if len(points) > 2:
|
||
points = [(int(p[0]), int(p[1])) for p in points]
|
||
draw.line(points + [points[0]], fill=(255, 255, 255), width=3)
|
||
|
||
if measurement and angle_type == 'sup':
|
||
x = measurement['x']
|
||
y_start = measurement['y_start']
|
||
y_end = measurement['y_end']
|
||
thickness_mm = measurement['thickness_mm']
|
||
roi_start = measurement['roi_start']
|
||
roi_end = measurement['roi_end']
|
||
bbox = measurement['bbox']
|
||
|
||
draw.rectangle(
|
||
[bbox['x'], bbox['y'], bbox['x'] + bbox['w'], bbox['y'] + bbox['h']],
|
||
outline=(0, 255, 0), # Chuyển sang xanh lá cho dễ nhìn
|
||
width=3 # Tăng độ dày khung
|
||
)
|
||
|
||
h = image_pil.size[1]
|
||
draw.line([(roi_start, 0), (roi_start, h)], fill=(0, 255, 255), width=2)
|
||
draw.line([(roi_end, 0), (roi_end, h)], fill=(0, 255, 255), width=2)
|
||
|
||
draw.line([(x, y_start), (x, y_end)], fill=(255, 0, 0), width=4)
|
||
|
||
radius = 4
|
||
draw.ellipse([x-radius, y_start-radius, x+radius, y_start+radius],
|
||
fill=(0, 255, 0), outline=(255, 255, 255), width=2)
|
||
draw.ellipse([x-radius, y_end-radius, x+radius, y_end+radius],
|
||
fill=(0, 255, 0), outline=(255, 255, 255), width=2)
|
||
|
||
text = f"{thickness_mm:.2f} mm"
|
||
|
||
try:
|
||
from PIL import ImageFont
|
||
font = ImageFont.load_default()
|
||
bbox_text = draw.textbbox((0, 0), text, font=font)
|
||
text_w = bbox_text[2] - bbox_text[0]
|
||
text_h = bbox_text[3] - bbox_text[1]
|
||
except:
|
||
text_w, text_h = 100, 20
|
||
|
||
text_x = x + 8
|
||
text_y = y_start - text_h - 8
|
||
|
||
draw.rectangle(
|
||
[text_x - 2, text_y - 2, text_x + text_w + 2, text_y + text_h + 2],
|
||
fill=(0, 0, 0)
|
||
)
|
||
draw.text((text_x, text_y), text, fill=(255, 0, 0))
|
||
|
||
return overlay_pil
|
||
|
||
def apply_clahe(image_pil):
|
||
"""Áp dụng thuật toán CLAHE để tăng độ tương phản. Phục vụ cả hiển thị và làm đầu vào AI."""
|
||
# Chuyển từ PIL sang OpenCV (numpy array)
|
||
img_array = np.array(image_pil)
|
||
# Chuyển sang thang độ xám (Gray) để xử lý CLAHE
|
||
gray = cv2.cvtColor(img_array, cv2.COLOR_RGB2GRAY)
|
||
|
||
# Tạo đối tượng CLAHE
|
||
clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8,8))
|
||
enhanced_gray = clahe.apply(gray)
|
||
|
||
# Chuyển ngược lại sang RGB (3 kênh) để tương thích với các models
|
||
enhanced_rgb = cv2.cvtColor(enhanced_gray, cv2.COLOR_GRAY2RGB)
|
||
return Image.fromarray(enhanced_rgb)
|
||
|
||
# # Mount thư mục static (CSS, JS)
|
||
# app.mount("/css", StaticFiles(directory="templates/css"), name="css")
|
||
# app.mount("/js", StaticFiles(directory="templates/js"), name="js")
|
||
|
||
|
||
# @app.get("/")
|
||
# async def read_index():
|
||
# html_file = Path(TEMPLATES_FOLDER) / "index.html"
|
||
# if html_file.exists():
|
||
# return FileResponse(html_file)
|
||
# return JSONResponse({"error": "Template not found"})
|
||
|
||
@app.post("/api/analyze")
|
||
async def analyze_image(
|
||
image: UploadFile = File(...),
|
||
angle_model: str = Query("convnext"),
|
||
inflammation_model: str = Query("efficientnet_b0"),
|
||
segment_model_sup: str = Query("deeplabv3"),
|
||
segment_model_post: str = Query("deeplabv3_resnet101")
|
||
):
|
||
try:
|
||
print(f"\n{'='*60}")
|
||
print(f"📊 NEW REQUEST")
|
||
print(f"Models: angle={angle_model}, inflam={inflammation_model}")
|
||
print(f" seg_sup={segment_model_sup}, seg_post={segment_model_post}")
|
||
print(f"{'='*60}")
|
||
|
||
contents = await image.read()
|
||
image_pil = Image.open(io.BytesIO(contents)).convert('RGB')
|
||
|
||
# Tạo ảnh tăng cường độ tương phản (Enhanced) cho mục đích hiển thị
|
||
enhanced_pil = apply_clahe(image_pil)
|
||
buffered_en = io.BytesIO()
|
||
enhanced_pil.save(buffered_en, format="PNG")
|
||
enhanced_str = base64.b64encode(buffered_en.getvalue()).decode()
|
||
|
||
result = {
|
||
'success': True,
|
||
'filename': image.filename,
|
||
'images': {
|
||
'enhanced': f"data:image/png;base64,{enhanced_str}"
|
||
},
|
||
'models_used': {
|
||
'angle': angle_model,
|
||
'inflammation': inflammation_model,
|
||
'segmentation_sup': segment_model_sup,
|
||
'segmentation_post': segment_model_post
|
||
}
|
||
}
|
||
|
||
angle_clf = load_angle_model(angle_model)
|
||
angle, angle_conf = predict_angle(angle_clf, image_pil)
|
||
result['angle'] = {'class': angle, 'confidence': angle_conf}
|
||
print(f"✅ Angle: {angle} ({angle_conf}%)")
|
||
|
||
if 'post-trans' in angle.lower():
|
||
print(f"🔍 POST-TRANS pipeline")
|
||
|
||
inflam_model = load_inflammation_model()
|
||
has_inflammation, inflam_conf = predict_inflammation(inflam_model, image_pil)
|
||
result['inflammation'] = {'detected': has_inflammation, 'confidence': inflam_conf}
|
||
print(f"✅ Inflammation: {has_inflammation} ({inflam_conf}%)")
|
||
|
||
if has_inflammation:
|
||
seg_model = load_segmentation_model_post(segment_model_post)
|
||
preds, masks = segment_image(seg_model, image_pil, segment_model_post, 'post')
|
||
|
||
if masks:
|
||
segmented_img = create_segmentation_overlay(image_pil, masks, None, 'post')
|
||
# Convert to base64
|
||
buffered = io.BytesIO()
|
||
segmented_img.save(buffered, format="PNG")
|
||
img_str = base64.b64encode(buffered.getvalue()).decode()
|
||
|
||
result['images']['segmented'] = f"data:image/png;base64,{img_str}"
|
||
|
||
detected_classes = [k for k, v in masks.items() if np.sum(v) > 0]
|
||
color_legend = []
|
||
for class_name in detected_classes:
|
||
if class_name in COLOR_MAP_POST:
|
||
color_legend.append({
|
||
'name': COLOR_MAP_POST[class_name]['name'],
|
||
'color': COLOR_MAP_POST[class_name]['color'],
|
||
'key': class_name
|
||
})
|
||
|
||
result['segmentation'] = {
|
||
'performed': True,
|
||
'classes_detected': detected_classes,
|
||
'color_legend': color_legend,
|
||
'angle_type': 'post'
|
||
}
|
||
|
||
print(f"✅ Segmentation POST completed")
|
||
else:
|
||
print(f"ℹ️ No inflammation detected - skipping segmentation POST")
|
||
|
||
elif 'sup-up-long' in angle.lower():
|
||
print(f"🔍 SUPRAPAT pipeline")
|
||
|
||
inflam_model = load_inflammation_model()
|
||
has_inflammation, inflam_conf = predict_inflammation(inflam_model, image_pil)
|
||
result['inflammation'] = {'detected': has_inflammation, 'confidence': inflam_conf}
|
||
print(f"✅ Inflammation: {has_inflammation} ({inflam_conf}%)")
|
||
|
||
if has_inflammation:
|
||
seg_model = load_segmentation_model_sup(segment_model_sup)
|
||
_, masks = segment_image(seg_model, image_pil, segment_model_sup, 'sup')
|
||
|
||
if masks:
|
||
measurement = measure_thickness_new(masks, image_pil.size)
|
||
|
||
segmented_img = create_segmentation_overlay(image_pil, masks, measurement, 'sup')
|
||
# Convert to base64
|
||
buffered = io.BytesIO()
|
||
segmented_img.save(buffered, format="PNG")
|
||
img_str = base64.b64encode(buffered.getvalue()).decode()
|
||
|
||
result['images']['segmented'] = f"data:image/png;base64,{img_str}"
|
||
|
||
if measurement:
|
||
result['measurement'] = {
|
||
'thickness_mm': measurement['thickness_mm'],
|
||
'thickness_px': measurement['thickness_px'],
|
||
'location_x': measurement['x'],
|
||
'y_start': measurement['y_start'],
|
||
'y_end': measurement['y_end']
|
||
}
|
||
print(f"✅ Measurement: {measurement['thickness_mm']:.2f}mm at x={measurement['x']}")
|
||
|
||
detected_classes = [k for k, v in masks.items() if np.sum(v) > 0]
|
||
color_legend = []
|
||
for class_name in detected_classes:
|
||
if class_name in COLOR_MAP_SUP:
|
||
color_legend.append({
|
||
'name': COLOR_MAP_SUP[class_name]['name'],
|
||
'color': COLOR_MAP_SUP[class_name]['color'],
|
||
'key': class_name
|
||
})
|
||
|
||
result['segmentation'] = {
|
||
'performed': True,
|
||
'classes_detected': detected_classes,
|
||
'color_legend': color_legend,
|
||
'angle_type': 'sup'
|
||
}
|
||
|
||
severity = analyze_inflammation_severity(masks, image_pil.size)
|
||
if severity:
|
||
result['severity'] = severity
|
||
print(f"✅ Severity: {severity['severity']}")
|
||
|
||
print(f"✅ Segmentation SUP completed")
|
||
else:
|
||
print(f"ℹ️ No inflammation detected - skipping segmentation SUP")
|
||
|
||
else:
|
||
print(f"ℹ️ Other angle - only angle classification")
|
||
|
||
print(f"{'='*60}\n")
|
||
return JSONResponse(result)
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
print(f"❌ Error: {e}")
|
||
print(traceback.format_exc())
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.get("/api/health")
|
||
async def health_check():
|
||
return JSONResponse({'status': 'healthy'})
|
||
|
||
def sanitize_name(name):
|
||
# Loại bỏ ký tự không hợp lệ cho folder trên Windows
|
||
if not name: return "unknown"
|
||
# Thay thế ký tự lạ bằng dấu gạch dưới
|
||
clean = re.sub(r'[\\/*?:"<>|]', "_", name)
|
||
# Loại bỏ dấu cách thừa
|
||
clean = clean.strip().replace(" ", "_")
|
||
return clean
|
||
|
||
class SaveDataRequest(BaseModel):
|
||
patient_info: dict
|
||
analysis_result: dict
|
||
images: dict
|
||
|
||
@app.post("/api/save")
|
||
async def save_patient_data(data: SaveDataRequest):
|
||
try:
|
||
p = data.patient_info
|
||
res = data.analysis_result
|
||
imgs = data.images
|
||
|
||
# 1. Tạo thư mục theo mã bệnh nhân (nhóm chính)
|
||
patient_id = sanitize_name(p.get('id', 'unknown'))
|
||
patient_name = sanitize_name(p.get('name', 'no_name'))
|
||
|
||
# Thư mục chính của bệnh nhân
|
||
patient_folder = f"{patient_id}_{patient_name}"
|
||
|
||
# Thư mục con theo thời gian hiện tại
|
||
timestamp_folder = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
|
||
# Tổng hợp đường dẫn: patients/ID_Name/TIMESTAMP/
|
||
target_dir = os.path.join("patients", patient_folder, timestamp_folder)
|
||
os.makedirs(target_dir, exist_ok=True)
|
||
|
||
# 2. Lưu info.txt
|
||
info_path = os.path.join(target_dir, "info.txt")
|
||
with open(info_path, "w", encoding="utf-8") as f:
|
||
f.write(f"--- THÔNG TIN BỆNH NHÂN ---\n")
|
||
f.write(f"Mã BN: {patient_id}\n")
|
||
f.write(f"Họ tên: {patient_name}\n")
|
||
f.write(f"Giới tính: {p.get('gender')}\n")
|
||
f.write(f"Tuổi: {p.get('age')}\n")
|
||
f.write(f"Chẩn đoán BS: {p.get('diagnosis')}\n\n")
|
||
|
||
f.write(f"--- KẾT QUẢ PHÂN TÍCH AI ---\n")
|
||
f.write(f"Góc chụp: {res.get('angle', {}).get('class')} ({res.get('angle', {}).get('confidence')}%)\n")
|
||
|
||
if 'inflammation' in res:
|
||
infl = res['inflammation']
|
||
f.write(f"Viêm nhiễm: {'Có' if infl['detected'] else 'Không'} ({infl['confidence']}%)\n")
|
||
|
||
if 'measurement' in res:
|
||
m = res['measurement']
|
||
f.write(f"Độ dày màng: {m['thickness_mm']} mm ({m['thickness_px']} px)\n")
|
||
f.write(f"Vị trí x: {m['location_x']}\n")
|
||
|
||
if 'severity' in res:
|
||
s = res['severity']
|
||
f.write(f"Mức độ: {s['severity']}\n")
|
||
f.write(f"Mô tả: {s['description']}\n")
|
||
|
||
# 3. Lưu ảnh
|
||
def save_base64_img(b64_str, filename):
|
||
if not b64_str: return
|
||
# Remove header if present
|
||
if "," in b64_str:
|
||
b64_str = b64_str.split(",")[1]
|
||
|
||
img_data = base64.b64decode(b64_str)
|
||
with open(os.path.join(target_dir, filename), "wb") as f:
|
||
f.write(img_data)
|
||
|
||
save_base64_img(imgs.get('original'), "original.png")
|
||
save_base64_img(imgs.get('segmented'), "segmented.png")
|
||
|
||
# 4. Tự động lưu PDF báo cáo
|
||
try:
|
||
pdf_bytes = generate_medical_report(p, res, imgs)
|
||
pdf_path = os.path.join(target_dir, "report.pdf")
|
||
with open(pdf_path, "wb") as f:
|
||
f.write(bytes(pdf_bytes))
|
||
print(f"📄 Report PDF saved to: {pdf_path}")
|
||
except Exception as pdf_err:
|
||
print(f"⚠️ Warning: Could not auto-save PDF: {pdf_err}")
|
||
|
||
print(f"✅ Data saved for patient: {patient_id}")
|
||
return {"success": True, "folder": target_dir}
|
||
|
||
except Exception as e:
|
||
print(f"❌ Save Error: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.post("/api/export-pdf")
|
||
async def export_patient_pdf(data: SaveDataRequest):
|
||
try:
|
||
pdf_bytes = generate_medical_report(
|
||
data.patient_info,
|
||
data.analysis_result,
|
||
data.images
|
||
)
|
||
|
||
filename = f"Phieu_Kham_{sanitize_name(data.patient_info.get('id', 'unknown'))}.pdf"
|
||
|
||
return Response(
|
||
content=bytes(pdf_bytes),
|
||
media_type="application/pdf",
|
||
headers={"Content-Disposition": f"attachment; filename={filename}"}
|
||
)
|
||
except Exception as e:
|
||
import traceback
|
||
print(f"❌ PDF Export Error: {e}")
|
||
print(traceback.format_exc())
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
if __name__ == '__main__':
|
||
print("Medical Image Analysis Server")
|
||
print(f"URL: http://127.0.0.1:8000")
|
||
print(f"Device: {device}")
|
||
uvicorn.run(app, host="127.0.0.1", port=8000) |