Spaces:
Sleeping
Sleeping
| # ============================================================ | |
| # 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() | |