"""
EyeQC - a clinician-facing retinal image quality-control workbench and FLAIR
foundation-model bench.
Run: python app.py
Deploy: see README.md (Hugging Face Spaces / Docker).
"""
from __future__ import annotations
try:
import spaces # HF ZeroGPU
GPU = spaces.GPU
except Exception: # not on Spaces -> no-op decorator
def GPU(*a, **k):
if len(a) == 1 and callable(a[0]) and not k:
return a[0]
return lambda f: f
import os
import numpy as np
import pandas as pd
import gradio as gr
from src.pipeline import (analyze_image, analyze_batch, results_to_dataframe,
metric_table, METRIC_NAMES, degradation_map_from_metrics)
from src import visualize as V
from src import interactive_viz as IV
from src import batch_effects as BE
from src import probes as PR
from src.conformal import ConformalGradability, synthetic_calibration
from src.flair_wrapper import ENGINE, DEFAULT_DISEASES
from src.theme import (THEME, CSS, hero_html, verdict_card_html, conformal_chip_html)
RUNS = os.path.join(os.getcwd(), "runs")
os.makedirs(RUNS, exist_ok=True)
# ------------------------------------------------------ deep A/V segmenter (RRWNet)
RRWNET = {"active": False, "status": "not attempted"}
def try_register_rrwnet():
"""Auto-wire RRWNet if weights + torch are present (weights/ or RRWNET_WEIGHTS)."""
try:
from src.rrwnet_seg import register, _find_weights
if _find_weights() is None:
RRWNET["status"] = ("no weights found — put rrwnet_RITE_1.pth in ./weights "
"or set RRWNET_WEIGHTS")
return
ok, status = register()
RRWNET["active"] = ok; RRWNET["status"] = status
except Exception as e:
RRWNET["status"] = f"unavailable: {e}"
# ------------------------------------------------------ lazy conformal calibrator
_CALIB = {"model": None}
def _heavy_degrade(rgb):
import cv2
return cv2.GaussianBlur(rgb, (0, 0), 7)
def get_calibrator():
if _CALIB["model"] is None:
sdir = "assets/samples"
refs = []
if os.path.isdir(sdir):
for f in sorted(os.listdir(sdir))[:6]:
try:
from PIL import Image
arr = np.array(Image.open(os.path.join(sdir, f)).convert("RGB"))
refs.append(analyze_image(arr, f, with_vessels=False))
except Exception:
pass
try:
_CALIB["model"] = synthetic_calibration(refs, _heavy_degrade, alpha=0.1)
except Exception:
_CALIB["model"] = ConformalGradability() # unfitted -> uncalibrated chip
return _CALIB["model"]
def qc_composite_of(rgb):
from src.fov import detect_fov
from src.qc_metrics import compute_all_metrics
from src.quality_score import composite_score
f = detect_fov(rgb)
return composite_score(compute_all_metrics(rgb, f))["composite"]
# ============================================================ SINGLE IMAGE
@GPU(duration=60)
def run_single(img):
if img is None:
return (None, None, None, None, None, None, None, None,
"Upload a fundus image to begin.
", "", [], None)
res = analyze_image(img)
s = res["summary"]
card = verdict_card_html(s)
pred = get_calibrator().predict(s["composite"] / 100.0)
chip = conformal_chip_html(pred)
reason = (f"Primary driver: {s['primary_reason']}"
+ (f"
Failing axes: {', '.join(s['failing'])}" if s['failing'] else "")
+ (f"
Borderline: {', '.join(s['borderline'])}" if s['borderline'] else "")
+ f"
weighted mean {s['weighted_mean']:.0f} · "
f"weakest-link {s['weakest_link']:.0f}
")
vess_overlay = vess_stats = None
if "vessels" in res:
vs = res["vessels"]
vess_overlay = (V.av_overlay(res["rgb"], vs) if "artery" in vs
else V.vessel_overlay(res["rgb"], vs))
vess_stats = V.vessel_stats_panel(vs)
if "avr" in vs and vs["avr"]["avr"] == vs["avr"]["avr"]: # not NaN
reason += (f""
f"Deep vasculature (RRWNet): AVR "
f"{vs['avr']['avr']:.2f} "
f"(arteriolar {vs['avr']['artery_caliber']:.1f}px / venular "
f"{vs['avr']['vein_caliber']:.1f}px). "
f"Normal ≈ 0.66; lower suggests "
f"arteriolar narrowing.
")
return (V.score_gauge(s), V.metric_radar(res["metrics"]), V.metric_bars(res["metrics"]),
res["fov_overlay"], res["problem_overlay"], vess_overlay, vess_stats,
card, reason, chip, metric_table(res), res)
def show_axis_heatmap(res, axis_name):
if res is None:
return None
from src.failure_analysis import per_axis_heatmap
metric = next((m for m in res["metrics"] if m["name"] == axis_name), None)
if metric is None or metric.get("_map") is None:
return res["rgb"]
return per_axis_heatmap(res["rgb"], res["fov"], metric)
# ============================================================ BATCH QC
@GPU(duration=120)
def run_batch(files, progress=gr.Progress()):
if not files:
return (None, None, None, None, pd.DataFrame(), None, None, None)
paths = [f.name if hasattr(f, "name") else f for f in files]
results = analyze_batch(paths, progress=progress)
df = results_to_dataframe(results)
ok = [r for r in results if "error" not in r]
clean = df[df["verdict"] != "ERROR"]
dist = V.cohort_distribution(clean) if len(ok) else None
heat = V.axis_heatmap(clean, METRIC_NAMES) if len(ok) >= 2 else None
thumbs = [r["rgb"] for r in ok]
labels = [f'{r["name"][:16]} {r["summary"]["composite"]:.0f}' for r in ok]
verdicts = [r["summary"]["verdict"] for r in ok]
panel = V.quality_panel(thumbs, labels, verdicts) if ok else None
csv_path = os.path.join(RUNS, "eyeqc_report.csv")
df.to_csv(csv_path, index=False)
# editable batch-assignment table
assign = pd.DataFrame({"image": [r["name"] for r in ok],
"batch": ["batch1"] * len(ok)})
return dist, heat, panel, df, assign, results, csv_path, results
# ============================================================ BATCH EFFECTS
def _features(results, source):
ok = [r for r in results if "error" not in r]
if source == "FLAIR embedding" and ENGINE.load():
X = np.array([ENGINE.embedding(r["rgb"]) for r in ok])
return X, ok, "FLAIR image embedding (512-d)"
X = np.array([r["descriptor"] for r in ok])
src = "interpretable QC + colour descriptor"
if source == "FLAIR embedding":
src += " (FLAIR unavailable - fell back)"
return X, ok, src
def _run_be(X, ok, batches, method, fsrc):
det_b = BE.detect_batch_effect(X, batches)
Xc = BE.combat(X, batches) if method == "ComBat" else BE.zstandardise_by_batch(X, batches)
det_a = BE.detect_batch_effect(Xc, batches)
emb_b, nm = BE.embed_2d(X, method="pca")
emb_a, _ = BE.embed_2d(Xc, method="pca")
fig = IV.animated_correction(emb_b, emb_a, batches)
def fmt(d):
a = "n/a" if np.isnan(d["auc"]) else f'{d["auc"]:.2f}'
s = "n/a" if np.isnan(d["silhouette"]) else f'{d["silhouette"]:.2f}'
return a, s
ab, sb = fmt(det_b); aa, sa = fmt(det_a)
html = f"""
Feature space: {fsrc} ·
Batches: {len(set(batches))}
·
Images: {len(ok)}
| | classifier AUC | silhouette | severity |
|---|
| before | {ab} | {sb} | {det_b['severity']} |
| after {method} | {aa} | {sa} | {det_a['severity']} |
AUC→0.5 and silhouette→0 mean batches are no longer
separable: technical variation harmonised. """
return html, IV.fig_to_iframe(fig)
@GPU(duration=120)
def run_be_from_table(results, assign_df, feature_source, method):
if not results:
return ("Run a Batch QC analysis first.
", None)
ok = [r for r in results if "error" not in r]
if len(ok) < 6:
return ("Need ≥6 analysed images.
", None)
# map image -> batch from the edited table
if isinstance(assign_df, pd.DataFrame):
amap = dict(zip(assign_df["image"], assign_df["batch"]))
else:
amap = {row[0]: row[1] for row in assign_df}
batches = [str(amap.get(r["name"], "batch1")) for r in ok]
if len(set(batches)) < 2:
return ("Assign images to at least 2 batches in the table.
", None)
X, ok, fsrc = _features(results, feature_source)
return _run_be(X, ok, batches, method, fsrc)
@GPU(duration=120)
def run_be_from_uploads(fa, fb, fc, fd, feature_source, method, progress=gr.Progress()):
slots = [("A", fa), ("B", fb), ("C", fc), ("D", fd)]
all_res, batches = [], []
for letter, files in slots:
if not files:
continue
paths = [f.name if hasattr(f, "name") else f for f in files]
rs = analyze_batch(paths, progress=progress)
for r in rs:
if "error" not in r:
all_res.append(r); batches.append(f"batch{letter}")
if len(set(batches)) < 2 or len(all_res) < 6:
return ("Upload ≥6 images across at least 2 batch slots.
",
None, all_res)
X, ok, fsrc = _features(all_res, feature_source)
html, fig = _run_be(X, ok, batches, method, fsrc)
return html, fig, all_res
# ============================================================ FLAIR BENCH
def flair_load():
if ENGINE.load():
return "✅ FLAIR loaded and ready."
return (f"⚠️ FLAIR unavailable on this host. Status: {ENGINE.status}\n\n"
"The QC pipeline works fully without FLAIR. Enable it by installing "
"torch + `git+https://github.com/jusiro/FLAIR.git` (see README).")
@GPU(duration=60)
def flair_disease(img, extra):
if img is None:
return "Upload a fundus image.", None
if not ENGINE.load():
return f"FLAIR unavailable: {ENGINE.status}", None
diseases = list(DEFAULT_DISEASES) + [d.strip() for d in (extra or "").split(",") if d.strip()]
dz = ENGINE.zero_shot_disease(img, diseases)
df = pd.DataFrame([{"finding": d["label"], "probability": round(d["prob"], 4),
"logit": round(d["logit"], 2)} for d in dz])
return "Zero-shot findings (ranked):", df
@GPU(duration=60)
def flair_vqa(img, question, candidates):
if img is None or not (question or "").strip():
return "Upload an image and ask a question.", None
if not ENGINE.load():
return f"FLAIR unavailable: {ENGINE.status}", None
out = ENGINE.vqa_answer(img, question, candidates)
msg = (f"**Answer:** {out['answer']} (confidence {out['confidence']:.2f}) \n"
f"answer set: {out['answer_set']} — FLAIR is contrastive, "
f"so VQA ranks candidate answers by image–text match")
df = pd.DataFrame([{"candidate answer": r["answer"], "prob": round(r["prob"], 3),
"logit": round(r["logit"], 2)} for r in out["ranked"]])
return msg, df
@GPU(duration=120)
def flair_disentangle(img, progress=gr.Progress()):
"""Full disentanglement: QC vs FLAIR-quality, DSP curves, entanglement, occlusion."""
if img is None:
return "Upload a fundus image.", None, None, None, None
res = analyze_image(img)
qc = res["summary"]["composite"]
if not ENGINE.load():
html = (f"Geometric QC composite: {qc:.0f}/100 "
f"({res['summary']['verdict']}).
FLAIR unavailable "
f"({ENGINE.status}) — load FLAIR for the full disentanglement suite.
")
return html, None, None, None, None
progress(0.2, desc="FLAIR quality vs disease")
d = ENGINE.quality_disentanglement(img, qc)
progress(0.4, desc="degradation sensitivity probe")
dsp = PR.degradation_sensitivity(ENGINE, res["rgb"], res["fov"], qc_composite_of,
levels=6)
progress(0.8, desc="occlusion spatial disentanglement")
occ = PR.occlusion_disentanglement(ENGINE, res["rgb"], res["fov"],
res["degradation_map"], grid=7)
html = f"""
Static read-out
Geometric QC composite: {d['qc_composite']:.0f}/100 ·
FLAIR quality read: {d['flair_quality']:.0f}/100
FLAIR top finding: {d['flair_disease']} (p={d['flair_disease_prob']:.2f})
Degradation Sensitivity Probe — {dsp['verdict']}
entanglement index {dsp['entanglement_index']:.2f},
robustness {dsp['robustness']:.2f}
Spatial disentanglement — confound {occ['confound']:.2f}. {occ['note']}
"""
dsp_fig = IV.dsp_figure(dsp)
dial = IV.entanglement_dial(dsp["entanglement_index"])
# occlusion overlay
import cv2
sal = (occ["saliency"] * 255).astype(np.uint8)
heat = cv2.applyColorMap(sal, cv2.COLORMAP_INFERNO)[..., ::-1]
over = np.clip(res["rgb"] * 0.55 + heat * 0.45, 0, 255).astype(np.uint8)
return html, dsp_fig, dial, over, res["degradation_map"]
# ============================================================ UI
def build():
with gr.Blocks(title="EyeQC") as demo:
gr.HTML(hero_html())
res_state = gr.State()
batch_state = gr.State()
with gr.Tabs():
# ---------------------------------------------------- single
with gr.Tab("① Single-image QC"):
with gr.Row():
with gr.Column(scale=5):
gr.HTML("Input
")
img_in = gr.Image(type="numpy", label="Fundus photograph", height=330)
run_btn = gr.Button("Analyse quality", variant="primary")
if os.path.isdir("assets/samples"):
gr.Examples([["assets/samples/" + f] for f in
sorted(os.listdir("assets/samples"))],
inputs=img_in, label="Example images")
verdict_html = gr.HTML()
conf_html = gr.HTML()
reason_html = gr.HTML()
with gr.Column(scale=4):
gr.HTML("Composite score
")
gauge_out = gr.Image(show_label=False, height=300)
with gr.Row():
radar_out = gr.Image(label="Quality profile", height=430)
bars_out = gr.Image(label="Axes ranked (worst first)", height=430)
gr.HTML("Vascular analysis
")
gr.Markdown(
("🟢 **RRWNet deep artery/vein segmentation active** — "
+ RRWNET["status"]) if RRWNET["active"] else
("⚪ Classical vessel backend. Deep A/V (RRWNet): "
+ RRWNET["status"]),
elem_classes="md-note")
with gr.Row():
vessel_over = gr.Image(label="Vessel map (teal=vesselness, amber=skeleton)", height=340)
vessel_stat = gr.Image(label="Structural descriptors", height=340)
gr.HTML("Field detection & failure localisation
")
with gr.Row():
fov_out = gr.Image(label="Detected retinal field", height=330)
problem_out = gr.Image(label="Composite problem map", height=330)
with gr.Row():
axis_pick = gr.Dropdown(METRIC_NAMES, value="Vessel Visibility",
label="Inspect one axis")
axis_heat = gr.Image(label="Per-axis heatmap", height=330)
table_out = gr.Dataframe(
headers=["Axis", "Measurement", "Score", "Status", "Clinical note"],
datatype=["str"] * 5, wrap=True, row_count=(10, "fixed"))
run_btn.click(run_single, [img_in],
[gauge_out, radar_out, bars_out, fov_out, problem_out,
vessel_over, vessel_stat, verdict_html, reason_html,
conf_html, table_out, res_state])
axis_pick.change(show_axis_heatmap, [res_state, axis_pick], [axis_heat])
# ---------------------------------------------------- batch
with gr.Tab("② Batch QC & cohort"):
gr.HTML("Upload a batch of fundus images
")
files_in = gr.File(file_count="multiple", file_types=["image"],
label="Drop many images")
batch_btn = gr.Button("Analyse batch", variant="primary")
dist_out = gr.Image(label="Cohort quality distribution & verdicts", height=330)
panel_out = gr.Image(label="Quality panel", height=520)
heat_out = gr.Image(label="Per-axis scores across cohort", height=380)
table_batch = gr.Dataframe(label="Per-image QC report", wrap=True)
gr.HTML("Assign images to batches "
"(edit the batch column, then use tab ③)
")
assign_tbl = gr.Dataframe(headers=["image", "batch"],
datatype=["str", "str"], interactive=True,
label="Batch assignment (editable)")
csv_out = gr.File(label="Download CSV report")
batch_btn.click(run_batch, [files_in],
[dist_out, heat_out, panel_out, table_batch,
assign_tbl, batch_state, csv_out, batch_state])
# ---------------------------------------------------- batch effects
with gr.Tab("③ Batch effects"):
gr.Markdown("Detect and **correct** systematic technical variation across "
"cameras / sites / days.", elem_classes="md-note")
with gr.Tabs():
with gr.Tab("Use cohort + assignment table"):
with gr.Row():
feat1 = gr.Radio(["QC descriptor", "FLAIR embedding"],
value="QC descriptor", label="Feature space")
meth1 = gr.Radio(["ComBat", "z-standardise"], value="ComBat",
label="Correction")
be1_btn = gr.Button("Detect & correct batch effects", variant="primary")
be1_html = gr.HTML()
be1_plot = gr.HTML(label="Batch harmonisation (animated)")
be1_btn.click(run_be_from_table,
[batch_state, assign_tbl, feat1, meth1],
[be1_html, be1_plot])
with gr.Tab("Upload batch-wise"):
gr.Markdown("Upload each batch into its own slot.",
elem_classes="md-note")
with gr.Row():
fa = gr.File(file_count="multiple", file_types=["image"], label="Batch A")
fb = gr.File(file_count="multiple", file_types=["image"], label="Batch B")
with gr.Row():
fc = gr.File(file_count="multiple", file_types=["image"], label="Batch C")
fd = gr.File(file_count="multiple", file_types=["image"], label="Batch D")
with gr.Row():
feat2 = gr.Radio(["QC descriptor", "FLAIR embedding"],
value="QC descriptor", label="Feature space")
meth2 = gr.Radio(["ComBat", "z-standardise"], value="ComBat",
label="Correction")
be2_btn = gr.Button("Analyse & correct batch effects", variant="primary")
be2_html = gr.HTML()
be2_plot = gr.HTML(label="Batch harmonisation (animated)")
be2_btn.click(run_be_from_uploads,
[fa, fb, fc, fd, feat2, meth2],
[be2_html, be2_plot, batch_state])
# ---------------------------------------------------- FLAIR
with gr.Tab("④ FLAIR foundation-model bench"):
gr.Markdown("FLAIR (ResNet-50 + Bio-ClinicalBERT) for zero-shot disease "
"read-out, contrastive **VQA**, and quality↔pathology "
"**disentanglement** probes.", elem_classes="md-note")
flair_status = gr.Markdown()
gr.Button("Load FLAIR", variant="primary").click(flair_load, None, [flair_status])
with gr.Row():
flair_img = gr.Image(type="numpy", label="Fundus photograph", height=340)
with gr.Column():
with gr.Tab("Zero-shot disease"):
extra_dz = gr.Textbox(label="Extra findings (comma-separated)")
dz_btn = gr.Button("Run zero-shot")
dz_msg = gr.Markdown(); dz_tbl = gr.Dataframe()
dz_btn.click(flair_disease, [flair_img, extra_dz], [dz_msg, dz_tbl])
with gr.Tab("VQA"):
q_in = gr.Textbox(label="Question",
placeholder="e.g. what disease is shown? / is this gradable? / which eye?")
cand_in = gr.Textbox(label="Candidate answers (optional, comma-separated)",
placeholder="leave blank to auto-pick an answer set")
vqa_btn = gr.Button("Answer")
vqa_msg = gr.Markdown(); vqa_tbl = gr.Dataframe()
vqa_btn.click(flair_vqa, [flair_img, q_in, cand_in], [vqa_msg, vqa_tbl])
gr.HTML("Quality ↔ pathology disentanglement
")
dis_btn = gr.Button("Run disentanglement suite", variant="primary")
dis_html = gr.HTML()
with gr.Row():
dsp_plot = gr.Plot(label="Degradation sensitivity")
dial_plot = gr.Plot(label="Entanglement index")
with gr.Row():
occ_out = gr.Image(label="Disease saliency (occlusion)", height=340)
dmap_out = gr.Image(label="QC degradation map", height=340)
dis_btn.click(flair_disentangle, [flair_img],
[dis_html, dsp_plot, dial_plot, occ_out, dmap_out])
# ---------------------------------------------------- methods
with gr.Tab("ⓘ Methods"):
gr.Markdown(METHODS_MD)
gr.HTML("")
return demo
METHODS_MD = """
### EyeQC methodology
**Field-aware QC.** Every metric is measured inside an automatically detected
retinal disc, eroded to exclude the black border, field-edge ring and dark
corners. Ten axes (vessel visibility, focus, sharpness, illumination uniformity,
exposure, contrast, field definition, artifact burden, clipping, colour balance)
each return a physical value, a 0–1 score and a PASS/ACCEPTABLE/FAIL status.
**Composite = 65% weighted mean + 35% weakest-link**, so one fatal flaw cannot be
masked by strong performance elsewhere.
**Robust fundus-circle detection.** The disc centre is *not* assumed to be the
image centre. EyeQC segments the foreground, then fits the fundus circle by
algebraic least squares to the true boundary arc — excluding points on the image
frame, which are truncation edges — with an extent-based radius estimate that is
robust even when the circle is heavily cropped. Off-centre, letter-boxed and
partially-cropped fundus images are localised correctly before any metric is
computed. The same ROI mask feeds the deep segmenter's preprocessing.
**Rigorous vascular analysis.** A multi-scale vesselness map yields structural
descriptors (density, skeleton length, fractal dimension, fragmentation). The
gradability score is anchored to a blur-monotonic top-hat vessel-contrast measure.
When **RRWNet** deep artery/vein segmentation is enabled (weights in `./weights`),
EyeQC reports true vessel/artery/vein maps and the **arteriolar-to-venular ratio
(AVR)** — a validated cardiovascular biomarker.
**Conformal gradability.** Split-conformal prediction turns the quality score into
a calibrated set — {gradable}, {ungradable}, or {uncertain} — with a finite-sample
coverage guarantee, given a labelled (or surrogate) calibration set.
**Batch-effect harmonisation.** Per-image features (interpretable QC descriptors or
FLAIR embeddings) are tested for batch separability (cross-validated classifier AUC
+ silhouette) and corrected with ComBat or per-batch z-standardisation; an animated
embedding shows the harmonisation.
### Novel foundation-model disentanglement
**Degradation Sensitivity Probe (DSP).** Controlled degradations (defocus,
illumination, contrast) are swept at increasing severity while FLAIR's confidence
in the originally-predicted disease is tracked. The **entanglement index** is the
positive correlation between disease confidence and image quality across the sweep:
high means the disease call co-moves with quality — evidence the foundation model
is conflating degradation with pathology.
**Occlusion spatial disentanglement.** Occlusion saliency localises the pixels
driving FLAIR's disease call; the **confound score** is the fraction of that
evidence sitting on regions the QC pipeline flags as degraded.
Together these give a per-image, quantitative answer to *"is this disease read
real, or a quality artefact?"* — the question at the centre of trustworthy retinal
foundation models.
"""
def _resolve_port():
os.environ.pop("GRADIO_SERVER_PORT", None)
try:
return int(os.environ.get("PORT", 7860))
except ValueError:
return 7860
def try_load_flair():
try:
ENGINE.load()
except Exception as e:
print(f"[eyeqc] FLAIR eager-load skipped: {e}")
if __name__ == "__main__":
try_register_rrwnet()
try_load_flair()
print(f"[eyeqc] RRWNet: {RRWNET['status']}")
demo = build()
share = os.environ.get("SHARE", "0") == "1" and not os.environ.get("SPACE_ID")
app = demo.queue(max_size=24)
want = _resolve_port()
for port in [want, None]:
try:
app.launch(server_name="0.0.0.0", server_port=port, share=True,
theme=THEME, css=CSS, show_error=True, allowed_paths=[RUNS])
break
except OSError as e:
if port is None:
raise
print(f"[eyeqc] port {want} busy ({e}); scanning for a free port…")