anilegin commited on
Commit
b76e913
·
verified ·
1 Parent(s): bfe5b89

Update main.py

Browse files
Files changed (1) hide show
  1. main.py +24 -46
main.py CHANGED
@@ -32,33 +32,22 @@ def get_region_tensor(device):
32
  if p < 468: region_map[p] = region_id
33
  return region_map
34
 
35
- # --- REPLACE THIS FUNCTION IN MAIN.PY ---
36
  def calculate_density(coords, edge_index):
37
  # Pure PyTorch version (No extra dependencies required)
38
  row, col = edge_index
39
-
40
- # 1. Calculate distance for every edge
41
  dist = torch.norm(coords[row] - coords[col], dim=1)
42
-
43
- # 2. Sum distances for each point (Scatter Add)
44
  sum_dist = torch.zeros(468, device=coords.device)
45
  sum_dist.scatter_add_(0, row, dist)
46
-
47
- # 3. Count neighbors for each point
48
  count = torch.zeros(468, device=coords.device)
49
  count.scatter_add_(0, row, torch.ones_like(dist))
50
-
51
- # 4. Calculate Mean and Density
52
  mean_dist = sum_dist / (count + 1e-6)
53
  density = 1.0 / (mean_dist + 1e-6)
54
-
55
  return density.unsqueeze(1) # [1, 468, 1]
56
 
57
  # ==========================================
58
  # 2. MODEL ARCHITECTURES
59
  # ==========================================
60
 
61
- # --- A. ATTRIBUTE EXPERT (RegionAwareExpert) ---
62
  class RegionAwareExpert(nn.Module):
63
  def __init__(self):
64
  super().__init__()
@@ -66,13 +55,11 @@ class RegionAwareExpert(nn.Module):
66
  self.point_id_emb = nn.Embedding(468, 64)
67
  self.region_emb = nn.Embedding(9, 32)
68
  self.coord_proj = nn.Linear(3, 32)
69
-
70
  self.neck = nn.Sequential(
71
  nn.Linear(192, 256), nn.BatchNorm1d(256), nn.LeakyReLU(0.2), nn.Dropout(0.3),
72
  nn.Linear(256, 128), nn.LeakyReLU(0.2)
73
  )
74
- # Heads
75
- self.head_gate = nn.Linear(128, 1) # Ignored in this version (using Location Net instead)
76
  self.head_tech = nn.Linear(128, 3)
77
  self.head_dosage = nn.Linear(128, 8)
78
  self.head_depth = nn.Linear(128, 4)
@@ -92,7 +79,6 @@ class RegionAwareExpert(nn.Module):
92
  "product": self.head_prod(x)
93
  }
94
 
95
- # --- B. LOCATION FINDER (AnatomyLocationNet) [NEW!] ---
96
  class AnatomyLocationNet(nn.Module):
97
  def __init__(self):
98
  super().__init__()
@@ -101,7 +87,6 @@ class AnatomyLocationNet(nn.Module):
101
  self.region_emb = nn.Embedding(9, 32)
102
  self.density_proj = nn.Linear(1, 16)
103
  self.context_proj = nn.Linear(512, 32)
