",color="red",lw=0.8))
axes[1].axis("off"); axes[1].set_title("Centerline + FFR markers", fontsize=11, color="#1a6fa8")
# Panel 3: diameter profile
if len(arc_pos)>0:
pos_mm=arc_pos/px_mm
axes[2].plot(pos_mm,diams_px/px_mm,color="#cce0f0",lw=0.8,label="Raw")
axes[2].plot(pos_mm,diams_sm/px_mm,color="#1a6fa8",lw=1.8,label="Smoothed")
axes[2].axhline(ref_mm,color="#0d9e6e",ls="--",lw=1.2,label=f"Ref {ref_mm:.2f}mm")
for st in stenoses:
axes[2].axvline(st["position_mm"],color="red",ls=":",lw=1.,alpha=0.8)
axes[2].annotate(f'{st["pct_ds"]:.0f}%DS\nFFR≈{st["ffr"]:.2f}',
xy=(st["position_mm"],st["d_mm"]),
xytext=(st["position_mm"]+2,st["d_mm"]+ref_mm*0.15),
color="red",fontsize=7,fontweight="bold",
arrowprops=dict(arrowstyle="->",color="red",lw=0.7))
axes[2].set_xlabel("Position (mm)",fontsize=9)
axes[2].set_ylabel("Diameter (mm)",fontsize=9)
axes[2].legend(fontsize=8); axes[2].grid(alpha=0.3); axes[2].set_ylim(0)
axes[2].set_title("Lumen diameter profile", fontsize=11, color="#1a6fa8")
ffr_label = ("⚠ ISCHEMIC" if global_ffr<=0.80 else "✓ Non-ischemic")
fig.suptitle(f"FFR Analysis — Global FFR: {global_ffr:.3f} {ffr_label} | "
f"Stenoses: {len(stenoses)} | Ref: {ref_mm:.2f} mm",
fontsize=12,fontweight="bold",color="#1a2533",y=1.01)
plt.tight_layout()
# Convert figure to numpy array — compatible with all matplotlib versions
fig.canvas.draw()
w, h = fig.canvas.get_width_height()
try:
# matplotlib >= 3.8
buf = np.frombuffer(fig.canvas.buffer_rgba(), dtype=np.uint8)
buf = buf.reshape(h, w, 4)[:, :, :3] # drop alpha channel
except AttributeError:
# matplotlib < 3.8 fallback
buf = np.frombuffer(fig.canvas.tostring_rgb(), dtype=np.uint8)
buf = buf.reshape(h, w, 3)
plt.close(fig)
result = {
"global_ffr" : round(global_ffr,3),
"ischemic" : global_ffr<=0.80,
"n_stenoses" : len(stenoses),
"stenoses" : stenoses,
"ref_diam_mm" : round(ref_mm,2),
"vessel_len_mm": round(float(arc_pos[-1]/px_mm),1) if len(arc_pos)>0 else 0,
"n_unlocalized": len(unloc),
}
return result, buf
# ─────────────────────────────────────────────────────────────────────────────
# MODULE 6 — nnUNet OVERLAY
# ─────────────────────────────────────────────────────────────────────────────
# 26 maximally distinct colours — one per ARCADE class (0 = background, black)
# Generated by evenly spacing hue across HSV wheel at two brightness levels,
# then manually verified for distinctness. Each colour is unique.
# Format: (R, G, B)
ARCADE_COLOURS = {
0 : ( 0, 0, 0), # background — black (not rendered)
1 : (255, 0, 0), # LMCA — red
2 : ( 0, 120, 255), # LAD proximal — azure blue
3 : ( 0, 220, 0), # LAD mid — lime green
4 : (255, 165, 0), # LAD distal — orange
5 : (180, 0, 255), # D1 — violet
6 : ( 0, 230, 230), # D2 — cyan
7 : (255, 230, 0), # D3 — yellow
8 : (255, 0, 180), # Septal — hot pink
9 : ( 0, 180, 80), # LAD var — jade green
10: (255, 100, 0), # LCX proximal — deep orange
11: ( 40, 40, 255), # LCX mid — royal blue
12: ( 0, 255, 140), # LCX distal — spring green
13: (220, 0, 80), # OM1 — crimson
14: ( 0, 200, 255), # OM2 — sky blue
15: (200, 255, 0), # OM3 — chartreuse
16: (140, 0, 200), # LCX var — purple
17: (255, 200, 120), # RCA proximal — peach
18: ( 80, 255, 80), # RCA mid — bright green
19: (255, 60, 200), # RCA distal — rose pink
20: ( 0, 80, 200), # PDA — cobalt blue
21: (200, 140, 0), # PLV — dark amber
22: ( 80, 200, 200), # AM branch — teal
23: (255, 255, 120), # Conus — light yellow
24: (160, 60, 0), # SAN artery — brown
25: (180, 180, 180), # Other — silver grey
}
ARCADE_LABELS = {
1:"LMCA", 2:"LAD prox", 3:"LAD mid", 4:"LAD dist",
5:"D1", 6:"D2", 7:"D3", 8:"Septal",
9:"LAD var", 10:"LCX prox", 11:"LCX mid", 12:"LCX dist",
13:"OM1", 14:"OM2", 15:"OM3", 16:"LCX var",
17:"RCA prox",18:"RCA mid", 19:"RCA dist", 20:"PDA",
21:"PLV", 22:"AM", 23:"Conus", 24:"SAN",
25:"Other",
}
def render_nnunet_overlay(frame_rgb, seg_map, alpha=0.55):
"""
Renders nnUNet segmentation overlay.
Handles two cases:
- Binary output (classes 0,1): single vessel colour overlay
- 26-class output (ARCADE SYNTAX): full colour-coded overlay with legend
"""
unique_cls = [c for c in np.unique(seg_map) if c > 0]
overlay = frame_rgb.copy().astype(np.float32)
present_classes = []
# ── Binary case (0/1 only) ────────────────────────────────────────────────
if len(unique_cls) == 1 and unique_cls[0] == 1:
mask = (seg_map == 1)
col = np.array([0, 200, 255], dtype=np.float32) # cyan for vessel
overlay[mask] = overlay[mask] * (1 - alpha) + col * alpha
result = overlay.astype(np.uint8)
# Simple label
cv2.putText(result, "Vessel mask (binary)",
(8, 20), cv2.FONT_HERSHEY_SIMPLEX,
0.5, (0, 200, 255), 1, cv2.LINE_AA)
return result
# ── 26-class case ─────────────────────────────────────────────────────────
for cls_id, rgb_col in ARCADE_COLOURS.items():
if cls_id == 0:
continue
mask = (seg_map == cls_id)
if not mask.any():
continue
col = np.array(rgb_col, dtype=np.float32)
overlay[mask] = overlay[mask] * (1 - alpha) + col * alpha
present_classes.append(cls_id)
result = overlay.astype(np.uint8)
# Compact legend — bottom-left, only present classes, max 13 rows
if present_classes:
try:
n_rows = min(len(present_classes), 13)
row_h = 18
legend_h = n_rows * row_h + 10
legend_w = 148
H, W = result.shape[:2]
x0, y0 = 6, H - legend_h - 6
sub = result[y0:y0+legend_h, x0:x0+legend_w].astype(np.float32)
result[y0:y0+legend_h, x0:x0+legend_w] = (sub * 0.4).astype(np.uint8)
for i, cls_id in enumerate(present_classes[:13]):
r, g, b = ARCADE_COLOURS[cls_id]
lbl = ARCADE_LABELS.get(cls_id, f"cls{cls_id}")
yy = y0 + 6 + i * row_h
cv2.rectangle(result, (x0+4, yy), (x0+16, yy+11), (b, g, r), -1)
cv2.rectangle(result, (x0+4, yy), (x0+16, yy+11), (255,255,255), 1)
cv2.putText(result, lbl, (x0+20, yy+9),
cv2.FONT_HERSHEY_SIMPLEX, 0.35,
(255,255,255), 1, cv2.LINE_AA)
except Exception:
pass
return result
# ─────────────────────────────────────────────────────────────────────────────
# MODULE 7 — SYNTAX SCORE (simplified anatomical estimate)
# ─────────────────────────────────────────────────────────────────────────────
def compute_syntax_score(stenoses, seg_map):
"""
Simplified SYNTAX estimate:
Each stenosis contributes a weight based on DS% and
anatomical territory (estimated from nnUNet class ID).
Reference: Sianos et al. EuroIntervention 2005.
Note: true SYNTAX requires angiographic expertise; this is
an automated surrogate for display purposes.
"""
if not stenoses:
return 0.0, "Low (0)"
base_score = 0.
for st in stenoses:
ds = st["pct_ds"] / 100
# Weight by position: proximal vessels score higher
pos_w = max(0.5, 1.5 - st["position_mm"] / 100)
# Occlusion multiplier
occ_w = 5.0 if ds >= 0.99 else (2.5 if ds >= 0.70 else 1.5 if ds >= 0.50 else 1.0)
base_score += pos_w * occ_w * (1 + ds)
# Add segment count from nnUNet
n_segs = len(np.unique(seg_map)) - 1 # exclude background
base_score += n_segs * 0.3
score = round(base_score, 1)
if score < 23: tier = "Low (<23)"
elif score < 33: tier = "Intermediate (23–32)"
else: tier = "High (≥33)"
return score, tier
# ─────────────────────────────────────────────────────────────────────────────
# JSON ENCODER — handles numpy float32/int64 etc.
# ─────────────────────────────────────────────────────────────────────────────
class _NpEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, np.integer): return int(obj)
if isinstance(obj, np.floating): return float(obj)
if isinstance(obj, np.ndarray): return obj.tolist()
if isinstance(obj, np.bool_): return bool(obj)
return super().default(obj)
def _to_json(obj) -> str:
return json.dumps(obj, indent=2, cls=_NpEncoder)
# ─────────────────────────────────────────────────────────────────────────────
# MAIN PIPELINE FUNCTION
# ─────────────────────────────────────────────────────────────────────────────
def analyse(video_path, sten_conf, seg_conf, px_per_mm_override, progress=gr.Progress()):
"""
Full pipeline. Called by Gradio on button click.
Returns: (status_html, frame_img, seg_img, ffr_img, metrics_html, json_str)
"""
if video_path is None:
return (_status("error","No video uploaded."), None, None, None, "", "")
FFR_CFG["px_per_mm"] = px_per_mm_override
try:
# Step 1 — keyframe
progress(0.10, desc="Extracting best frame…")
frame_rgb, frame_idx, frame_score, total_frames = extract_best_frame(video_path)
# Step 2 — stenosis detection
progress(0.30, desc="Running Mask2Former…")
detections = run_mask2former(frame_rgb, sten_conf)
# Compute vessel overlap for each detection
binary_mask_for_overlap = np.ones((512,512),dtype=np.uint8) # placeholder until ResUNet
for det in detections:
b=det.get("bbox",[]);
if len(b)<4: det["overlap"]=0.; continue
x1,y1,x2,y2=[int(v) for v in b]
area=(x2-x1)*(y2-y1)
det["overlap"]=min(float(binary_mask_for_overlap[y1:y2,x1:x2].sum())/max(area,1),1.)
# Step 3a — binary segmentation (ResUNet)
progress(0.50, desc="Running ResUNet…")
binary_mask, binary_overlay = run_resunet(frame_rgb)
print(f"[ResUNet] binary_mask sum={binary_mask.sum()} "
f"unique={np.unique(binary_mask)} "
f"vessel%={100*binary_mask.mean():.1f}")
# Recompute overlap with real mask
for det in detections:
b=det.get("bbox",[])
if len(b)<4: continue
x1,y1,x2,y2=[int(v) for v in b]
area=(x2-x1)*(y2-y1)
det["overlap"]=float(binary_mask[y1:y2,x1:x2].sum())/max(area,1)
# Step 3b — YOLOv8m-seg 26-class segmentation
progress(0.65, desc="Running seg…")
seg_map = run_yolo_seg(frame_rgb, seg_conf)
seg_overlay = render_nnunet_overlay(frame_rgb, seg_map)
# Step 4 — FFR
progress(0.80, desc="Computing FFR…")
ffr_result, ffr_fig = run_ffr(frame_rgb, binary_mask, detections)
if "error" in ffr_result:
return (_status("warn", ffr_result["error"]),
frame_rgb, binary_overlay, seg_overlay, None,
_metrics_html(ffr_result, {}, frame_idx, total_frames),
_to_json(ffr_result))
# Step 5 — SYNTAX score
progress(0.92, desc="Computing SYNTAX score…")
syntax_score, syntax_tier = compute_syntax_score(ffr_result["stenoses"], seg_map)
ffr_result["syntax_score"] = syntax_score
ffr_result["syntax_tier"] = syntax_tier
progress(1.00, desc="Done.")
status = _status("ok",
f"Analysis complete — Frame {frame_idx}/{total_frames} "
f"(quality score {frame_score:.1f})")
metrics = _metrics_html(ffr_result, {"score":frame_score,"idx":frame_idx},
frame_idx, total_frames)
return (status, frame_rgb, binary_overlay, seg_overlay, ffr_fig, metrics,
_to_json(ffr_result))
except FileNotFoundError as e:
return (_status("error", str(e)), None, None, None, None,
_metrics_html({},{},0,0),
_to_json({"error":str(e)}))
except Exception as e:
import traceback
tb = traceback.format_exc()
return (_status("error", f"{type(e).__name__}: {e}"), None, None, None, None,
_metrics_html({},{},0,0),
_to_json({"error":str(e),"traceback":tb}))
# ─────────────────────────────────────────────────────────────────────────────
# HTML HELPERS
# ─────────────────────────────────────────────────────────────────────────────
def _status(kind: str, msg: str) -> str:
colours = {"ok":"#0d9e6e","warn":"#d98b00","error":"#d64040"}
icons = {"ok":"✓","warn":"⚠","error":"✗"}
col = colours.get(kind, "#1a6fa8")
ico = icons.get(kind, "•")
return (f''
f'{ico} {msg}
')
def _metrics_html(ffr_result, frame_info, frame_idx, total) -> str:
if not ffr_result or "error" in ffr_result:
return 'Metrics
Run analysis to see results.
'
f'| #{i+1} | '
f'{st["position_mm"]:.1f} mm | '
f'{st["pct_ds"]:.1f}% | '
f'{st["d_mm"]:.2f} mm | '
f'{st["ffr"]:.3f} | '
f'
'
)
return f"""
Key metrics
{gffr if gffr!="—" else "—"}
Global FFR
SYNTAX tier: {stier} |
Frame {frame_idx}/{total} |
Scale: {FFR_CFG["px_per_mm"]} px/mm
{'Stenosis table
| # | Position | DS% | Min diam | FFR |
|---|
' + stenosis_rows + '
Angiographic video
')
video_input = gr.Video(
label="Upload XCA video (mp4 / avi / dicom-wrapped)",
elem_id="upload-zone",
height=220,
)
btn_demo1 = gr.Button("▶ XCA Video Run", variant="secondary", size="sm")
gr.HTML('
Click to load the demo XCA video, or upload your own above.
')
gr.HTML('
Model controls
')
sten_conf = gr.Slider(
minimum=0.05, maximum=0.95, value=0.30, step=0.05,
label="Stenosis confidence threshold (Mask2Former)",
info="Detections below this confidence are discarded",
)
seg_conf = gr.Slider(
minimum=0.05, maximum=0.95, value=0.25, step=0.05,
label="Segmentation confidence threshold",
info="Detections below this confidence are discarded ",
)
px_per_mm = gr.Slider(
minimum=2.0, maximum=6.0, value=3.75, step=0.25,
label="Scale (px / mm)",
info="ARCADE default = 3.75 px/mm (0.267 mm/px). Adjust if using non-ARCADE data.",
)
gr.HTML('
Pipeline overview
- Best frame extracted from video
- Stenosis detection — Mask2Former
- Coronary segmentation — 26-class anatomy labelling
- Binary vessel mask — ResUNet
- FFR estimation — QFR v4
- SYNTAX score computation
FFR threshold: ≤ 0.80 → ischemic
Formula: 1 − (0.33·DS + 0.60·DS²)
Reference: Tu et al. JACC 2016
Model checkpoints
Auto-downloaded from
MuhammadAdil63/
angio-ai-checkpoints
mask2former_best.pth
binary_best.pth
best.pt
""")
with gr.Row():
with gr.Column(scale=2):
btn_analyse = gr.Button("▶ Run analysis",
elem_id="btn-analyse", variant="primary")
with gr.Column(scale=1):
btn_reset = gr.Button("↺ Clear", elem_id="btn-reset")
status_out = gr.HTML(
'Upload a video and press Run analysis.
')
# ── PAGE 2: RESULTS ───────────────────────────────────────────────
with gr.TabItem("📊 Results", id="tab-results"):
gr.HTML('')
status_out2 = gr.HTML()
metrics_out = gr.HTML()
with gr.Row():
with gr.Column():
gr.HTML('Best frame extracted
')
frame_out = gr.Image(label="Best keyframe", height=300)
with gr.Column():
gr.HTML('Binary vessel mask (ResUNet)
')
binary_out = gr.Image(label="Green overlay — vessel pixels", height=300)
with gr.Row():
with gr.Column():
gr.HTML('Coronary segmentation (26-class)
')
seg_out = gr.Image(label="26-class coronary overlay", height=300)
with gr.Column():
gr.HTML('FFR analysis
')
ffr_out = gr.Image(label="FFR pipeline output", height=300)
with gr.Row():
with gr.Column():
gr.HTML('Full results (JSON)
')
json_out = gr.Code(label="", language="json", lines=12)
# Footer
gr.HTML("""
""")
# ── WIRING ────────────────────────────────────────────────────────────
def run_and_switch(video, sc, sg, px):
return analyse(video, sc, sg, px)
def load_demo1():
return _get_demo_video(DEMO_VIDEO_1_NAME)
btn_demo1.click(fn=load_demo1, inputs=[], outputs=[video_input])
btn_analyse.click(
fn=run_and_switch,
inputs=[video_input, sten_conf, seg_conf, px_per_mm],
outputs=[status_out, frame_out, binary_out, seg_out, ffr_out, metrics_out, json_out],
).then(
# Copy status HTML to results page status box
fn=lambda s: s,
inputs=[status_out],
outputs=[status_out2],
).then(
# Switch to Results tab via JavaScript
fn=None,
js="""
() => {
const tabs = document.querySelectorAll('.tab-nav button');
if (tabs.length >= 2) tabs[1].click();
}
""",
)
btn_reset.click(
fn=lambda: (
None, None, None, None, None,
'Upload a video and press Run analysis.
',
"", "",
),
outputs=[video_input, frame_out, binary_out, seg_out, ffr_out,
status_out, metrics_out, json_out],
)
# ─────────────────────────────────────────────────────────────────────────────
# ENTRY POINT
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
demo.launch(
server_name="0.0.0.0",
server_port=7860,
share=True,
)