# ============================================================ # AI-Assisted Dry Eye Disease Multimodal Prototype # Hugging Face Spaces / Gradio app # Modules: # 1. Clinical questionnaire model / fallback score # 2. MGD-1K mask-based meibography model # 3. Synthetic-cohort lipidomics model # ============================================================ import os import json import joblib import cv2 import numpy as np import pandas as pd import gradio as gr MODEL_DIR = "models" CLINICAL_MODEL_PATH = os.path.join(MODEL_DIR, "clinical_xgboost_smote_pipeline.pkl") CLINICAL_FEATURES_PATH = os.path.join(MODEL_DIR, "clinical_all_features.pkl") CLINICAL_THRESHOLD_PATH = os.path.join(MODEL_DIR, "clinical_best_threshold.pkl") MEIBO_MODEL_PATH = os.path.join(MODEL_DIR, "mgd1k_binary_mask_model.pkl") MEIBO_FEATURES_PATH = os.path.join(MODEL_DIR, "mgd1k_binary_mask_features.pkl") LIPID_MODEL_PATH = os.path.join(MODEL_DIR, "lipidomics_synthetic_xgboost_model.pkl") LIPID_FEATURES_PATH = os.path.join(MODEL_DIR, "lipidomics_top20_features.pkl") def safe_load(path, default=None): try: if os.path.exists(path): return joblib.load(path) except Exception as e: print(f"Could not load {path}: {e}") return default clinical_model = safe_load(CLINICAL_MODEL_PATH, None) clinical_features = safe_load(CLINICAL_FEATURES_PATH, []) clinical_threshold = safe_load(CLINICAL_THRESHOLD_PATH, 0.5) try: clinical_threshold = float(clinical_threshold) except Exception: clinical_threshold = 0.5 meibo_model = safe_load(MEIBO_MODEL_PATH, None) meibo_features = safe_load(MEIBO_FEATURES_PATH, []) lipid_model = safe_load(LIPID_MODEL_PATH, None) lipid_features = safe_load(LIPID_FEATURES_PATH, []) # ------------------------------------------------------------ # Meibography module: gland mask + eyelid mask -> features # ------------------------------------------------------------ def load_binary_mask_from_file(file_path): img = cv2.imread(file_path, cv2.IMREAD_GRAYSCALE) if img is None: raise ValueError("Could not read image. Please upload a valid mask image.") img = cv2.resize(img, (512, 512)) _, binary = cv2.threshold(img, 50, 255, cv2.THRESH_BINARY) return binary def extract_meibography_features(gland_mask_path, eyelid_mask_path): gland = load_binary_mask_from_file(gland_mask_path) eyelid = load_binary_mask_from_file(eyelid_mask_path) gland_pixels = int(np.sum(gland > 0)) eyelid_pixels = int(np.sum(eyelid > 0)) total_pixels = int(gland.shape[0] * gland.shape[1]) if eyelid_pixels == 0: eyelid_pixels = total_pixels gland_ratio_to_eyelid = gland_pixels / eyelid_pixels gland_ratio_to_total = gland_pixels / total_pixels dropout_ratio = 1 - gland_ratio_to_eyelid num_labels, labels, stats, centroids = cv2.connectedComponentsWithStats(gland, connectivity=8) component_areas = [] for i in range(1, num_labels): area = stats[i, cv2.CC_STAT_AREA] if area > 20: component_areas.append(area) gland_count = len(component_areas) if gland_count > 0: avg_area = float(np.mean(component_areas)) max_area = float(np.max(component_areas)) min_area = float(np.min(component_areas)) std_area = float(np.std(component_areas)) else: avg_area = max_area = min_area = std_area = 0.0 return { "gland_pixels": gland_pixels, "eyelid_pixels": eyelid_pixels, "gland_ratio_to_eyelid": gland_ratio_to_eyelid, "gland_ratio_to_total": gland_ratio_to_total, "dropout_ratio": dropout_ratio, "gland_count": gland_count, "avg_component_area": avg_area, "max_component_area": max_area, "min_component_area": min_area, "std_component_area": std_area, } def predict_meibography(gland_mask, eyelid_mask): if meibo_model is None: return None, "Meibography model is missing. Add mgd1k_binary_mask_model.pkl to models/." if gland_mask is None or eyelid_mask is None: return None, "Upload both a gland mask and an eyelid mask." feats = extract_meibography_features(gland_mask, eyelid_mask) X = pd.DataFrame([feats]) if meibo_features: X = X[meibo_features] prob = float(meibo_model.predict_proba(X)[0][1]) uncertainty = float(1 - abs(prob - 0.5) * 2) label = "Moderate/Severe MGD structural risk" if prob >= 0.5 else "Low/Mild MGD structural risk" text = ( f"Prediction: {label}\n" f"High MGD structural probability: {prob*100:.2f}%\n" f"Uncertainty: {uncertainty*100:.2f}%\n" f"Extracted morphology: gland/eyelid ratio={feats['gland_ratio_to_eyelid']:.3f}, " f"dropout ratio={feats['dropout_ratio']:.3f}, gland count={feats['gland_count']}" ) return {"prob": prob, "uncertainty": uncertainty, "label": label, "features": feats}, text # ------------------------------------------------------------ # Clinical module # ------------------------------------------------------------ def clinical_fallback_score(patient): score = 0.0 if patient.get("Redness_in_eye") == "Y": score += 2.0 if patient.get("Discomfort_Eye_strain") == "Y": score += 2.0 if patient.get("Itchiness_Irritation_in_eye") == "Y": score += 2.0 if float(patient.get("Average_screen_time", 0)) >= 6: score += 1.5 if float(patient.get("Sleep_quality", 3)) <= 2: score += 1.0 if float(patient.get("Age", 30)) >= 45: score += 1.5 prob = min(max(score / 10.0, 0), 1) return prob, 0.25, "Fallback rule-based clinical score" def predict_clinical(age, avg_screen_time, sleep_quality, redness, eye_strain, itchiness, gender, sleep_duration, stress_level, heart_rate, daily_steps, physical_activity, height, weight, sleep_disorder, wake_night, sleepy_day, caffeine, alcohol, smoking, medical_issue, medication, device_before_bed, blue_filter, systolic_bp, diastolic_bp): patient = { "Gender": gender, "Age": age, "Sleep_duration": sleep_duration, "Sleep_quality": sleep_quality, "Stress_level": stress_level, "Heart_rate": heart_rate, "Daily_steps": daily_steps, "Physical_activity": physical_activity, "Height": height, "Weight": weight, "Sleep_disorder": sleep_disorder, "Wake_up_during_night": wake_night, "Feel_sleepy_during_day": sleepy_day, "Caffeine_consumption": caffeine, "Alcohol_consumption": alcohol, "Smoking": smoking, "Medical_issue": medical_issue, "Ongoing_medication": medication, "Smart_device_before_bed": device_before_bed, "Average_screen_time": avg_screen_time, "Blue_light_filter": blue_filter, "Discomfort_Eye_strain": eye_strain, "Redness_in_eye": redness, "Itchiness_Irritation_in_eye": itchiness, "Systolic_BP": systolic_bp, "Diastolic_BP": diastolic_bp, } if clinical_model is not None: try: X = pd.DataFrame([patient]) if clinical_features: for col in clinical_features: if col not in X.columns: X[col] = np.nan X = X[clinical_features] prob = float(clinical_model.predict_proba(X)[0][1]) uncertainty = float(1 - abs(prob - clinical_threshold) * 2) uncertainty = max(0.0, min(1.0, uncertainty)) method = "Trained clinical model" except Exception as e: prob, uncertainty, method = clinical_fallback_score(patient) method += f" (trained clinical model could not run: {e})" else: prob, uncertainty, method = clinical_fallback_score(patient) label = "Clinical DED risk pattern" if prob >= clinical_threshold else "Low clinical DED risk pattern" text = ( f"Prediction: {label}\n" f"Clinical DED probability: {prob*100:.2f}%\n" f"Uncertainty: {uncertainty*100:.2f}%\n" f"Method: {method}" ) return {"prob": prob, "uncertainty": uncertainty, "label": label}, text # ------------------------------------------------------------ # Lipidomics module # ------------------------------------------------------------ def read_table(file_path): if file_path is None: return None lower = file_path.lower() if lower.endswith(".csv"): return pd.read_csv(file_path) if lower.endswith(".tsv") or lower.endswith(".txt"): return pd.read_csv(file_path, sep="\t") if lower.endswith(".xlsx") or lower.endswith(".xls"): return pd.read_excel(file_path) return pd.read_csv(file_path) def predict_lipidomics(lipid_file): if lipid_model is None: return None, "Lipidomics model missing. Add lipidomics_synthetic_xgboost_model.pkl to models/." if not lipid_features: return None, "Lipidomics top-20 feature file missing." if lipid_file is None: return None, "Upload a CSV/TSV/XLSX file with one row containing the top-20 lipid biomarker columns." df = read_table(lipid_file) if df is None or df.empty: return None, "Could not read lipidomics file." for col in lipid_features: if col not in df.columns: df[col] = np.nan X = df[lipid_features].copy() X = X.apply(pd.to_numeric, errors="coerce") X = X.fillna(X.median(numeric_only=True)).fillna(0) sample = X.iloc[[0]] prob = float(lipid_model.predict_proba(sample)[0][1]) uncertainty = float(1 - abs(prob - 0.5) * 2) label = "DED-like lipidomic molecular pattern" if prob >= 0.5 else "Normal-like lipidomic molecular pattern" # find strongest available biomarkers in the uploaded row by absolute value top_vals = sample.iloc[0].abs().sort_values(ascending=False).head(5) biomarkers = ", ".join([str(x) for x in top_vals.index]) text = ( f"Prediction: {label}\n" f"Lipidomics DED probability: {prob*100:.2f}%\n" f"Uncertainty: {uncertainty*100:.2f}%\n" f"Top entered lipid values by magnitude: {biomarkers}\n" f"Important limitation: this lipidomics model is synthetic-cohort based and not clinically validated." ) return {"prob": prob, "uncertainty": uncertainty, "label": label}, text # ------------------------------------------------------------ # Ensemble # ------------------------------------------------------------ def phenotype_report(clinical_prob, meibo_prob, lipid_prob): if meibo_prob >= 0.65 and lipid_prob >= 0.65: return "Mixed MGD-dominant + lipidomic molecular DED pattern" if meibo_prob >= 0.65: return "MGD / evaporative structural DED pattern" if lipid_prob >= 0.65: return "Lipidomic molecular dysregulation DED pattern" if clinical_prob >= 0.65: return "Symptom/lifestyle-supported DED risk pattern" return "Low-risk or unclear DED pattern" def make_recommendation(final_score, clinical_prob, meibo_prob, lipid_prob): lines = ["Research-support interpretation only; not a diagnostic medical device."] if final_score >= 0.75: lines.append("Overall risk is high. Confirm with ophthalmic examination and standard clinical tests.") elif final_score >= 0.50: lines.append("Overall risk is moderate. Recommend clinical confirmation and follow-up.") else: lines.append("Overall risk is low-to-moderate based on supplied inputs.") if meibo_prob >= 0.65: lines.append("Structural signal suggests MGD involvement; examine gland loss, expressibility, and tear-film evaporation.") if lipid_prob >= 0.65: lines.append("Molecular signal suggests lipid remodeling; useful for future biosensor/biomarker validation.") if clinical_prob >= 0.65: lines.append("Clinical signal suggests symptomatic risk; review redness, eye strain, itching, screen exposure, and sleep quality.") return "\n".join(lines) def run_system(gland_mask, eyelid_mask, lipid_file, age, avg_screen_time, sleep_quality, redness, eye_strain, itchiness, gender, sleep_duration, stress_level, heart_rate, daily_steps, physical_activity, height, weight, sleep_disorder, wake_night, sleepy_day, caffeine, alcohol, smoking, medical_issue, medication, device_before_bed, blue_filter, systolic_bp, diastolic_bp, clinical_weight, meibo_weight, lipid_weight): clinical_res, clinical_text = predict_clinical( age, avg_screen_time, sleep_quality, redness, eye_strain, itchiness, gender, sleep_duration, stress_level, heart_rate, daily_steps, physical_activity, height, weight, sleep_disorder, wake_night, sleepy_day, caffeine, alcohol, smoking, medical_issue, medication, device_before_bed, blue_filter, systolic_bp, diastolic_bp ) meibo_res, meibo_text = predict_meibography(gland_mask, eyelid_mask) lipid_res, lipid_text = predict_lipidomics(lipid_file) probs = [] weights = [] names = [] if clinical_res: probs.append(clinical_res["prob"]); weights.append(clinical_weight); names.append("Clinical") if meibo_res: probs.append(meibo_res["prob"]); weights.append(meibo_weight); names.append("Meibography") if lipid_res: probs.append(lipid_res["prob"]); weights.append(lipid_weight); names.append("Lipidomics") if not probs: return "No valid modality input was supplied.", clinical_text, meibo_text, lipid_text probs = np.array(probs, dtype=float) weights = np.array(weights, dtype=float) weights = weights / weights.sum() final_score = float(np.sum(weights * probs)) ensemble_uncertainty = float(np.std(probs)) clinical_prob = clinical_res["prob"] if clinical_res else 0.0 meibo_prob = meibo_res["prob"] if meibo_res else 0.0 lipid_prob = lipid_res["prob"] if lipid_res else 0.0 if final_score >= 0.75: final_label = "High DED risk" elif final_score >= 0.50: final_label = "Moderate DED risk" else: final_label = "Low DED risk" phenotype = phenotype_report(clinical_prob, meibo_prob, lipid_prob) rec = make_recommendation(final_score, clinical_prob, meibo_prob, lipid_prob) report = f""" # Multimodal DED Prototype Report ## Final Result - **Final DED risk category:** {final_label} - **Final DED probability:** {final_score*100:.2f}% - **Ensemble disagreement / uncertainty:** {ensemble_uncertainty*100:.2f}% - **Likely pattern:** {phenotype} ## Modality Scores - Clinical risk probability: **{clinical_prob*100:.2f}%** - Meibography structural probability: **{meibo_prob*100:.2f}%** - Lipidomics molecular probability: **{lipid_prob*100:.2f}%** ## Interpretation {rec} ## Prototype Limitations - This system is for research support only. - The clinical module is symptom/lifestyle based and had weak standalone performance. - The meibography module currently requires gland and eyelid masks, not raw meibography images. - The lipidomics module was trained using a synthetic cohort generated from a very small public dataset. """ return report, clinical_text, meibo_text, lipid_text # ------------------------------------------------------------ # UI # ------------------------------------------------------------ with gr.Blocks(title="DED Multimodal AI Prototype") as demo: gr.Markdown(""" # AI-Assisted Dry Eye Disease Multimodal Prototype This prototype combines **clinical symptoms**, **meibography structural masks**, and **tear lipidomics biomarkers** to estimate DED risk and likely pattern. **Minimum doctor inputs for a decision:** 1. Age, screen time, sleep quality, redness, eye strain, itchiness/irritation. 2. Meibomian gland mask and eyelid mask from meibography segmentation. 3. A lipidomics CSV/TSV/XLSX file containing the top-20 lipid biomarker columns. This is a **research prototype**, not a medical diagnostic device. """) with gr.Row(): with gr.Column(): gr.Markdown("## Clinical minimum inputs") age = gr.Number(value=45, label="Age") avg_screen_time = gr.Number(value=7, label="Average screen time (hours/day)") sleep_quality = gr.Slider(1, 5, value=2, step=1, label="Sleep quality (1=poor, 5=excellent)") redness = gr.Radio(["Y", "N"], value="Y", label="Redness in eye") eye_strain = gr.Radio(["Y", "N"], value="Y", label="Discomfort / eye strain") itchiness = gr.Radio(["Y", "N"], value="Y", label="Itchiness / irritation") with gr.Column(): gr.Markdown("## Meibography masks") gland_mask = gr.Image(type="filepath", label="Upload meibomian gland mask") eyelid_mask = gr.Image(type="filepath", label="Upload eyelid mask") with gr.Column(): gr.Markdown("## Lipidomics file") lipid_file = gr.File(label="Upload lipidomics CSV/TSV/XLSX", file_types=[".csv", ".tsv", ".txt", ".xlsx", ".xls"]) gr.File(value="lipidomics_template.csv", label="Download/use lipidomics template", interactive=False) with gr.Accordion("Optional clinical fields for model compatibility", open=False): with gr.Row(): gender = gr.Radio(["M", "F"], value="M", label="Gender") sleep_duration = gr.Number(value=6.5, label="Sleep duration") stress_level = gr.Slider(1, 5, value=3, step=1, label="Stress level") heart_rate = gr.Number(value=80, label="Heart rate") daily_steps = gr.Number(value=8000, label="Daily steps") with gr.Row(): physical_activity = gr.Number(value=60, label="Physical activity") height = gr.Number(value=170, label="Height cm") weight = gr.Number(value=75, label="Weight kg") systolic_bp = gr.Number(value=120, label="Systolic BP") diastolic_bp = gr.Number(value=80, label="Diastolic BP") with gr.Row(): sleep_disorder = gr.Radio(["Y", "N"], value="N", label="Sleep disorder") wake_night = gr.Radio(["Y", "N"], value="N", label="Wake up at night") sleepy_day = gr.Radio(["Y", "N"], value="N", label="Sleepy during day") caffeine = gr.Radio(["Y", "N"], value="Y", label="Caffeine") alcohol = gr.Radio(["Y", "N"], value="N", label="Alcohol") with gr.Row(): smoking = gr.Radio(["Y", "N"], value="N", label="Smoking") medical_issue = gr.Radio(["Y", "N"], value="N", label="Medical issue") medication = gr.Radio(["Y", "N"], value="N", label="Ongoing medication") device_before_bed = gr.Radio(["Y", "N"], value="Y", label="Smart device before bed") blue_filter = gr.Radio(["Y", "N"], value="N", label="Blue light filter") with gr.Accordion("Ensemble weights", open=False): clinical_weight = gr.Slider(0, 1, value=0.20, step=0.05, label="Clinical weight") meibo_weight = gr.Slider(0, 1, value=0.50, step=0.05, label="Meibography weight") lipid_weight = gr.Slider(0, 1, value=0.30, step=0.05, label="Lipidomics weight") run_btn = gr.Button("Run DED Prototype Analysis") report = gr.Markdown() clinical_out = gr.Textbox(label="Clinical module output", lines=5) meibo_out = gr.Textbox(label="Meibography module output", lines=6) lipid_out = gr.Textbox(label="Lipidomics module output", lines=6) inputs = [ gland_mask, eyelid_mask, lipid_file, age, avg_screen_time, sleep_quality, redness, eye_strain, itchiness, gender, sleep_duration, stress_level, heart_rate, daily_steps, physical_activity, height, weight, sleep_disorder, wake_night, sleepy_day, caffeine, alcohol, smoking, medical_issue, medication, device_before_bed, blue_filter, systolic_bp, diastolic_bp, clinical_weight, meibo_weight, lipid_weight, ] run_btn.click(fn=run_system, inputs=inputs, outputs=[report, clinical_out, meibo_out, lipid_out]) if __name__ == "__main__": demo.launch()