104
-
105
  self.neck = nn.Sequential(
106
  nn.Linear(176, 256), nn.BatchNorm1d(256), nn.LeakyReLU(0.2), nn.Dropout(0.4),
107
  nn.Linear(256, 128), nn.LeakyReLU(0.2), nn.Linear(128, 1)
@@ -109,14 +94,11 @@ class AnatomyLocationNet(nn.Module):
109
 
110
  def forward(self, coords, pids, rids, den, ctx):
111
  B, N, _ = coords.shape
112
- # FIX: Add .unsqueeze(1) before .expand
113
  c_emb = self.context_proj(ctx).unsqueeze(1).expand(-1, N, -1)
114
-
115
  combined = torch.cat([self.coord_proj(coords), self.point_id_emb(pids),
116
  self.region_emb(rids), self.density_proj(den), c_emb], dim=-1)
117
  return self.neck(combined.view(-1, 176)).view(B, N, 1)
118
 
119
- # --- C. BACKBONE (SmartClinicalNet - Feature Extractor) ---
120
  class GatedFusion(nn.Module):
121
  def __init__(self, geo_dim, context_dim):
122
  super().__init__()
@@ -161,12 +143,10 @@ DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
161
 
162
  print("🔄 Initializing SOTA Clinical System...")
163
 
164
- # 1. Analyzers
165
  mp_mesh = mp.solutions.face_mesh.FaceMesh(static_image_mode=True, max_num_faces=1)
166
  face_app = FaceAnalysis(name='buffalo_l', providers=['CPUExecutionProvider'])
167
  face_app.prepare(ctx_id=0, det_size=(640, 640))
168
 
169
- # 2. Static Data
170
  mp_edges = mp.solutions.face_mesh.FACEMESH_TESSELATION
171
  s, t = [], []
172
  for src, dst in mp_edges:
@@ -175,26 +155,21 @@ GLOBAL_EDGE_INDEX = torch.tensor([s, t], dtype=torch.long).to(DEVICE)
175
  STATIC_REGION_IDS = get_region_tensor(DEVICE).unsqueeze(0)
176
  STATIC_POINT_IDS = torch.arange(468, dtype=torch.long, device=DEVICE).unsqueeze(0)
177
 
178
- # 3. Load ALL 3 Models
179
  backbone = SmartClinicalNet(hidden=32, heads=2).to(DEVICE)
180
  expert_model = RegionAwareExpert().to(DEVICE)
181
- location_model = AnatomyLocationNet().to(DEVICE) # New Model
182
 
183
- # Load Weights
184
  if os.path.exists("smart_clinical_model.pth"):
185
  backbone.load_state_dict(torch.load("smart_clinical_model.pth", map_location=DEVICE), strict=False)
186
  backbone.eval()
187
- print("✅ Backbone Loaded")
188
 
189
  if os.path.exists("region_expert.pth"):
190
  expert_model.load_state_dict(torch.load("region_expert.pth", map_location=DEVICE))
191
  expert_model.eval()
192
- print("✅ Attribute Expert Loaded")
193
 
194
  if os.path.exists("anatomy_location.pth"):
195
  location_model.load_state_dict(torch.load("anatomy_location.pth", map_location=DEVICE))
196
  location_model.eval()
197
- print("✅ Location Finder Loaded")
198
 
199
  print("✅ System Ready.")
200
 
@@ -228,23 +203,26 @@ def apply_nms_indices(landmarks_np, probs_np, spacing_mm=12.0, threshold=0.5):
228
  sorted_idx = rest[survivors]
229
  return keep
230
 
 
 
 
 
 
231
  # ==========================================
232
  # 5. API ENDPOINT
233
  # ==========================================
234
  @app.get("/")
235
  def home():
236
- return {"message": "SOTA Clinical AI (Location + Attributes) V6"}
237
 
238
  @app.post("/predict")
239
  async def predict_injections(file: UploadFile = File(...)):
240
- # 1. Image Processing
241
  contents = await file.read()
242
  nparr = np.frombuffer(contents, np.uint8)
243
  img_bgr = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
244
  if img_bgr is None: return JSONResponse(status_code=400, content={"error": "Invalid image"})
245
  img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
246
 
247
- # 2. Extract Base Features
248
  res = mp_mesh.process(img_rgb)
249
  if not res.multi_face_landmarks:
250
  return JSONResponse(status_code=400, content={"error": "No face detected"})
@@ -258,34 +236,24 @@ async def predict_injections(file: UploadFile = File(...)):
258
  emb = torch.tensor(faces[0].embedding).float().to(DEVICE) if faces else torch.zeros(512).to(DEVICE)
259
  virt_prof = get_virtual_profile_norm(x_geo_norm.cpu()).to(DEVICE)
260
 
261
- # 3. Prepare Inputs
262
  data = Data(x=x_geo_norm, edge_index=GLOBAL_EDGE_INDEX, embedding=emb.unsqueeze(0), virt_prof=virt_prof.unsqueeze(0))
263
-
264
- # --- CALCULATE DENSITY (Required for new model) ---
265
  density = calculate_density(x_geo_norm, GLOBAL_EDGE_INDEX).unsqueeze(0).to(DEVICE)
266
 
267
  with torch.no_grad():
268
- # A. Run Location Finder (The 94% Model)
269
- # Inputs: Coords, IDs, Regions, Density, Context
270
  logits_loc = location_model(x_geo_norm.unsqueeze(0), STATIC_POINT_IDS, STATIC_REGION_IDS, density, emb.unsqueeze(0))
271
  probs_loc = torch.sigmoid(logits_loc).squeeze().cpu().numpy()
272
 
273
- # B. Run Attribute Expert (Only if we need attributes)
274
  smart_features = backbone.get_features(data).unsqueeze(0)
275
  coords_input = x_geo_norm.unsqueeze(0)
276
  preds = expert_model(smart_features, coords_input, STATIC_POINT_IDS, STATIC_REGION_IDS)
277
 
278
- # Attributes
279
  prob_t = torch.softmax(preds['tech'], dim=-1).squeeze().cpu().numpy()
280
  prob_d = torch.softmax(preds['dosage'], dim=-1).squeeze().cpu().numpy()
281
  prob_de = torch.softmax(preds['depth'], dim=-1).squeeze().cpu().numpy()
282
  prob_p = torch.softmax(preds['product'], dim=-1).squeeze().cpu().numpy()
283
 
284
- # 4. Post-Processing & NMS
285
  h, w, _ = img_bgr.shape
286
  pixel_coords = np.array([[p.x*w, p.y*h, p.z*w] for p in lms[:468]])
287
-
288
- # Use the NEW high-accuracy probabilities for NMS
289
  optimal_indices = apply_nms_indices(pixel_coords, probs_loc, spacing_mm=12.0, threshold=0.4)
290
 
291
  classes_tech = ["Bolus", "Fanning", "Microbolus"]
@@ -295,15 +263,25 @@ async def predict_injections(file: UploadFile = File(...)):
295
 
296
  optimal_list = []
297
  for idx in optimal_indices:
 
 
 
298
  pt_info = {
299
- "point_id": idx,
300
  "confidence": float(f"{probs_loc[idx]:.4f}"),
301
  "coordinates": {"x": lms[idx].x, "y": lms[idx].y},
302
  "attributes": {
303
- "technique": classes_tech[np.argmax(prob_t[idx])],
304
- "dosage": classes_dosage[np.argmax(prob_d[idx])],
305
- "depth": classes_depth[np.argmax(prob_de[idx])],
306
- "product": classes_prod[np.argmax(prob_p[idx])]
 
 
 
 
 
 
 
307
  }
308
  }
309
  optimal_list.append(pt_info)
@@ -312,7 +290,7 @@ async def predict_injections(file: UploadFile = File(...)):
312
 
313
  return {
314
  "status": "success",
315
- "message": "SOTA V6: Dual-Model Active",
316
  "summary": {
317
  "total_optimal": len(optimal_list),
318
  "max_confidence": float(probs_loc.max())
 
32
  if p < 468: region_map[p] = region_id
33
  return region_map
34
 
 
35
  def calculate_density(coords, edge_index):
36
  # Pure PyTorch version (No extra dependencies required)
37
  row, col = edge_index
 
 
38
  dist = torch.norm(coords[row] - coords[col], dim=1)
 
 
39
  sum_dist = torch.zeros(468, device=coords.device)
40
  sum_dist.scatter_add_(0, row, dist)
 
 
41
  count = torch.zeros(468, device=coords.device)
42
  count.scatter_add_(0, row, torch.ones_like(dist))
 
 
43
  mean_dist = sum_dist / (count + 1e-6)
44
  density = 1.0 / (mean_dist + 1e-6)
 
45
  return density.unsqueeze(1) # [1, 468, 1]
46
 
47
  # ==========================================
48
  # 2. MODEL ARCHITECTURES
49
  # ==========================================
50
 
 
51
  class RegionAwareExpert(nn.Module):
52
  def __init__(self):
53
  super().__init__()
 
55
  self.point_id_emb = nn.Embedding(468, 64)
56
  self.region_emb = nn.Embedding(9, 32)
57
  self.coord_proj = nn.Linear(3, 32)
 
58
  self.neck = nn.Sequential(
59
  nn.Linear(192, 256), nn.BatchNorm1d(256), nn.LeakyReLU(0.2), nn.Dropout(0.3),
60
  nn.Linear(256, 128), nn.LeakyReLU(0.2)
61
  )
62
+ self.head_gate = nn.Linear(128, 1)
 
63
  self.head_tech = nn.Linear(128, 3)
64
  self.head_dosage = nn.Linear(128, 8)
65
  self.head_depth = nn.Linear(128, 4)
 
79
  "product": self.head_prod(x)
80
  }
81
 
 
82
  class AnatomyLocationNet(nn.Module):
83
  def __init__(self):
84
  super().__init__()
 
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
 
95
  def forward(self, coords, pids, rids, den, ctx):
96
  B, N, _ = coords.shape
 
97
  c_emb = self.context_proj(ctx).unsqueeze(1).expand(-1, N, -1)
 
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)
101
 
 
102
  class GatedFusion(nn.Module):
103
  def __init__(self, geo_dim, context_dim):
104
  super().__init__()
 
143
 
144
  print("🔄 Initializing SOTA Clinical System...")
145
 
 
146
  mp_mesh = mp.solutions.face_mesh.FaceMesh(static_image_mode=True, max_num_faces=1)
147
  face_app = FaceAnalysis(name='buffalo_l', providers=['CPUExecutionProvider'])
148
  face_app.prepare(ctx_id=0, det_size=(640, 640))
149
 
 
150
  mp_edges = mp.solutions.face_mesh.FACEMESH_TESSELATION
151
  s, t = [], []
152
  for src, dst in mp_edges:
 
155
  STATIC_REGION_IDS = get_region_tensor(DEVICE).unsqueeze(0)
156
  STATIC_POINT_IDS = torch.arange(468, dtype=torch.long, device=DEVICE).unsqueeze(0)
157
 
 
158
  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()
 
173
 
174
  print("✅ System Ready.")
175
 
 
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 (V7 Full Data Response)"}
217
 
218
  @app.post("/predict")
219
  async def predict_injections(file: UploadFile = File(...)):
 
220
  contents = await file.read()
221
  nparr = np.frombuffer(contents, np.uint8)
222
  img_bgr = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
223
  if img_bgr is None: return JSONResponse(status_code=400, content={"error": "Invalid image"})
224
  img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
225
 
 
226
  res = mp_mesh.process(img_rgb)
227
  if not res.multi_face_landmarks:
228
  return JSONResponse(status_code=400, content={"error": "No face detected"})
 
236
  emb = torch.tensor(faces[0].embedding).float().to(DEVICE) if faces else torch.zeros(512).to(DEVICE)
237
  virt_prof = get_virtual_profile_norm(x_geo_norm.cpu()).to(DEVICE)
238
 
 
239
  data = Data(x=x_geo_norm, edge_index=GLOBAL_EDGE_INDEX, embedding=emb.unsqueeze(0), virt_prof=virt_prof.unsqueeze(0))
 
 
240
  density = calculate_density(x_geo_norm, GLOBAL_EDGE_INDEX).unsqueeze(0).to(DEVICE)
241
 
242
  with torch.no_grad():
 
 
243
  logits_loc = location_model(x_geo_norm.unsqueeze(0), STATIC_POINT_IDS, STATIC_REGION_IDS, density, emb.unsqueeze(0))
244
  probs_loc = torch.sigmoid(logits_loc).squeeze().cpu().numpy()
245
 
 
246
  smart_features = backbone.get_features(data).unsqueeze(0)
247
  coords_input = x_geo_norm.unsqueeze(0)
248
  preds = expert_model(smart_features, coords_input, STATIC_POINT_IDS, STATIC_REGION_IDS)
249
 
 
250
  prob_t = torch.softmax(preds['tech'], dim=-1).squeeze().cpu().numpy()
251
  prob_d = torch.softmax(preds['dosage'], dim=-1).squeeze().cpu().numpy()
252
  prob_de = torch.softmax(preds['depth'], dim=-1).squeeze().cpu().numpy()
253
  prob_p = torch.softmax(preds['product'], dim=-1).squeeze().cpu().numpy()
254
 
 
255
  h, w, _ = img_bgr.shape
256
  pixel_coords = np.array([[p.x*w, p.y*h, p.z*w] for p in lms[:468]])
 
 
257
  optimal_indices = apply_nms_indices(pixel_coords, probs_loc, spacing_mm=12.0, threshold=0.4)
258
 
259
  classes_tech = ["Bolus", "Fanning", "Microbolus"]
 
263
 
264
  optimal_list = []
265
  for idx in optimal_indices:
266
+ # Extract distributions for this specific point
267
+ p_t, p_d, p_de, p_p = prob_t[idx], prob_d[idx], prob_de[idx], prob_p[idx]
268
+
269
  pt_info = {
270
+ "point_id": int(idx),
271
  "confidence": float(f"{probs_loc[idx]:.4f}"),
272
  "coordinates": {"x": lms[idx].x, "y": lms[idx].y},
273
  "attributes": {
274
+ # Top Picks
275
+ "technique": classes_tech[np.argmax(p_t)],
276
+ "dosage": classes_dosage[np.argmax(p_d)],
277
+ "depth": classes_depth[np.argmax(p_de)],
278
+ "product": classes_prod[np.argmax(p_p)],
279
+
280
+ # Full Probabilities (Fixes the 0% UI bug)
281
+ "technique_probs": get_top_probs(p_t, classes_tech),
282
+ "dosage_probs": get_top_probs(p_d, classes_dosage),
283
+ "depth_probs": get_top_probs(p_de, classes_depth),
284
+ "product_probs": get_top_probs(p_p, classes_prod)
285
  }
286
  }
287
  optimal_list.append(pt_info)
 
290
 
291
  return {
292
  "status": "success",
293
+ "message": "SOTA V7: Full Stats Active",
294
  "summary": {
295
  "total_optimal": len(optimal_list),
296
  "max_confidence": float(probs_loc.max())