import gradio as gr
import pandas as pd
import numpy as np
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import seaborn as sns
import joblib
import shap
import pandas as pd
def add_features(df: pd.DataFrame) -> pd.DataFrame:
df = df.copy()
df['creatinine_sodium_ratio'] = df['serum_creatinine'] / df['serum_sodium']
df['is_elderly'] = (df['age'] >= 65).astype(int)
return df
xgb_pipeline = joblib.load('heart_failure_xgb_pipeline.joblib')
cph_model = joblib.load('heart_failure_cph_model.joblib')
kmeans_artifact = joblib.load('heart_failure_kmeans.joblib')
xgb_model = xgb_pipeline.named_steps['xgb']
xgb_scaler = xgb_pipeline.named_steps['scaler']
kmeans_model = kmeans_artifact['model']
kmeans_scaler = kmeans_artifact['scaler']
kmeans_features = kmeans_artifact['features']
cluster_profiles = kmeans_artifact['profiles']
explainer = shap.TreeExplainer(xgb_model)
PHENOTYPE_NAMES = {
0: "Phenotype A",
1: "Phenotype B",
2: "Phenotype C",
}
PHENOTYPE_COLORS = {
0: "#6366F1",
1: "#F59E0B",
2: "#EC4899",
}
def describe_phenotype(cluster_id, profiles):
"""Generates a human-readable clinical description for a cluster."""
row = profiles.loc[cluster_id]
mortality_pct = row['DEATH_EVENT'] * 100
age_desc = "elderly" if row['age'] >= 65 else "middle-aged" if row['age'] >= 50 else "younger"
ef_desc = "reduced" if row['ejection_fraction'] < 35 else "mildly reduced" if row['ejection_fraction'] < 45 else "preserved"
cr_desc = "elevated" if row['serum_creatinine'] > 1.5 else "borderline" if row['serum_creatinine'] > 1.2 else "normal"
return (
f"Predominantly {age_desc} patients (avg age {row['age']:.0f}) with "
f"{ef_desc} ejection fraction ({row['ejection_fraction']:.0f}%) and "
f"{cr_desc} serum creatinine ({row['serum_creatinine']:.1f} mg/dL). "
f"Cohort mortality rate: {mortality_pct:.0f}%."
)
def predict_and_explain(
age, anaemia, cpk, diabetes, ejection_fraction,
high_blood_pressure, platelets, serum_creatinine,
serum_sodium, sex, smoking
):
anaemia_val = 1 if anaemia == "Yes" else 0
diabetes_val = 1 if diabetes == "Yes" else 0
hbp_val = 1 if high_blood_pressure == "Yes" else 0
sex_val = 1 if sex == "Male" else 0
smoking_val = 1 if smoking == "Yes" else 0
patient_dict = {
'age': float(age),
'anaemia': int(anaemia_val),
'creatinine_phosphokinase': int(cpk),
'diabetes': int(diabetes_val),
'ejection_fraction': int(ejection_fraction),
'high_blood_pressure': int(hbp_val),
'platelets': float(platelets),
'serum_creatinine': float(serum_creatinine),
'serum_sodium': int(serum_sodium),
'sex': int(sex_val),
'smoking': int(smoking_val)
}
df_patient = pd.DataFrame([patient_dict])
df_patient = add_features(df_patient)
prob = float(xgb_pipeline.predict_proba(df_patient)[0][1])
risk_label = "HIGH RISK" if prob >= 0.5 else "STANDARD RISK"
color = "#FF4B4B" if prob >= 0.5 else "#2E7D32"
risk_md = f"
"
risk_md += f"
CLASSIFICATION
"
risk_md += f"
{risk_label}
"
risk_md += f"
Calculated Risk Probability: {prob:.2%}
"
risk_md += "
"
patient_continuous = df_patient[kmeans_features].values
patient_scaled_cluster = kmeans_scaler.transform(patient_continuous)
cluster_id = int(kmeans_model.predict(patient_scaled_cluster)[0])
pheno_name = PHENOTYPE_NAMES[cluster_id]
pheno_color = PHENOTYPE_COLORS[cluster_id]
pheno_desc = describe_phenotype(cluster_id, cluster_profiles)
pheno_md = f""
pheno_md += f"
PATIENT PHENOTYPE
"
pheno_md += f"
{pheno_name}
"
pheno_md += f"
{pheno_desc}
"
pheno_md += "
"
profile_row = cluster_profiles.loc[cluster_id]
labels = {
'age': 'Avg Age', 'ejection_fraction': 'Avg EF%',
'serum_creatinine': 'Avg Creatinine', 'serum_sodium': 'Avg Sodium',
'creatinine_phosphokinase': 'Avg CPK', 'platelets': 'Avg Platelets'
}
for feat in kmeans_features:
val = profile_row[feat]
fmt = f"{val:,.0f}" if feat in ('platelets', 'creatinine_phosphokinase') else f"{val:.1f}"
pheno_md += f"{labels[feat]}: {fmt}"
pheno_md += "
"
patient_scaled_xgb = pd.DataFrame(xgb_scaler.transform(df_patient), columns=df_patient.columns)
shap_vals = explainer(patient_scaled_xgb)
plt.close('all')
shap.plots.waterfall(shap_vals[0], show=False)
fig_shap = plt.gcf()
fig_shap.set_size_inches(9, 4.5)
plt.tight_layout()
patient_cph_df = df_patient.copy()
surv = cph_model.predict_survival_function(patient_cph_df)
plt.figure(figsize=(9, 4.5))
fig_surv = plt.gcf()
plt.plot(surv.index, surv.values, color='dodgerblue', linewidth=2.5)
plt.title("Predicted Patient Survival Probability Over Time", fontsize=12, fontweight='bold', pad=15)
plt.xlabel("Follow-up Duration (Days)", fontsize=10)
plt.ylabel("Survival Probability", fontsize=10)
plt.ylim(0, 1.05)
plt.grid(True, linestyle='--', alpha=0.5)
plt.tight_layout()
return risk_md, pheno_md, fig_surv, fig_shap
app = gr.Blocks()
with app:
gr.HTML("""
Heart Failure Mortality & Survival Predictor
Dual ML Pipeline: XGBoost Risk Classification & Cox Proportional Hazards Survival Curves
""")
with gr.Row():
with gr.Column(scale=1):
gr.HTML("""
Patient Clinical Profile
""")
age = gr.Slider(minimum=40, maximum=100, value=65, step=1, label="Age (Years)")
with gr.Row():
sex = gr.Dropdown(choices=["Female", "Male"], value="Male", label="Sex")
smoking = gr.Dropdown(choices=["No", "Yes"], value="No", label="Smoking Status")
with gr.Row():
anaemia = gr.Dropdown(choices=["No", "Yes"], value="No", label="Anaemia")
diabetes = gr.Dropdown(choices=["No", "Yes"], value="No", label="Diabetes")
high_blood_pressure = gr.Dropdown(choices=["No", "Yes"], value="No", label="Hypertension")
ejection_fraction = gr.Slider(minimum=10, maximum=80, value=35, step=1, label="Ejection Fraction (%)")
serum_creatinine = gr.Slider(minimum=0.5, maximum=10.0, value=1.2, step=0.1, label="Serum Creatinine (mg/dL)")
serum_sodium = gr.Slider(minimum=110, maximum=150, value=137, step=1, label="Serum Sodium (mEq/L)")
with gr.Row():
cpk = gr.Number(value=250, label="CPK Enzyme Level (mcg/L)", precision=0)
platelets = gr.Number(value=263000, label="Platelets Count (kiloplatelets/mL)", precision=0)
predict_btn = gr.Button("Evaluate Patient Risk Profile", variant="primary")
with gr.Column(scale=2):
gr.HTML("""
Diagnostic Results
""")
output_risk = gr.HTML(value="Enter patient parameters on the left and click 'Evaluate' to run diagnostic pipeline.
")
output_pheno = gr.HTML()
output_surv = gr.Plot(label="Survival Probability Curve")
output_shap = gr.Plot(label="SHAP Risk Attribution")
predict_btn.click(
fn=predict_and_explain,
inputs=[
age, anaemia, cpk, diabetes, ejection_fraction,
high_blood_pressure, platelets, serum_creatinine,
serum_sodium, sex, smoking
],
outputs=[output_risk, output_pheno, output_surv, output_shap]
)
if __name__ == '__main__':
app.launch(theme=gr.themes.Soft(primary_hue="blue", secondary_hue="slate"))