app.py

"""
Glaucoma CDSS — HuggingFace Spaces
Input : retinal fundus image (JPEG / PNG)
Output: plain-text clinical report + downloadable PDF
"""

import os
import sys
import tempfile

import cv2
import numpy as np
import gradio as gr

sys.path.insert(0, os.path.dirname(__file__))

from phase3pipeline import Phase3Pipeline


# ── CONFIG ────────────────────────────────────────────────────────────────
REPO_ID = "Nj-1111/EyeeSEE"
EPOCH = None
TOKEN = os.getenv("HF_TOKEN_2") or os.getenv("HF_TOKEN")
# ──────────────────────────────────────────────────────────────────────────

print(f"Loading model repo={REPO_ID} epoch={'latest' if EPOCH is None else EPOCH}")

pipeline = Phase3Pipeline(
    repo_id=REPO_ID,
    epoch=EPOCH,
    mc_passes=20,
    uncertainty_threshold=0.05,
    token=TOKEN,
)

print("Model ready.")


# ── IMAGE LOADER ──────────────────────────────────────────────────────────
def _load_image(path: str) -> np.ndarray:
    img = cv2.imread(path)

    if img is None:
        raw = np.fromfile(path, dtype=np.uint8)
        img = cv2.imdecode(raw, cv2.IMREAD_COLOR)

    if img is None:
        raise ValueError(f"Cannot read image: {path}")

    return img


# ── REPORT BUILDER ────────────────────────────────────────────────────────
def _build_text(r: dict) -> str:
    isnt = r["isnt"]

    lines = [
        "=" * 52,
        "  GLAUCOMA CDSS — CLINICAL REPORT",
        "=" * 52,
        f"  vCDR              : {r['vcdr']:.4f}",
        f"  Risk Level        : {r['risk_level']}",
        f"  Uncertainty       : {r['uncertainty']:.6f}",
        f"  Sanity Check      : {'PASSED' if r['sanity_passed'] else 'CORRECTED (auto)'}",
        f"  ISNT Rule         : {'Satisfied' if isnt['rule_satisfied'] else 'Violated'}",
        "",
        "  ISNT Rim Thickness",
        f"    Inferior  : {isnt['inferior']:.2f}",
        f"    Superior  : {isnt['superior']:.2f}",
        f"    Nasal     : {isnt['nasal']:.2f}",
        f"    Temporal  : {isnt['temporal']:.2f}",
        "",
        "  Structural",
        f"    Disc Area : {r['disc_area_px']:,} px",
        f"    Cup Area  : {r['cup_area_px']:,} px",
        f"    Cup/Disc  : {r['cup_area_px'] / max(r['disc_area_px'], 1) * 100:.1f}%",
        f"    MC Passes : 20",
    ]

    warnings = r.get("warnings", [])
    if warnings:
        lines += ["", "  Warnings"]
        for w in warnings:
            lines.append(f"    ! {w}")

    lines += [
        "",
        "=" * 52,
        "  DISCLAIMER: Research prototype.",
        "  Not a certified medical device.",
        "  Validate with a qualified ophthalmologist.",
        "=" * 52,
    ]

    return "\n".join(lines)


# ── PDF BUILDER ───────────────────────────────────────────────────────────
def _build_pdf(text: str):
    try:
        from fpdf import FPDF

        pdf = FPDF()
        pdf.add_page()
        pdf.set_font("Courier", size=11)

        for line in text.splitlines():
            pdf.cell(0, 7, txt=line, new_x="LMARGIN", new_y="NEXT")

        out = tempfile.NamedTemporaryFile(delete=False, suffix=".pdf")
        pdf.output(out.name)
        return out.name

    except Exception as e:
        print(f"PDF generation failed: {e}")
        return None


# ── INFERENCE ─────────────────────────────────────────────────────────────
def analyse(file_path):
    if file_path is None:
        return "No file uploaded.", None, "Please upload a JPEG or PNG fundus image."

    try:
        print(f"Received file: {file_path}")

        img = _load_image(file_path)
        print(f"Image loaded successfully. Shape: {img.shape}")

        result = pipeline.run(img)
        print("Pipeline inference completed.")

        text = _build_text(result["report"])
        pdf = _build_pdf(text)

        status = "Analysis completed successfully."
        if pdf is None:
            status = "Analysis completed, but PDF generation failed."

        return text, pdf, status

    except Exception as e:
        print(f"Analysis error: {e}")
        return f"Error: {e}", None, f"Analysis failed: {e}"


def set_busy():
    return gr.update(interactive=False), "Running analysis, please wait..."


def set_ready():
    return gr.update(interactive=True)


# ── UI ────────────────────────────────────────────────────────────────────
with gr.Blocks(title="Glaucoma CDSS") as demo:
    gr.Markdown(
        "## Glaucoma CDSS\n"
        "Upload a retinal fundus image to receive a clinical screening report."
    )

    with gr.Row():
        file_in = gr.File(
            label="Fundus Image (JPEG / PNG)",
            file_types=[".jpg", ".jpeg", ".png"],
            type="filepath",
        )

        run_btn = gr.Button("Analyse", variant="primary")

    status_box = gr.Textbox(
        label="Status",
        value="Awaiting image upload.",
        interactive=False,
    )

    report_box = gr.Textbox(
        label="Clinical Report",
        lines=28,
        interactive=False,
    )

    pdf_out = gr.File(label="Download PDF Report")

    gr.Markdown(
        "_Research prototype — NOT a medical device. "
        "All results must be reviewed by a qualified ophthalmologist._"
    )

    run_btn.click(
        fn=set_busy,
        inputs=None,
        outputs=[run_btn, status_box],
        queue=False,
    ).then(
        fn=analyse,
        inputs=[file_in],
        outputs=[report_box, pdf_out, status_box],
    ).then(
        fn=set_ready,
        inputs=None,
        outputs=[run_btn],
        queue=False,
    )

if __name__ == "__main__":
    demo.launch()