Spaces:
Sleeping
Sleeping
Update main.py
Browse files
main.py
CHANGED
|
@@ -33,7 +33,7 @@ def get_region_tensor(device):
|
|
| 33 |
return region_map
|
| 34 |
|
| 35 |
def calculate_density(coords, edge_index):
|
| 36 |
-
# Pure PyTorch
|
| 37 |
row, col = edge_index
|
| 38 |
dist = torch.norm(coords[row] - coords[col], dim=1)
|
| 39 |
sum_dist = torch.zeros(468, device=coords.device)
|
|
@@ -87,6 +87,7 @@ class AnatomyLocationNet(nn.Module):
|
|
| 87 |
self.region_emb = nn.Embedding(9, 32)
|
| 88 |
self.density_proj = nn.Linear(1, 16)
|
| 89 |
self.context_proj = nn.Linear(512, 32)
|
|
|
|
| 90 |
self.neck = nn.Sequential(
|
| 91 |
nn.Linear(176, 256), nn.BatchNorm1d(256), nn.LeakyReLU(0.2), nn.Dropout(0.4),
|
| 92 |
nn.Linear(256, 128), nn.LeakyReLU(0.2), nn.Linear(128, 1)
|
|
@@ -94,7 +95,9 @@ class AnatomyLocationNet(nn.Module):
|
|
| 94 |
|
| 95 |
def forward(self, coords, pids, rids, den, ctx):
|
| 96 |
B, N, _ = coords.shape
|
| 97 |
-
|
|
|
|
|
|
|
| 98 |
combined = torch.cat([self.coord_proj(coords), self.point_id_emb(pids),
|
| 99 |
self.region_emb(rids), self.density_proj(den), c_emb], dim=-1)
|
| 100 |
return self.neck(combined.view(-1, 176)).view(B, N, 1)
|
|
@@ -106,6 +109,7 @@ class GatedFusion(nn.Module):
|
|
| 106 |
self.gate_net = nn.Sequential(nn.Linear(geo_dim * 2, geo_dim // 2), nn.ReLU(), nn.Linear(geo_dim // 2, geo_dim), nn.Sigmoid())
|
| 107 |
def forward(self, x_geo, x_ctx):
|
| 108 |
ctx_adapted = self.context_adapter(x_ctx)
|
|
|
|
| 109 |
ctx_expanded = ctx_adapted.expand(-1, 468, -1)
|
| 110 |
combined = torch.cat([x_geo, ctx_expanded], dim=-1)
|
| 111 |
return x_geo + (self.gate_net(combined) * ctx_expanded)
|
|
@@ -159,14 +163,13 @@ backbone = SmartClinicalNet(hidden=32, heads=2).to(DEVICE)
|
|
| 159 |
expert_model = RegionAwareExpert().to(DEVICE)
|
| 160 |
location_model = AnatomyLocationNet().to(DEVICE)
|
| 161 |
|
|
|
|
| 162 |
if os.path.exists("smart_clinical_model.pth"):
|
| 163 |
backbone.load_state_dict(torch.load("smart_clinical_model.pth", map_location=DEVICE), strict=False)
|
| 164 |
backbone.eval()
|
| 165 |
-
|
| 166 |
if os.path.exists("region_expert.pth"):
|
| 167 |
expert_model.load_state_dict(torch.load("region_expert.pth", map_location=DEVICE))
|
| 168 |
expert_model.eval()
|
| 169 |
-
|
| 170 |
if os.path.exists("anatomy_location.pth"):
|
| 171 |
location_model.load_state_dict(torch.load("anatomy_location.pth", map_location=DEVICE))
|
| 172 |
location_model.eval()
|
|
@@ -203,17 +206,16 @@ def apply_nms_indices(landmarks_np, probs_np, spacing_mm=12.0, threshold=0.5):
|
|
| 203 |
sorted_idx = rest[survivors]
|
| 204 |
return keep
|
| 205 |
|
| 206 |
-
# Helper to format Top 3 probabilities
|
| 207 |
def get_top_probs(probs_array, class_list):
|
| 208 |
sorted_idxs = np.argsort(probs_array)[::-1][:3]
|
| 209 |
return {class_list[i]: float(f"{probs_array[i]:.4f}") for i in sorted_idxs}
|
| 210 |
|
| 211 |
# ==========================================
|
| 212 |
-
# 5. API ENDPOINT (
|
| 213 |
# ==========================================
|
| 214 |
@app.get("/")
|
| 215 |
def home():
|
| 216 |
-
return {"message": "SOTA Clinical AI
|
| 217 |
|
| 218 |
@app.post("/predict")
|
| 219 |
async def predict_injections(file: UploadFile = File(...)):
|
|
@@ -224,7 +226,7 @@ async def predict_injections(file: UploadFile = File(...)):
|
|
| 224 |
if img_bgr is None: return JSONResponse(status_code=400, content={"error": "Invalid image"})
|
| 225 |
img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
|
| 226 |
|
| 227 |
-
# 2. Extract
|
| 228 |
res = mp_mesh.process(img_rgb)
|
| 229 |
if not res.multi_face_landmarks:
|
| 230 |
return JSONResponse(status_code=400, content={"error": "No face detected"})
|
|
@@ -239,15 +241,16 @@ async def predict_injections(file: UploadFile = File(...)):
|
|
| 239 |
virt_prof = get_virtual_profile_norm(x_geo_norm.cpu()).to(DEVICE)
|
| 240 |
|
| 241 |
# 3. Prepare Inputs
|
|
|
|
| 242 |
data = Data(x=x_geo_norm, edge_index=GLOBAL_EDGE_INDEX, embedding=emb.unsqueeze(0), virt_prof=virt_prof.unsqueeze(0))
|
| 243 |
density = calculate_density(x_geo_norm, GLOBAL_EDGE_INDEX).unsqueeze(0).to(DEVICE)
|
| 244 |
|
| 245 |
with torch.no_grad():
|
| 246 |
-
# A.
|
| 247 |
logits_loc = location_model(x_geo_norm.unsqueeze(0), STATIC_POINT_IDS, STATIC_REGION_IDS, density, emb.unsqueeze(0))
|
| 248 |
probs_loc = torch.sigmoid(logits_loc).squeeze().cpu().numpy()
|
| 249 |
|
| 250 |
-
# B.
|
| 251 |
smart_features = backbone.get_features(data).unsqueeze(0)
|
| 252 |
coords_input = x_geo_norm.unsqueeze(0)
|
| 253 |
preds = expert_model(smart_features, coords_input, STATIC_POINT_IDS, STATIC_REGION_IDS)
|
|
@@ -257,38 +260,35 @@ async def predict_injections(file: UploadFile = File(...)):
|
|
| 257 |
prob_de = torch.softmax(preds['depth'], dim=-1).squeeze().cpu().numpy()
|
| 258 |
prob_p = torch.softmax(preds['product'], dim=-1).squeeze().cpu().numpy()
|
| 259 |
|
| 260 |
-
# 4.
|
| 261 |
h, w, _ = img_bgr.shape
|
| 262 |
pixel_coords = np.array([[p.x*w, p.y*h, p.z*w] for p in lms[:468]])
|
| 263 |
optimal_indices_list = apply_nms_indices(pixel_coords, probs_loc, spacing_mm=12.0, threshold=0.4)
|
| 264 |
optimal_set = set(optimal_indices_list)
|
| 265 |
|
|
|
|
| 266 |
classes_tech = ["Bolus", "Fanning", "Microbolus"]
|
| 267 |
classes_dosage = ["0.01ml", "0.02ml", "0.05ml", "0.1ml", "0.2ml", "0.3ml", "0.5ml", "1.0ml"]
|
| 268 |
classes_depth = ["Periosteal", "Subdermal", "Hypodermic", "Dermal"]
|
| 269 |
classes_prod = ["XXL", "XL", "L", "M", "S", "Hydro", "Induce", "Lips"]
|
| 270 |
|
| 271 |
-
# 5. Build Response
|
| 272 |
all_points_list = []
|
| 273 |
|
| 274 |
-
# Iterate through ALL points, not just optimal ones
|
| 275 |
for idx in range(468):
|
| 276 |
-
# Extract distributions for this specific point
|
| 277 |
p_t, p_d, p_de, p_p = prob_t[idx], prob_d[idx], prob_de[idx], prob_p[idx]
|
| 278 |
|
| 279 |
pt_info = {
|
| 280 |
"point_id": int(idx),
|
| 281 |
"confidence": float(f"{probs_loc[idx]:.4f}"),
|
| 282 |
-
"is_optimal": idx in optimal_set,
|
| 283 |
"coordinates": {"x": lms[idx].x, "y": lms[idx].y},
|
| 284 |
"attributes": {
|
| 285 |
-
# Top Picks
|
| 286 |
"technique": classes_tech[np.argmax(p_t)],
|
| 287 |
"dosage": classes_dosage[np.argmax(p_d)],
|
| 288 |
"depth": classes_depth[np.argmax(p_de)],
|
| 289 |
"product": classes_prod[np.argmax(p_p)],
|
| 290 |
|
| 291 |
-
# Full Probabilities (Fixes the 0% UI bug)
|
| 292 |
"technique_probs": get_top_probs(p_t, classes_tech),
|
| 293 |
"dosage_probs": get_top_probs(p_d, classes_dosage),
|
| 294 |
"depth_probs": get_top_probs(p_de, classes_depth),
|
|
@@ -306,7 +306,6 @@ async def predict_injections(file: UploadFile = File(...)):
|
|
| 306 |
"total_optimal": len(optimal_set),
|
| 307 |
"max_confidence": float(probs_loc.max())
|
| 308 |
},
|
| 309 |
-
# NOTE: Returned key is 'injection_sites' to represent full dataset
|
| 310 |
"injection_sites": all_points_list,
|
| 311 |
"all_probabilities": [float(f"{p:.4f}") for p in probs_loc.tolist()],
|
| 312 |
"all_coordinates": all_coords_list
|
|
|
|
| 33 |
return region_map
|
| 34 |
|
| 35 |
def calculate_density(coords, edge_index):
|
| 36 |
+
# Pure PyTorch: No scatter dependency issues
|
| 37 |
row, col = edge_index
|
| 38 |
dist = torch.norm(coords[row] - coords[col], dim=1)
|
| 39 |
sum_dist = torch.zeros(468, device=coords.device)
|
|
|
|
| 87 |
self.region_emb = nn.Embedding(9, 32)
|
| 88 |
self.density_proj = nn.Linear(1, 16)
|
| 89 |
self.context_proj = nn.Linear(512, 32)
|
| 90 |
+
|
| 91 |
self.neck = nn.Sequential(
|
| 92 |
nn.Linear(176, 256), nn.BatchNorm1d(256), nn.LeakyReLU(0.2), nn.Dropout(0.4),
|
| 93 |
nn.Linear(256, 128), nn.LeakyReLU(0.2), nn.Linear(128, 1)
|
|
|
|
| 95 |
|
| 96 |
def forward(self, coords, pids, rids, den, ctx):
|
| 97 |
B, N, _ = coords.shape
|
| 98 |
+
# ✅ FIXED: Safe expansion. Unsqueeze dim 1, then expand to N.
|
| 99 |
+
c_emb = self.context_proj(ctx).unsqueeze(1).expand(B, N, -1)
|
| 100 |
+
|
| 101 |
combined = torch.cat([self.coord_proj(coords), self.point_id_emb(pids),
|
| 102 |
self.region_emb(rids), self.density_proj(den), c_emb], dim=-1)
|
| 103 |
return self.neck(combined.view(-1, 176)).view(B, N, 1)
|
|
|
|
| 109 |
self.gate_net = nn.Sequential(nn.Linear(geo_dim * 2, geo_dim // 2), nn.ReLU(), nn.Linear(geo_dim // 2, geo_dim), nn.Sigmoid())
|
| 110 |
def forward(self, x_geo, x_ctx):
|
| 111 |
ctx_adapted = self.context_adapter(x_ctx)
|
| 112 |
+
# Expand context to match geometry
|
| 113 |
ctx_expanded = ctx_adapted.expand(-1, 468, -1)
|
| 114 |
combined = torch.cat([x_geo, ctx_expanded], dim=-1)
|
| 115 |
return x_geo + (self.gate_net(combined) * ctx_expanded)
|
|
|
|
| 163 |
expert_model = RegionAwareExpert().to(DEVICE)
|
| 164 |
location_model = AnatomyLocationNet().to(DEVICE)
|
| 165 |
|
| 166 |
+
# Load weights safely
|
| 167 |
if os.path.exists("smart_clinical_model.pth"):
|
| 168 |
backbone.load_state_dict(torch.load("smart_clinical_model.pth", map_location=DEVICE), strict=False)
|
| 169 |
backbone.eval()
|
|
|
|
| 170 |
if os.path.exists("region_expert.pth"):
|
| 171 |
expert_model.load_state_dict(torch.load("region_expert.pth", map_location=DEVICE))
|
| 172 |
expert_model.eval()
|
|
|
|
| 173 |
if os.path.exists("anatomy_location.pth"):
|
| 174 |
location_model.load_state_dict(torch.load("anatomy_location.pth", map_location=DEVICE))
|
| 175 |
location_model.eval()
|
|
|
|
| 206 |
sorted_idx = rest[survivors]
|
| 207 |
return keep
|
| 208 |
|
|
|
|
| 209 |
def get_top_probs(probs_array, class_list):
|
| 210 |
sorted_idxs = np.argsort(probs_array)[::-1][:3]
|
| 211 |
return {class_list[i]: float(f"{probs_array[i]:.4f}") for i in sorted_idxs}
|
| 212 |
|
| 213 |
# ==========================================
|
| 214 |
+
# 5. API ENDPOINT (FINAL VERSION)
|
| 215 |
# ==========================================
|
| 216 |
@app.get("/")
|
| 217 |
def home():
|
| 218 |
+
return {"message": "SOTA Clinical AI - Ready"}
|
| 219 |
|
| 220 |
@app.post("/predict")
|
| 221 |
async def predict_injections(file: UploadFile = File(...)):
|
|
|
|
| 226 |
if img_bgr is None: return JSONResponse(status_code=400, content={"error": "Invalid image"})
|
| 227 |
img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
|
| 228 |
|
| 229 |
+
# 2. Extract Features
|
| 230 |
res = mp_mesh.process(img_rgb)
|
| 231 |
if not res.multi_face_landmarks:
|
| 232 |
return JSONResponse(status_code=400, content={"error": "No face detected"})
|
|
|
|
| 241 |
virt_prof = get_virtual_profile_norm(x_geo_norm.cpu()).to(DEVICE)
|
| 242 |
|
| 243 |
# 3. Prepare Inputs
|
| 244 |
+
# unsqueeze(0) for batch size of 1
|
| 245 |
data = Data(x=x_geo_norm, edge_index=GLOBAL_EDGE_INDEX, embedding=emb.unsqueeze(0), virt_prof=virt_prof.unsqueeze(0))
|
| 246 |
density = calculate_density(x_geo_norm, GLOBAL_EDGE_INDEX).unsqueeze(0).to(DEVICE)
|
| 247 |
|
| 248 |
with torch.no_grad():
|
| 249 |
+
# A. Location Finder
|
| 250 |
logits_loc = location_model(x_geo_norm.unsqueeze(0), STATIC_POINT_IDS, STATIC_REGION_IDS, density, emb.unsqueeze(0))
|
| 251 |
probs_loc = torch.sigmoid(logits_loc).squeeze().cpu().numpy()
|
| 252 |
|
| 253 |
+
# B. Attribute Expert
|
| 254 |
smart_features = backbone.get_features(data).unsqueeze(0)
|
| 255 |
coords_input = x_geo_norm.unsqueeze(0)
|
| 256 |
preds = expert_model(smart_features, coords_input, STATIC_POINT_IDS, STATIC_REGION_IDS)
|
|
|
|
| 260 |
prob_de = torch.softmax(preds['depth'], dim=-1).squeeze().cpu().numpy()
|
| 261 |
prob_p = torch.softmax(preds['product'], dim=-1).squeeze().cpu().numpy()
|
| 262 |
|
| 263 |
+
# 4. Optimal Indices (For UI Highlighting)
|
| 264 |
h, w, _ = img_bgr.shape
|
| 265 |
pixel_coords = np.array([[p.x*w, p.y*h, p.z*w] for p in lms[:468]])
|
| 266 |
optimal_indices_list = apply_nms_indices(pixel_coords, probs_loc, spacing_mm=12.0, threshold=0.4)
|
| 267 |
optimal_set = set(optimal_indices_list)
|
| 268 |
|
| 269 |
+
# Classes
|
| 270 |
classes_tech = ["Bolus", "Fanning", "Microbolus"]
|
| 271 |
classes_dosage = ["0.01ml", "0.02ml", "0.05ml", "0.1ml", "0.2ml", "0.3ml", "0.5ml", "1.0ml"]
|
| 272 |
classes_depth = ["Periosteal", "Subdermal", "Hypodermic", "Dermal"]
|
| 273 |
classes_prod = ["XXL", "XL", "L", "M", "S", "Hydro", "Induce", "Lips"]
|
| 274 |
|
| 275 |
+
# 5. Build Response
|
| 276 |
all_points_list = []
|
| 277 |
|
|
|
|
| 278 |
for idx in range(468):
|
|
|
|
| 279 |
p_t, p_d, p_de, p_p = prob_t[idx], prob_d[idx], prob_de[idx], prob_p[idx]
|
| 280 |
|
| 281 |
pt_info = {
|
| 282 |
"point_id": int(idx),
|
| 283 |
"confidence": float(f"{probs_loc[idx]:.4f}"),
|
| 284 |
+
"is_optimal": idx in optimal_set,
|
| 285 |
"coordinates": {"x": lms[idx].x, "y": lms[idx].y},
|
| 286 |
"attributes": {
|
|
|
|
| 287 |
"technique": classes_tech[np.argmax(p_t)],
|
| 288 |
"dosage": classes_dosage[np.argmax(p_d)],
|
| 289 |
"depth": classes_depth[np.argmax(p_de)],
|
| 290 |
"product": classes_prod[np.argmax(p_p)],
|
| 291 |
|
|
|
|
| 292 |
"technique_probs": get_top_probs(p_t, classes_tech),
|
| 293 |
"dosage_probs": get_top_probs(p_d, classes_dosage),
|
| 294 |
"depth_probs": get_top_probs(p_de, classes_depth),
|
|
|
|
| 306 |
"total_optimal": len(optimal_set),
|
| 307 |
"max_confidence": float(probs_loc.max())
|
| 308 |
},
|
|
|
|
| 309 |
"injection_sites": all_points_list,
|
| 310 |
"all_probabilities": [float(f"{p:.4f}") for p in probs_loc.tolist()],
|
| 311 |
"all_coordinates": all_coords_list
|