",
unsafe_allow_html=True,
)
cmp_dataset = st.selectbox(
"Dataset",
PRECOMPUTED_DATASETS,
format_func=lambda d: DATASET_DISPLAY.get(d, d),
key="cmp_dataset",
help="Both models are evaluated on the same dataset — this ensures the comparison is fair.",
)
# Expose cmp_a_dataset / cmp_b_dataset as aliases so downstream code is unchanged.
cmp_a_dataset = cmp_dataset
cmp_b_dataset = cmp_dataset
st.divider()
st.markdown(
"
⚠️ MONAI models only. Only architectures from the
MONAI framework are supported (UNet, Attention UNet, UNETR,
SwinUNETR, VNet). The architecture is inferred from the .pt
file structure; you must confirm the correct architecture
below.
A demonstration of pre-computed robustness evaluations across
5 MedSegBench datasets, 3 architectures
(UNet · Attention UNet · UNETR), and 3 training seeds
— browse instantly, no compute required.
""",
unsafe_allow_html=True,
)
try:
bundle = load_precomputed_bundle(explore_dataset, explore_arch, explore_seed)
except FileNotFoundError as _e:
st.error(str(_e))
st.stop()
except Exception as _e:
st.error(f"Failed to load bundle: {_e}")
st.stop()
card = bundle.card
card_df: pd.DataFrame = bundle_to_streamlit_eval_results(bundle)["df"]
artifacts: list[str] = bundle.artifacts
dataset_name: str = bundle.dataset
modality_name: str = bundle.modality
n_samples: int = bundle.n_samples
k1, k2, k3, k4 = st.columns(4)
with k1:
st.metric("Clean Dice", f"{card.clean_metrics.dice:.4f}")
with k2:
st.metric("Corrupted Dice", f"{card.corrupted_metrics_macro.dice:.4f}")
with k3:
hd95_val = card.corrupted_metrics_macro.hd95
st.metric(
"Corrupted HD95",
(
"N/A"
if hd95_val is None or np.isnan(float(hd95_val))
else f"{float(hd95_val):.4f}"
),
)
_exp_assess, _exp_thresholds, _exp_score, _exp_comps = _fresh_safety(card, card_df)
with k4:
st.metric("Robustness Score", f"{_exp_score:.2f}")
tab_card, tab_quality, tab_perturb, tab_dl = st.tabs(
[
"Robustness Card",
"Aggregate by Image Quality",
"Per-Perturbation",
"Downloads",
]
)
with tab_card:
_render_robustness_card_visual(
card,
_exp_assess,
_exp_thresholds,
_exp_score,
_exp_comps,
bundle_df=card_df,
)
with tab_quality:
st.markdown(
"
Aggregate view: each point is one "
"perturbation type × severity level. The line shows the binned mean; "
"the band captures variability across artifact types at similar "
"image quality.
",
unsafe_allow_html=True,
)
_render_quality_vs_seg_chart(card_df, artifacts, key_prefix="exp")
with tab_perturb:
st.markdown(
"
Per-perturbation view: x-axis is "
"signed severity (negative = sharpening, positive = degradation). "
"Use this to compare how different artifact families stress the "
"model.
Click the button below to open your browser's "
"print dialog. Select Save as PDF as the destination to export the "
"current page (including the robustness card tab) as a PDF document.
",
unsafe_allow_html=True,
)
_print_as_pdf_button("Print current page as PDF")
st.stop()
# ===========================================================================
# MODE 2: COMPARE MODELS
# ===========================================================================
if eval_mode == "Compare Models":
st.markdown(
f"""
Compare Models
Select two models evaluated on the same dataset
to view their robustness cards side-by-side with
per-metric win/loss indicators.
Per-perturbation severity curves for both "
"models. Select a dataset and perturbation type to compare how each "
"model degrades under increasing corruption.
",
unsafe_allow_html=True,
)
_df_a: pd.DataFrame = _ev_a["df"]
_df_b: pd.DataFrame = _ev_b["df"]
_art_opts_a = sorted(_df_a["artifact"].astype(str).unique().tolist())
_art_opts_b = sorted(_df_b["artifact"].astype(str).unique().tolist())
_art_union = sorted(set(_art_opts_a) | set(_art_opts_b))
_sel_art = st.selectbox(
"Perturbation type",
_art_union,
format_func=lambda a: ARTIFACT_LABELS.get(a, a),
key="cmp_sel_art",
)
_cmp_seg_choices = [
m
for m in ["dice", "hd95", "hd100", "asd"]
if m in _df_a.columns or m in _df_b.columns
]
_cmp_metric = st.selectbox(
"Segmentation metric",
_cmp_seg_choices,
format_func=lambda k: {
"dice": "Dice",
"hd95": "HD95",
"hd100": "HD100",
"asd": "ASD",
}.get(k, k),
key="cmp_metric",
)
_sub_a = _df_a[_df_a["artifact"] == _sel_art].sort_values("severity")
_sub_b = _df_b[_df_b["artifact"] == _sel_art].sort_values("severity")
if _PLOTLY:
_fig_cmp = go.Figure()
if not _sub_a.empty and _cmp_metric in _sub_a.columns:
_fig_cmp.add_trace(
go.Scatter(
x=_sub_a["severity"].tolist(),
y=_sub_a[_cmp_metric].tolist(),
mode="lines+markers",
name=f"A — {PRECOMPUTED_ARCHS.get(cmp_a_arch, cmp_a_arch)} / Seed {cmp_a_seed}",
line=dict(color="#0f766e", width=2.5),
)
)
if not _sub_b.empty and _cmp_metric in _sub_b.columns:
_fig_cmp.add_trace(
go.Scatter(
x=_sub_b["severity"].tolist(),
y=_sub_b[_cmp_metric].tolist(),
mode="lines+markers",
name=f"B — {PRECOMPUTED_ARCHS.get(cmp_b_arch, cmp_b_arch)} / Seed {cmp_b_seed}",
line=dict(color="#1d4ed8", width=2.5),
)
)
_fig_cmp.add_vline(
x=0.0, line_dash="dash", line_color="rgba(100,116,139,0.6)"
)
_fig_cmp.update_layout(
height=480,
template="plotly_white",
hovermode="x unified",
xaxis_title="Signed severity (− sharpen / + degrade)",
yaxis_title={
"dice": "Dice",
"hd95": "HD95 (vox)",
"hd100": "HD100 (vox)",
"asd": "ASD (vox)",
}.get(_cmp_metric, _cmp_metric),
legend=dict(orientation="h", yanchor="bottom", y=1.02, x=0),
margin=dict(l=20, r=20, t=20, b=20),
)
st.plotly_chart(_fig_cmp, use_container_width=True)
else:
st.info("Install plotly for interactive charts.")
with tab_dl_cmp:
st.markdown("#### Model A Downloads")
col1, col2 = st.columns(2)
with col1:
st.download_button(
"Download A — JSON",
data=_ev_a["card_json"],
file_name=f"card_A_{cmp_a_dataset}_{cmp_a_arch}_seed{cmp_a_seed}.json",
mime="application/json",
use_container_width=True,
)
with col2:
st.download_button(
"Download A — CSV",
data=_ev_a["df"].to_csv(index=False).encode(),
file_name=f"metrics_A_{cmp_a_dataset}_{cmp_a_arch}_seed{cmp_a_seed}.csv",
mime="text/csv",
use_container_width=True,
)
st.markdown("#### Model B Downloads")
col3, col4 = st.columns(2)
with col3:
st.download_button(
"Download B — JSON",
data=_ev_b["card_json"],
file_name=f"card_B_{cmp_b_dataset}_{cmp_b_arch}_seed{cmp_b_seed}.json",
mime="application/json",
use_container_width=True,
)
with col4:
st.download_button(
"Download B — CSV",
data=_ev_b["df"].to_csv(index=False).encode(),
file_name=f"metrics_B_{cmp_b_dataset}_{cmp_b_arch}_seed{cmp_b_seed}.csv",
mime="text/csv",
use_container_width=True,
)
st.markdown("#### Print / Save as PDF")
_print_as_pdf_button("Print comparison page as PDF")
st.stop()
# ===========================================================================
# MODE 3: UPLOAD YOUR MODEL
# ===========================================================================
st.markdown(
"""
Upload Your Model — Single Sample Evaluation
Run a live robustness sweep on one image using your own MONAI checkpoint.
""",
unsafe_allow_html=True,
)
st.markdown(
"""
⚠️ MONAI models only. This mode supports MONAI-framework
architectures: UNet, Attention UNet, UNETR, SwinUNETR, and VNet.
Upload a .pt / .ckpt file; the architecture
is inferred from the weight tensor shapes. You are responsible for
confirming the correct architecture in the sidebar before running.
Ground-truth masks are optional — if omitted, metrics are computed
relative to the unperturbed baseline prediction.
""",
unsafe_allow_html=True,
)
if not run_btn and "upload_eval_results" not in st.session_state:
st.info("Configure the sidebar and press **Run Evaluation** to start.")
st.stop()
# ── Run evaluation on button press ─────────────────────────────────────────
if run_btn:
with st.spinner("Preparing model…"):
if ckpt_bytes is not None:
try:
model = load_model_from_bytes(
ckpt_bytes, final_arch, final_in_ch, final_out_ch, final_spatial
)
actual_out = model.out_channels
except RuntimeError as e:
st.error(
f"Failed to load checkpoint with architecture '{final_arch}'. "
f"Verify the architecture matches the checkpoint. "
f"Error: {str(e)[:200]}…"
)
st.stop()
else:
model = get_demo_model(final_arch, final_in_ch, final_out_ch, final_spatial)
actual_out = final_out_ch
gt_mask: torch.Tensor | None = None
with st.spinner("Loading image…"):
if npz_file is not None:
_npz_bytes = npz_file.getvalue()
image, _display_np, _npz_lbl = load_npz_sample(_npz_bytes, npz_sample_idx)
vol_affine = np.eye(4)
if _npz_lbl is not None:
gt_mask = torch.from_numpy(_npz_lbl).long()
elif png_file is not None:
image, _display_np = load_png_bytes(
png_file.getvalue(), filename=png_file.name
)
vol_affine = np.eye(4)
if png_label_file is not None:
label_arr = load_png_label_bytes(
png_label_file.getvalue(), filename=png_label_file.name
)
gt_mask = torch.from_numpy(label_arr).long()
elif nii_file is not None:
image, _, vol_affine = load_nifti_bytes(
nii_file.getvalue(), filename=nii_file.name
)
else:
image, _syn_gt = make_synthetic_volume()
gt_mask = _syn_gt
vol_affine = np.eye(4)
st.caption("Using synthetic 64³ ellipsoid (demo).")
if gt_file is not None:
with st.spinner("Loading ground-truth label…"):
gt_mask = load_label_bytes(gt_file.getvalue(), filename=gt_file.name)
img_spatial = image.dim() - 2
if img_spatial != final_spatial:
st.error(
f"Dimension mismatch: image has **{img_spatial}D** but model is "
f"configured for **{final_spatial}D**."
)
st.stop()
img_ch = image.shape[1]
if img_ch != final_in_ch:
st.error(
f"Channel mismatch: image has **{img_ch}** channel(s) but model "
f"expects **{final_in_ch}**."
)
st.stop()
if gt_mask is not None:
expected_sp = tuple(image.shape[2:])
if tuple(gt_mask.shape) != expected_sp:
st.error(
f"Label shape {tuple(gt_mask.shape)} doesn't match image "
f"spatial shape {expected_sp}."
)
st.stop()
if gt_mask is None:
st.info(
"No ground-truth label — segmentation metrics will be computed "
"relative to the **unperturbed baseline prediction**."
)
is_2d = image.dim() == 4
severity_schedule = build_severity_schedule(
n_sharp=int(n_sharp),
n_degrade=int(n_degrade),
max_sharp=float(max_sharp),
max_degrade=float(max_degrade),
)
total_levels = len(severity_schedule)
baseline_idx = severity_schedule.index(0.0)
perturbed_vols: list[tuple[float, np.ndarray]] = []
pred_vols: list[np.ndarray] = []
dim_label = "2-D" if is_2d else "3-D"
_prog = st.progress(0, text=f"Running {dim_label} inference…")
for i, sev in enumerate(severity_schedule):
perturbed = get_perturbed_image(
image, sev, artifact=artifact, modality=modality, seed=int(seed)
)
pred = run_inference(model, perturbed)
perturbed_sq = perturbed.squeeze(0)
if is_2d:
disp = perturbed_sq.mean(0).cpu().numpy()[np.newaxis]
pred_disp = pred.cpu().numpy()[np.newaxis]
else:
disp = perturbed_sq.squeeze(0).cpu().numpy()
pred_disp = pred.cpu().numpy()
perturbed_vols.append((sev, disp))
pred_vols.append(pred_disp)
_prog.progress(
(i + 1) / total_levels,
text=f"Inference {i+1}/{total_levels} (sev={sev:+.2f})",
)
_prog.empty()
if gt_mask is not None:
reference_tensor = gt_mask
reference_label = "vs. GT"
else:
reference_tensor = torch.from_numpy(pred_vols[baseline_idx]).long()
if is_2d:
reference_tensor = reference_tensor.squeeze(0)
reference_label = "vs. baseline pred"
rows: list[dict] = []
_prog2 = st.progress(0, text="Computing metrics…")
for i, sev in enumerate(severity_schedule):
pred_for_metric = torch.from_numpy(pred_vols[i]).long()
if is_2d:
pred_for_metric = pred_for_metric.squeeze(0)
metrics = compute_all_metrics(pred_for_metric, reference_tensor, actual_out)
direction = "sharp" if sev < 0 else ("base" if sev == 0.0 else "degrade")
row: dict = {"severity": round(sev, 4), "direction": direction}
row.update({k: round(v, 5) for k, v in metrics.items()})
rows.append(row)
_prog2.progress((i + 1) / total_levels, text=f"Metrics {i+1}/{total_levels}")
_prog2.empty()
baseline_vol = perturbed_vols[baseline_idx][1]
for i, (sev, pt_np) in enumerate(perturbed_vols):
rows[i]["rmse"] = round(float(np.sqrt(np.mean((pt_np - baseline_vol) ** 2))), 6)
rows[i]["psnr"] = round(psnr(pt_np, baseline_vol), 4)
rows[i]["ssim"] = round(ssim_score(pt_np, baseline_vol), 6)
baseline_row = rows[baseline_idx]
degrade_rows = [r for r in rows if r["direction"] == "degrade"]
def _rood(key: str, higher: bool) -> dict[str, float]:
base_val = baseline_row.get(key, float("nan"))
lvl_vals = [r.get(key, float("nan")) for r in degrade_rows]
return {
f"wm_{key}": wm_metric_t(base_val, lvl_vals),
f"m_ddeg_{key}": m_ddeg_t(base_val, lvl_vals, higher_is_better=higher),
}
rood_metrics: dict[str, float] = {}
rood_metrics.update(_rood("dice", higher=True))
if _SCIPY and "hd95" in baseline_row:
rood_metrics.update(_rood("hd95", higher=False))
else:
rood_metrics["wm_hd95"] = float("nan")
rood_metrics["m_ddeg_hd95"] = float("nan")
rood_out = {
"wm_dsc": rood_metrics.get("wm_dice", float("nan")),
"wm_hd95": rood_metrics.get("wm_hd95", float("nan")),
"m_ddeg": rood_metrics.get("m_ddeg_dice", float("nan")),
"m_ddeg_hd": rood_metrics.get("m_ddeg_hd95", float("nan")),
}
df_up = pd.DataFrame(rows)
gt_np_viewer: np.ndarray | None = None
if gt_mask is not None:
gt_np_viewer = gt_mask.cpu().numpy()
if gt_np_viewer.ndim == 2:
gt_np_viewer = gt_np_viewer[np.newaxis]
st.session_state["upload_eval_results"] = {
"rows": rows,
"df": df_up,
"perturbed_vols": perturbed_vols,
"pred_vols": pred_vols,
"baseline_idx": baseline_idx,
"total_levels": total_levels,
"severity_schedule": severity_schedule,
"actual_out": actual_out,
"gt_mask_np": gt_np_viewer,
"reference_label": reference_label,
"artifact": artifact,
"vol_affine": vol_affine,
"rood": rood_out,
"is_2d": is_2d,
}
st.success(
f"Evaluation complete — {total_levels} levels. Metrics computed {reference_label}."
)
# ── Render upload eval results ──────────────────────────────────────────────
_up_ev = st.session_state["upload_eval_results"]
rows = _up_ev["rows"]
df = _up_ev["df"]
perturbed_vols = _up_ev["perturbed_vols"]
pred_vols = _up_ev["pred_vols"]
baseline_idx = _up_ev["baseline_idx"]
total_levels = _up_ev["total_levels"]
severity_schedule = _up_ev["severity_schedule"]
actual_out = _up_ev["actual_out"]
gt_np = _up_ev["gt_mask_np"]
reference_label = _up_ev["reference_label"]
artifact = _up_ev["artifact"]
vol_affine = _up_ev["vol_affine"]
rood = _up_ev["rood"]
is_2d = _up_ev.get("is_2d", False)
available_metrics: list[str] = (
["dice"] + (["hd95", "hd100", "asd"] if _SCIPY else []) + ["rmse", "psnr", "ssim"]
)
available_metrics = [m for m in available_metrics if m in df.columns]
_D, _H, _W = perturbed_vols[0][1].shape
_vol_label = "2-D" if is_2d else "3-D"
# ── Summary KPIs ─────────────────────────────────────────────────────────────
baseline_row_data = (
df[df["severity"] == 0.0].iloc[0] if 0.0 in df["severity"].values else None
)
c1, c2, c3, c4 = st.columns(4)
with c1:
_dice_val = (
f"{baseline_row_data['dice']:.4f}"
if baseline_row_data is not None
and not np.isnan(baseline_row_data.get("dice", float("nan")))
else "N/A"
)
_dice_lbl = f"Baseline Dice ({reference_label})"
st.metric(_dice_lbl, _dice_val)
with c2:
_wm = rood.get("wm_dsc", float("nan"))
st.metric("wmDSCt (↑ better)", "N/A" if np.isnan(_wm) else f"{_wm:.4f}")
with c3:
_md = rood.get("m_ddeg", float("nan"))
st.metric("mDDegt (↓ better)", "N/A" if np.isnan(_md) else f"{_md:.4f}")
with c4:
_wh = rood.get("wm_hd95", float("nan"))
st.metric("wmHD95t (↓ better)", "N/A" if np.isnan(_wh) else f"{_wh:.4f}")
# ── Tabs ──────────────────────────────────────────────────────────────────────
tab_curve, tab_images, tab_seg, tab_table = st.tabs(
[
"Metrics",
f"Image Quality ({_vol_label})",
f"Segmentation ({_vol_label})",
"Metrics Table",
]
)
# ── Tab: Metrics ─────────────────────────────────────────────────────────────
with tab_curve:
if not _PLOTLY:
st.line_chart(df.set_index("severity")[available_metrics])
else:
active_metrics = st.multiselect(
"Metrics to display",
available_metrics,
default=[m for m in ["dice", "ssim"] if m in available_metrics],
format_func=lambda k: _METRIC_META.get(k, {}).get("label", k),
key="up_metrics_sel",
)
if not active_metrics:
st.warning("Select at least one metric.")
else:
fig_curve = make_subplots(specs=[[{"secondary_y": True}]])
for mk in active_metrics:
meta = _METRIC_META.get(mk, {})
sec = meta.get("axis", "left") == "right"
fig_curve.add_trace(
go.Scatter(
x=df["severity"].tolist(),
y=df[mk].tolist(),
mode="lines+markers",
name=meta.get("label", mk),
line=dict(color=meta.get("colour", "#334155"), width=2.3),
marker=dict(size=6),
),
secondary_y=sec,
)
fig_curve.add_vline(
x=0.0, line_dash="dash", line_color="rgba(100,116,139,0.6)"
)
fig_curve.update_layout(
height=480,
template="plotly_white",
hovermode="x unified",
legend=dict(orientation="h", yanchor="bottom", y=1.02, x=0),
margin=dict(l=20, r=20, t=20, b=20),
)
fig_curve.update_xaxes(title_text="Signed severity (− sharpen / + degrade)")
fig_curve.update_yaxes(title_text="Segmentation / SSIM", secondary_y=False)
fig_curve.update_yaxes(title_text="HD / RMSE / PSNR", secondary_y=True)
st.plotly_chart(fig_curve, use_container_width=True)
# ── Tab: Image Quality ────────────────────────────────────────────────────────
with tab_images:
show_seg_overlay = st.toggle(
"Show segmentation overlay", value=True, key="up_seg_overlay_imgs"
)
if _D == 1:
overview_sev = st.select_slider(
"Severity level",
options=[round(s, 4) for s in severity_schedule],
value=0.0,
key="up_sev_slider_imgs",
)
overview_idx = [round(s, 4) for s in severity_schedule].index(
round(overview_sev, 4)
)
_, vol_disp = perturbed_vols[overview_idx]
pred_disp = pred_vols[overview_idx]
_render_three_planes(
vol_disp,
pred_disp if show_seg_overlay else None,
gt_np,
height=280,
key_prefix=f"up_img_{overview_idx}",
)
else:
overview_sev = st.select_slider(
"Severity level",
options=[round(s, 4) for s in severity_schedule],
value=0.0,
key="up_sev_slider_3d",
)
overview_idx = [round(s, 4) for s in severity_schedule].index(
round(overview_sev, 4)
)
ax_i = st.slider("Axial slice", 0, max(0, _D - 1), _D // 2, key="up_ax")
cor_i = st.slider("Coronal slice", 0, max(0, _H - 1), _H // 2, key="up_cor")
sag_i = st.slider("Sagittal slice", 0, max(0, _W - 1), _W // 2, key="up_sag")
_, vol_disp = perturbed_vols[overview_idx]
pred_disp = pred_vols[overview_idx]
_render_three_planes(
vol_disp,
pred_disp if show_seg_overlay else None,
gt_np,
height=220,
key_prefix=f"up_3d_{overview_idx}",
ax_idx=ax_i,
cor_idx=cor_i,
sag_idx=sag_i,
)
# ── Tab: Segmentation ─────────────────────────────────────────────────────────
with tab_seg:
seg_col_a, seg_col_b = st.columns(2)
with seg_col_a:
sev_a = st.select_slider(
"Severity A",
options=[round(s, 4) for s in severity_schedule],
value=0.0,
key="up_sev_a",
)
with seg_col_b:
sev_b = st.select_slider(
"Severity B",
options=[round(s, 4) for s in severity_schedule],
value=round(severity_schedule[-1], 4),
key="up_sev_b",
)
sev_a_idx = [round(s, 4) for s in severity_schedule].index(round(sev_a, 4))
sev_b_idx = [round(s, 4) for s in severity_schedule].index(round(sev_b, 4))
col_left, col_right = st.columns(2)
with col_left:
st.caption(f"Severity {sev_a:+.2f}")
_, vol_a = perturbed_vols[sev_a_idx]
_render_three_planes(
vol_a,
pred_vols[sev_a_idx],
gt_np,
height=220,
key_prefix=f"up_seg_a_{sev_a_idx}",
)
with col_right:
st.caption(f"Severity {sev_b:+.2f}")
_, vol_b = perturbed_vols[sev_b_idx]
_render_three_planes(
vol_b,
pred_vols[sev_b_idx],
gt_np,
height=220,
key_prefix=f"up_seg_b_{sev_b_idx}",
)
# ── Tab: Metrics Table ────────────────────────────────────────────────────────
with tab_table:
st.dataframe(df, use_container_width=True, hide_index=True)
st.download_button(
"Download metrics CSV",
data=df.to_csv(index=False).encode(),
file_name="upload_eval_metrics.csv",
mime="text/csv",
)
if _NIB and vol_affine is not None:
baseline_pred_np = pred_vols[baseline_idx]
if is_2d:
baseline_pred_np = baseline_pred_np.squeeze(0)
nii_bytes_out = save_nifti_bytes(baseline_pred_np.astype(np.int16), vol_affine)
if nii_bytes_out:
st.download_button(
"Download baseline prediction (NIfTI)",
data=nii_bytes_out,
file_name="baseline_prediction.nii.gz",
mime="application/gzip",
)