",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")
if not detections:
ffr_title = "No stenosis detected"
elif not stenoses:
ffr_title = f"{len(detections)} detected, none quantifiable on the measured branch"
else:
ffr_label = ("⚠ ISCHEMIC" if global_ffr<=0.80 else "✓ Non-ischemic")
ffr_title = f"Branch FFR: {global_ffr:.3f} {ffr_label} | Quantified: {len(stenoses)}/{len(detections)}"
fig.suptitle(f"FFR Analysis — {ffr_title} | 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),
"n_detected" : len(detections),
"unlocalized" : [{"segment":u.get("segment","—"),"score":round(float(u.get("score",0)),3),
"bbox":[int(v) for v in u.get("bbox",[])],"reason":u.get("reason","")} for u in unloc],
"n_significant": sum(1 for s_ in stenoses if s_["pct_ds"]>=50),
}
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 = dict(SYNTAX_SEGMENTS)
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)
# ─────────────────────────────────────────────────────────────────────────────
# SYNTAX segment weighting factors (Sianos et al., EuroIntervention 2005, Table 1).
# Keys = ARCADE label ids (see SYNTAX_SEGMENTS). Verify against syntaxscore.org before clinical use.
_SYNTAX_W_RIGHT = {1:1, 2:1, 3:1, 4:1, 5:5, 6:3.5, 7:2.5, 8:1, 9:1, 10:1, 11:0.5, 12:0.5,
13:1.5, 14:1, 15:1, 16:0.5, 17:0.5, 18:0.5, 19:0, 20:0.5, 21:0.5, 22:0.5, 23:0.5,
24:1, 25:0.5}
_SYNTAX_W_LEFT = {1:0, 2:0, 3:0, 4:0, 5:6, 6:3.5, 7:2.5, 8:1, 9:1, 10:1, 11:0.5, 12:0.5,
13:2.5, 14:1, 15:1, 16:1.5, 17:1, 18:1, 19:1, 20:0, 21:0, 22:0, 23:0,
24:1, 25:1}
def segment_for_detection(det, seg_map, dilate_px=7):
"""Majority anatomical label under the (slightly dilated) stenosis mask."""
m = det.get("mask")
if m is None:
x1,y1,x2,y2 = [int(v) for v in det["bbox"]]
m = np.zeros_like(seg_map, dtype=np.uint8); m[y1:y2+1, x1:x2+1] = 1
m = cv2.dilate(m.astype(np.uint8), np.ones((dilate_px, dilate_px), np.uint8)) > 0
labs = seg_map[m]; labs = labs[labs > 0]
if labs.size == 0:
return 0, "unassigned"
sid = int(np.bincount(labs).argmax())
return sid, SYNTAX_SEGMENTS.get(sid, "unassigned")
def compute_syntax_score(stenoses, dominance="right"):
"""
Anatomical SYNTAX estimate: sum over lesions with DS >= 50% of
segment weight x 2 (non-occlusive) or x 5 (occlusion, DS >= 99%).
NOT scored yet: adverse lesion characteristics (bifurcation, ostial, length > 20 mm,
tortuosity, calcification, thrombus, diffuse disease) -> true SYNTAX is >= this value.
Lesions without an assigned segment are listed but not scored.
"""
W = _SYNTAX_W_LEFT if dominance == "left" else _SYNTAX_W_RIGHT
score, scored, unassigned = 0.0, 0, 0
for st in stenoses:
if st["pct_ds"] < 50:
st["syntax_points"] = 0.0; continue
sid = st.get("segment_id", 0)
if not sid:
st["syntax_points"] = None; unassigned += 1; continue
pts = W.get(sid, 0) * (5 if st["pct_ds"] >= 99 else 2)
st["syntax_points"] = pts; score += pts; scored += 1
score = round(score, 1)
tier = "Low (≤22)" if score <= 22 else ("Intermediate (23–32)" if score <= 32 else "High (≥33)")
return score, tier, scored, unassigned
# ─────────────────────────────────────────────────────────────────────────────
# 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, dominance="right", 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="Detecting stenoses (YOLOv8-seg ensemble)…")
detections = run_stenosis(frame_rgb, sten_conf)
# 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}")
# Vessel overlap of each detection (used by FFR to drop off-vessel detections)
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)
for det in detections:
det["segment_id"], det["segment"] = segment_for_detection(det, 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:
ffr_result["n_detected"] = len(detections)
ffr_result["detections"] = [{"segment":d.get("segment","—"),"score":round(d["score"],3),
"bbox":d["bbox"]} for d in detections]
return (_status("warn", "Vessel centreline could not be traced — FFR/QCA not available. "
+ 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, n_scored, n_unassigned = compute_syntax_score(ffr_result["stenoses"], dominance)
ffr_result["syntax_score"] = syntax_score
ffr_result["syntax_tier"] = syntax_tier
ffr_result["syntax_lesions_scored"] = n_scored
ffr_result["syntax_lesions_unassigned"] = n_unassigned
ffr_result["dominance"] = dominance
progress(1.00, desc="Done.")
n_det = len(detections)
status = _status("ok",
("No stenosis detected — " if n_det == 0 else f"{n_det} stenosis candidate(s) detected — ")
+ f"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 _badge(val, lbl, cls="", small=False):
st = ' style="font-size:14px"' if small else ""
return (f'')
def _metrics_html(ffr_result, frame_info, frame_idx, total) -> str:
if not ffr_result:
return 'Metrics
Run analysis to see results.
Measured branch: '
f'ref diameter {ffr_result.get("ref_diam_mm","—")} mm, length {ffr_result.get("vessel_len_mm","—")} mm '
f' | Frame {frame_idx}/{total} | Scale {FFR_CFG["px_per_mm"]} px/mm
')
return f"""
Key metrics
{_badge("0", "Stenoses detected", "ok")}
{_badge("✓ None", "Stenosis", "ok", small=True)}
{_badge("N/A", "FFR (no lesion)")}
{_badge("0", "SYNTAX (anatomical)", "ok")}
No stenosis detected at the current
confidence threshold. This is not a clinical exclusion of coronary artery disease.
{extra}
"""
# ── Case B: detections, but QCA/FFR failed (no centreline) ───────────────
if err:
rows = "".join(f'| #{i+1} | {d["segment"]} | '
f'{d["score"]:.2f} |
'
for i, d in enumerate(ffr_result.get("detections", [])))
return f"""
Key metrics
{_badge(n_det, "Stenoses detected", "borderline")}
{_badge("N/A", "FFR / %DS", "", small=True)}
{_badge("N/A", "SYNTAX", "", small=True)}
Vessel centreline could not be traced, so
diameter stenosis, FFR and SYNTAX could not be computed.
Detections
| # |
Segment | Confidence |
|---|
{rows}
| #{i+1} | '
f'{st.get("segment","—")} | '
f'{st["score"]:.2f} | '
f'{st["pct_ds"]:.0f}% | '
f'{st["d_mm"]:.2f} mm | '
f'{st["ffr"]:.2f} | '
f'{pts_s} | '
f'quantified |
')
for j, u in enumerate(ffr_result.get("unlocalized", [])):
rows += (f'| #{n_q+j+1} | '
f'{u["segment"]} | {u["score"]:.2f} | '
f'— | — | — | '
f'— | {u["reason"]} |
')
th = "".join(f'{h} | ' for h in
["#", "Segment", "Conf.", "DS%", "Min diam", "FFR", "SYNTAX pts", "Status"])
return f"""
Key metrics
{_badge(n_det, "Stenoses detected", "borderline")}
{_badge(f"{n_sig}", "Significant (DS ≥ 50%)", sig_cls)}
{_badge(ffr_val, "Branch FFR (QFR est.)", ffr_cls)}
{_badge(isch_lbl, "Ischaemia", isch_cls, small=True)}
{_badge(f"{syn:g}", "SYNTAX (anatomical)", syn_cls)}
{_badge(f'{ffr_result.get("ref_diam_mm","—")} mm', "Ref diameter (measured branch)")}
{_badge(f'{ffr_result.get("vessel_len_mm","—")} mm', "Measured branch length")}
Quantified on measured branch: {n_q}/{n_det} |
SYNTAX tier: {stier} ({ffr_result.get("dominance","right")} dominance; lesions ≥50% DS only;
adverse lesion features not yet scored{"; " + str(ffr_result.get("syntax_lesions_unassigned")) + " lesion(s) without segment" if ffr_result.get("syntax_lesions_unassigned") else ""}) |
Frame {frame_idx}/{total} | Scale: {FFR_CFG["px_per_mm"]} px/mm
"""
# ─────────────────────────────────────────────────────────────────────────────
# GRADIO APP
# ─────────────────────────────────────────────────────────────────────────────
with gr.Blocks(css=CSS, title="AngioAI Pro") as demo:
# Header
gr.HTML(_header_html())
with gr.Tabs() as tabs:
# ── PAGE 1: UPLOAD ────────────────────────────────────────────────
with gr.TabItem("📤 Upload & Configure", id="tab-upload"):
gr.HTML('')
with gr.Row():
# Left — upload + controls
with gr.Column(scale=2):
gr.HTML('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.20, step=0.05,
label="Stenosis confidence threshold (YOLOv8-seg ensemble)",
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.",
)
dominance = gr.Radio(
choices=["right", "left"], value="right",
label="Coronary dominance (for SYNTAX weighting)",
)
gr.HTML('
Pipeline overview
- Best frame extracted from video
- Stenosis detection — YOLOv8-seg ensemble (3 models)
- 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
Stenosis: weights/stenosis/
yolov8x_seg.pt · yolov8l_sgd.pt · yolov8l_adamw.pt
Others auto-downloaded from
MuhammadAdil63/angio-ai-checkpoints
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, dom):
return analyse(video, sc, sg, px, dom)
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, dominance],
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=os.environ.get("SPACE_ID") is None, # public share link only when run locally
)