app.py

"""
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,
                "<div class='reason'>Upload a fundus image to begin.</div>", "", [], 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"<div class='reason'><b>Primary driver:</b> {s['primary_reason']}"
              + (f"<br><b>Failing axes:</b> {', '.join(s['failing'])}" if s['failing'] else "")
              + (f"<br><b>Borderline:</b> {', '.join(s['borderline'])}" if s['borderline'] else "")
              + f"<br><span class='md-note'>weighted mean {s['weighted_mean']:.0f} · "
                f"weakest-link {s['weakest_link']:.0f}</span></div>")
    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"<div class='reason' style='margin-top:8px'>"
                       f"<b>Deep vasculature (RRWNet):</b> AVR "
                       f"<b>{vs['avr']['avr']:.2f}</b> "
                       f"(arteriolar {vs['avr']['artery_caliber']:.1f}px / venular "
                       f"{vs['avr']['vein_caliber']:.1f}px). "
                       f"<span class='md-note'>Normal ≈ 0.66; lower suggests "
                       f"arteriolar narrowing.</span></div>")
    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"""<div class='reason'>
      <b>Feature space:</b> {fsrc} &nbsp;·&nbsp; <b>Batches:</b> {len(set(batches))}
      &nbsp;·&nbsp; <b>Images:</b> {len(ok)}
      <table><tr><th>&nbsp;</th><th>classifier AUC</th><th>silhouette</th><th>severity</th></tr>
      <tr><td><b>before</b></td><td>{ab}</td><td>{sb}</td><td>{det_b['severity']}</td></tr>
      <tr><td><b>after {method}</b></td><td>{aa}</td><td>{sa}</td><td>{det_a['severity']}</td></tr></table>
      <span class='md-note'>AUC&rarr;0.5 and silhouette&rarr;0 mean batches are no longer
      separable: technical variation harmonised.</span></div>"""
    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 ("<div class='reason'>Run a Batch QC analysis first.</div>", None)
    ok = [r for r in results if "error" not in r]
    if len(ok) < 6:
        return ("<div class='reason'>Need &ge;6 analysed images.</div>", 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 ("<div class='reason'>Assign images to at least 2 batches in the table.</div>", 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 ("<div class='reason'>Upload &ge;6 images across at least 2 batch slots.</div>",
                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"<span class='md-note'>answer set: {out['answer_set']} — FLAIR is contrastive, "
           f"so VQA ranks candidate answers by image–text match</span>")
    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"<div class='reason'><b>Geometric QC composite:</b> {qc:.0f}/100 "
                f"({res['summary']['verdict']}).<br>FLAIR unavailable "
                f"({ENGINE.status}) — load FLAIR for the full disentanglement suite.</div>")
        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"""<div class='reason'>
      <b>Static read-out</b><br>
      Geometric QC composite: <b>{d['qc_composite']:.0f}/100</b> &nbsp;·&nbsp;
      FLAIR quality read: <b>{d['flair_quality']:.0f}/100</b><br>
      FLAIR top finding: <b>{d['flair_disease']}</b> (p={d['flair_disease_prob']:.2f})<br><br>
      <b>Degradation Sensitivity Probe</b> — {dsp['verdict']}<br>
      entanglement index <b>{dsp['entanglement_index']:.2f}</b>,
      robustness <b>{dsp['robustness']:.2f}</b><br><br>
      <b>Spatial disentanglement</b> — confound {occ['confound']:.2f}. {occ['note']}
    </div>"""
    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("<div class='section-title'>Input</div>")
                        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("<div class='section-title'>Composite score</div>")
                        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("<div class='section-title'>Vascular analysis</div>")
                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("<div class='section-title'>Field detection &amp; failure localisation</div>")
                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("<div class='section-title'>Upload a batch of fundus images</div>")
                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("<div class='section-title'>Assign images to batches "
                        "(edit the batch column, then use tab ③)</div>")
                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&harr;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("<div class='section-title'>Quality &harr; pathology disentanglement</div>")
                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("<div class='footer-note'>EyeQC — research & operational QC support tool. "
                "Not a medical device; verdicts assist but do not replace expert grading. "
                "Metrics computed inside the detected retinal field. "
                "FLAIR: Silva-Rodríguez et al., <i>Medical Image Analysis</i> 2024.</div>")
    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…")