# 🏥 HOSPITAL READMISSION PREDICTOR - HUGGING FACE SPACES DEPLOYMENT # Optimized for Hugging Face Spaces with proper model loading import os import pandas as pd import numpy as np import gradio as gr from datetime import datetime import warnings warnings.filterwarnings('ignore') # For Hugging Face Spaces - try importing required libraries try: import numpy as np print("✅ numpy imported successfully") except ImportError: print("❌ numpy not available - installing...") os.system("pip install numpy==1.26.4") import numpy as np try: import joblib print("✅ joblib imported successfully") except ImportError: print("❌ joblib not available - installing...") os.system("pip install joblib") import joblib try: import json print("✅ json imported successfully") except ImportError: print("❌ json not available") try: import pandas as pd print("✅ pandas imported successfully") except ImportError: print("❌ pandas not available - installing...") os.system("pip install pandas") import pandas as pd print("🚀 Initializing Hospital Readmission Predictor for Hugging Face Spaces...") # Model paths for Hugging Face Spaces MODEL_PATH = "models/production_model.pkl" PREPROCESSOR_PATH = "models/smoteenn_preprocessor.pkl" MODEL_INFO_PATH = "models/model_info.json" print(f"📁 Looking for model files...") print(f"🤖 Model file exists: {os.path.exists(MODEL_PATH)}") print(f"⚙️ Preprocessor file exists: {os.path.exists(PREPROCESSOR_PATH)}") print(f"📋 Model info file exists: {os.path.exists(MODEL_INFO_PATH)}") # Load models with error handling for Hugging Face Spaces model = None preprocessor = None model_info = None try: if os.path.exists(MODEL_PATH): print("📥 Loading production model...") # Try to fix numpy._core issue try: import numpy as np # Force numpy to load properly np.random.seed(42) except: print("⚠️ Numpy issue detected, attempting fix...") model = joblib.load(MODEL_PATH) print("✅ Production model loaded successfully!") else: print("⚠️ Model file not found - using demo mode") if os.path.exists(PREPROCESSOR_PATH): print("📥 Loading preprocessor...") preprocessor = joblib.load(PREPROCESSOR_PATH) print("✅ Preprocessor loaded successfully!") else: print("⚠️ Preprocessor file not found") if os.path.exists(MODEL_INFO_PATH): print("📥 Loading model information...") with open(MODEL_INFO_PATH, 'r') as f: model_info = json.load(f) print("✅ Model information loaded successfully!") print(f"🎯 Model Accuracy: {model_info['accuracy']:.2%}") print(f"🏆 AUC Score: {model_info['auc']:.4f}") else: print("⚠️ Model info file not found - using default info") model_info = { "model_type": "RandomForest Classifier", "accuracy": 0.7471, "auc": 0.827, "feature_count": 62, "training_samples": 12236, "training_technique": "SMOTEENN", "created_date": "2025-09-10" } except Exception as e: print(f"❌ Error loading models: {e}") print("🔄 Running in demo mode...") # Ensure model_info has default values even in error case if model_info is None: model_info = { "model_type": "RandomForest Classifier (Demo Mode)", "accuracy": 0.7471, "auc": 0.827, "feature_count": 62, "training_samples": 12236, "training_technique": "SMOTEENN", "created_date": "2025-09-10" } class HospitalReadmissionPredictor: def __init__(self, model, preprocessor, model_info): self.model = model self.preprocessor = preprocessor self.model_info = model_info def calculate_lace_score(self, length_of_stay, acuity, comorbidity, emergency_visits): """Calculate LACE score for readmission risk""" # Length of stay points if length_of_stay >= 14: length_points = 7 elif length_of_stay >= 7: length_points = 5 elif length_of_stay >= 4: length_points = 4 elif length_of_stay >= 3: length_points = 3 elif length_of_stay == 2: length_points = 2 elif length_of_stay == 1: length_points = 1 else: length_points = 0 # Acuity points (admission type) acuity_points = 3 if acuity == "Emergency" else 0 # Comorbidity points (Charlson index approximation) if comorbidity >= 4: comorbidity_points = 5 elif comorbidity >= 3: comorbidity_points = 3 elif comorbidity >= 2: comorbidity_points = 2 elif comorbidity == 1: comorbidity_points = 1 else: comorbidity_points = 0 # Emergency visits points if emergency_visits >= 4: emergency_points = 4 elif emergency_visits >= 2: emergency_points = 2 elif emergency_visits == 1: emergency_points = 1 else: emergency_points = 0 return length_points + acuity_points + comorbidity_points + emergency_points def calculate_hospital_score(self, hemoglobin, discharge_sodium, length_of_stay, procedure_count, admission_type, comorbidity_index): """Calculate HOSPITAL score""" score = 0 # Hemoglobin < 12 g/dL if hemoglobin < 12: score += 1 # Discharge from Oncology (approximated) if procedure_count >= 3 and comorbidity_index >= 2: score += 2 # Sodium < 135 mEq/L if discharge_sodium < 135: score += 1 # Procedure during stay if procedure_count >= 1: score += 1 # Index admission type (non-elective) if admission_type in ["Emergency", "Urgent"]: score += 1 # Length of stay >= 5 days if length_of_stay >= 5: score += 2 return score def _prepare_features(self, age, time_in_hospital, n_lab_procedures, n_procedures, n_medications, n_outpatient, n_inpatient, n_emergency, medical_specialty, primary_diagnosis, admission_type, discharge_disposition, glucose_test, a1c_test, diabetes_med, change_diabetes_med, insulin, hemoglobin, sodium): """Prepare features for the ML model""" features = [] # Basic numerical features features.extend([ age, time_in_hospital, n_lab_procedures, n_procedures, n_medications, n_outpatient, n_inpatient, n_emergency, hemoglobin, sodium ]) # Age groups (one-hot encoding) features.extend([ 1 if age < 30 else 0, 1 if 30 <= age < 50 else 0, 1 if 50 <= age < 70 else 0, 1 if age >= 70 else 0, ]) # Medical specialty encoding specialty_map = { "InternalMedicine": 0, "Cardiology": 1, "Surgery": 2, "Family/GeneralPractice": 3, "Endocrinology": 4, "Orthopedics": 5, "Psychiatry": 6, "Pediatrics": 7, "Emergency/Trauma": 8, "Other": 9 } specialty_features = [0] * 10 if medical_specialty in specialty_map: specialty_features[specialty_map[medical_specialty]] = 1 features.extend(specialty_features) # Diagnosis encoding diagnosis_map = { "Circulatory": 0, "Diabetes": 1, "Respiratory": 2, "Digestive": 3, "Genitourinary": 4, "Injury": 5, "Musculoskeletal": 6, "Neoplasms": 7, "Mental Disorders": 8, "Other": 8 } diagnosis_features = [0] * 9 if primary_diagnosis in diagnosis_map: diagnosis_features[diagnosis_map[primary_diagnosis]] = 1 features.extend(diagnosis_features) # Admission type, discharge disposition, diabetes indicators features.extend([ 1 if admission_type == "Emergency" else 0, 1 if admission_type == "Urgent" else 0, 1 if admission_type == "Elective" else 0, 1 if discharge_disposition == "Home" else 0, 1 if discharge_disposition == "Home Health Service" else 0, 1 if discharge_disposition == "Skilled Nursing Facility" else 0, 1 if diabetes_med == "Yes" else 0, 1 if glucose_test in [">200", ">300"] else 0, 1 if a1c_test in [">7", ">8", ">9"] else 0, 1 if insulin in ["Up", "Steady"] else 0 ]) # Clinical risk indicators features.extend([ 1 if hemoglobin < 12 else 0, 1 if sodium < 135 else 0, 1 if time_in_hospital >= 7 else 0, 1 if n_medications >= 15 else 0, 1 if n_emergency >= 2 else 0, ]) # Clinical scores lace_score = self.calculate_lace_score( time_in_hospital, admission_type, min(3, (n_inpatient + n_emergency) // 2), n_emergency ) hospital_score = self.calculate_hospital_score( hemoglobin, sodium, time_in_hospital, n_procedures, admission_type, min(3, (n_inpatient + n_emergency) // 2) ) features.extend([lace_score, hospital_score]) # Pad to 62 features while len(features) < 62: features.append(0) features = features[:62] return np.array(features).reshape(1, -1) def predict_readmission(self, age, time_in_hospital, n_lab_procedures, n_procedures, n_medications, n_outpatient, n_inpatient, n_emergency, medical_specialty, primary_diagnosis, admission_type, discharge_disposition, glucose_test, a1c_test, diabetes_med, change_diabetes_med, insulin, hemoglobin, sodium): """Main prediction function""" try: # Calculate clinical scores for interpretability lace_score = self.calculate_lace_score( time_in_hospital, admission_type, min(3, (n_inpatient + n_emergency) // 2), n_emergency ) hospital_score = self.calculate_hospital_score( hemoglobin, sodium, time_in_hospital, n_procedures, admission_type, min(3, (n_inpatient + n_emergency) // 2) ) # Use actual ML model if available, otherwise use clinical scoring if self.model is not None: try: input_features = self._prepare_features( age, time_in_hospital, n_lab_procedures, n_procedures, n_medications, n_outpatient, n_inpatient, n_emergency, medical_specialty, primary_diagnosis, admission_type, discharge_disposition, glucose_test, a1c_test, diabetes_med, change_diabetes_med, insulin, hemoglobin, sodium ) readmission_probability = self.model.predict_proba(input_features)[0][1] model_prediction = self.model.predict(input_features)[0] prediction_source = "🤖 ML Model Prediction" except Exception as e: print(f"Model prediction failed: {e}, using clinical scoring") readmission_probability = self._clinical_prediction( age, time_in_hospital, n_medications, n_emergency, n_inpatient, lace_score, hospital_score, hemoglobin, sodium ) model_prediction = 1 if readmission_probability > 0.5 else 0 prediction_source = "📊 Clinical Scoring (Demo Mode)" else: # Demo mode - use clinical scoring readmission_probability = self._clinical_prediction( age, time_in_hospital, n_medications, n_emergency, n_inpatient, lace_score, hospital_score, hemoglobin, sodium ) model_prediction = 1 if readmission_probability > 0.5 else 0 prediction_source = "📊 Clinical Scoring (Demo Mode)" # Risk factor analysis risk_factors = self._analyze_risk_factors( age, time_in_hospital, n_medications, n_lab_procedures, n_procedures, n_emergency, n_inpatient, diabetes_med, glucose_test, a1c_test, insulin, hemoglobin, sodium, medical_specialty, discharge_disposition ) # Risk categorization if readmission_probability >= 0.7: risk_level = "🔴 VERY HIGH RISK" risk_color = "#d32f2f" recommendation = "Immediate intervention required. Consider discharge planning team, home health services, and close follow-up within 48-72 hours." elif readmission_probability >= 0.5: risk_level = "🟠 HIGH RISK" risk_color = "#f57c00" recommendation = "Enhanced discharge planning recommended. Schedule follow-up within 7 days and consider transitional care services." elif readmission_probability >= 0.3: risk_level = "🟡 MODERATE RISK" risk_color = "#ffa000" recommendation = "Standard discharge planning with follow-up within 14 days. Monitor medication adherence." else: risk_level = "🟢 LOW RISK" risk_color = "#388e3c" recommendation = "Routine discharge planning. Standard follow-up care as clinically indicated." # Create result HTML return self._create_result_html( risk_level, risk_color, readmission_probability, model_prediction, lace_score, hospital_score, risk_factors, recommendation, prediction_source, age, time_in_hospital, n_medications, n_lab_procedures, n_procedures, n_emergency, n_inpatient, n_outpatient, medical_specialty, primary_diagnosis, hemoglobin, sodium ) except Exception as e: return self._create_error_output(f"❌ Prediction Error: {str(e)}") def _clinical_prediction(self, age, time_in_hospital, n_medications, n_emergency, n_inpatient, lace_score, hospital_score, hemoglobin, sodium): """Clinical scoring-based prediction for demo mode""" base_risk = 0.2 # Base 20% risk # Age factor if age >= 75: base_risk += 0.15 elif age >= 65: base_risk += 0.10 elif age >= 50: base_risk += 0.05 # Length of stay factor if time_in_hospital >= 10: base_risk += 0.20 elif time_in_hospital >= 7: base_risk += 0.15 elif time_in_hospital >= 4: base_risk += 0.10 # Medication complexity if n_medications >= 20: base_risk += 0.15 elif n_medications >= 15: base_risk += 0.10 elif n_medications >= 10: base_risk += 0.05 # Healthcare utilization if n_emergency >= 3: base_risk += 0.15 elif n_emergency >= 1: base_risk += 0.10 if n_inpatient >= 2: base_risk += 0.10 elif n_inpatient >= 1: base_risk += 0.05 # Clinical indicators if hemoglobin < 10: base_risk += 0.10 elif hemoglobin < 12: base_risk += 0.05 if sodium < 130: base_risk += 0.10 elif sodium < 135: base_risk += 0.05 # LACE and HOSPITAL scores base_risk += lace_score * 0.02 base_risk += hospital_score * 0.03 return min(base_risk, 0.95) # Cap at 95% def _analyze_risk_factors(self, age, time_in_hospital, n_medications, n_lab_procedures, n_procedures, n_emergency, n_inpatient, diabetes_med, glucose_test, a1c_test, insulin, hemoglobin, sodium, medical_specialty, discharge_disposition): """Analyze and return risk factors""" risk_factors = [] if age >= 75: risk_factors.append("Advanced age (75+)") elif age >= 65: risk_factors.append("Elderly (65-74)") if time_in_hospital >= 10: risk_factors.append("Extended hospitalization (10+ days)") elif time_in_hospital >= 7: risk_factors.append("Long hospitalization (7-9 days)") if n_medications >= 20: risk_factors.append("High medication burden (20+)") elif n_medications >= 15: risk_factors.append("Moderate medication burden (15-19)") if n_emergency >= 3: risk_factors.append("Frequent emergency visits (3+)") elif n_emergency >= 1: risk_factors.append("Recent emergency visits") if n_inpatient >= 2: risk_factors.append("Multiple previous admissions") if diabetes_med == "Yes": risk_factors.append("Diabetes medication") if glucose_test in [">200", ">300"]: risk_factors.append("Poor glucose control") if a1c_test in [">8", ">9"]: risk_factors.append("Poor diabetes control (HbA1c)") if hemoglobin < 10: risk_factors.append("Severe anemia") elif hemoglobin < 12: risk_factors.append("Mild anemia") if sodium < 130: risk_factors.append("Severe hyponatremia") elif sodium < 135: risk_factors.append("Mild hyponatremia") if medical_specialty in ["Cardiology", "Surgery", "InternalMedicine"]: risk_factors.append(f"High-risk specialty ({medical_specialty})") if discharge_disposition in ["Home Health Service", "Skilled Nursing Facility"]: risk_factors.append("Post-acute care needs") return risk_factors def _create_result_html(self, risk_level, risk_color, readmission_probability, model_prediction, lace_score, hospital_score, risk_factors, recommendation, prediction_source, age, time_in_hospital, n_medications, n_lab_procedures, n_procedures, n_emergency, n_inpatient, n_outpatient, medical_specialty, primary_diagnosis, hemoglobin, sodium): """Create formatted HTML result""" clinical_risk_score = len(risk_factors) return f"""

🏥 Hospital Readmission Risk Assessment

{prediction_source}

{risk_level}

Readmission Probability: {readmission_probability:.1%}

Prediction: {'Readmission' if model_prediction == 1 else 'No Readmission'}

Risk Factors: {clinical_risk_score}

LACE Score: {lace_score}

HOSPITAL Score: {hospital_score}

📊 Patient Summary

Age: {age} years
Length of Stay: {time_in_hospital} days
Medications: {n_medications}
Lab Procedures: {n_lab_procedures}
Procedures: {n_procedures}
Emergency Visits: {n_emergency}
Previous Admissions: {n_inpatient}
Outpatient Visits: {n_outpatient}
Specialty: {medical_specialty}
Primary Diagnosis: {primary_diagnosis}
Hemoglobin: {hemoglobin} g/dL
Sodium: {sodium} mEq/L

⚠️ Identified Risk Factors

{''.join([f'
• {factor}
' for factor in risk_factors]) if risk_factors else '

✅ No major risk factors identified

'}

💡 Clinical Recommendations

{recommendation}

""" def _create_error_output(self, error_message): """Create formatted error output""" return f"""

❌ Error

{error_message}

Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

""" # Initialize predictor predictor = HospitalReadmissionPredictor(model, preprocessor, model_info) print("✅ Hospital Readmission Predictor initialized!") # Create Gradio interface def create_gradio_interface(): """Create the Gradio interface for Hugging Face Spaces""" with gr.Blocks( title="🏥 Hospital Readmission Risk Predictor" ) as demo: gr.Markdown(""" # 🏥 Hospital Readmission Risk Predictor ## AI-Powered Clinical Decision Support System **Model Performance:** 74.71% Accuracy | AUC: 0.827 | RandomForest + SMOTEENN This system uses machine learning to assess hospital readmission risk based on patient data and clinical indicators. """) with gr.Row(): # Left Column with gr.Column(scale=1): gr.Markdown("### 👤 Patient Demographics") age = gr.Slider(18, 100, value=65, step=1, label="Age (years)") gr.Markdown("### 🏥 Hospital Stay") time_in_hospital = gr.Slider(1, 30, value=4, step=1, label="Length of Stay (days)") admission_type = gr.Dropdown( ["Elective", "Emergency", "Urgent", "Not Available"], value="Emergency", label="Admission Type" ) discharge_disposition = gr.Dropdown( ["Home", "Home Health Service", "Skilled Nursing Facility", "Other"], value="Home", label="Discharge Disposition" ) gr.Markdown("### ⚕️ Medical Specialty") medical_specialty = gr.Dropdown( ["InternalMedicine", "Cardiology", "Surgery", "Family/GeneralPractice", "Endocrinology", "Orthopedics", "Psychiatry", "Pediatrics", "Other"], value="InternalMedicine", label="Primary Medical Specialty" ) primary_diagnosis = gr.Dropdown( ["Circulatory", "Diabetes", "Respiratory", "Digestive", "Genitourinary", "Injury", "Musculoskeletal", "Neoplasms", "Mental Disorders", "Other"], value="Circulatory", label="Primary Diagnosis" ) # Middle Column with gr.Column(scale=1): gr.Markdown("### 🔬 Procedures & Tests") n_lab_procedures = gr.Slider(0, 150, value=40, step=1, label="Lab Procedures") n_procedures = gr.Slider(0, 15, value=2, step=1, label="Medical Procedures") n_medications = gr.Slider(1, 50, value=12, step=1, label="Number of Medications") gr.Markdown("### 📅 Healthcare History") n_outpatient = gr.Slider(0, 40, value=2, step=1, label="Outpatient Visits (Past Year)") n_inpatient = gr.Slider(0, 20, value=1, step=1, label="Inpatient Admissions (Past Year)") n_emergency = gr.Slider(0, 25, value=1, step=1, label="Emergency Visits (Past Year)") # Right Column with gr.Column(scale=1): gr.Markdown("### 🩺 Laboratory Values") hemoglobin = gr.Slider(5.0, 18.0, value=12.5, step=0.1, label="Hemoglobin (g/dL)") sodium = gr.Slider(120, 150, value=140, step=1, label="Serum Sodium (mEq/L)") gr.Markdown("### 🍯 Diabetes Management") glucose_test = gr.Dropdown( ["None", "Norm", ">200", ">300", "Not Performed"], value="None", label="Glucose Test" ) a1c_test = gr.Dropdown( ["None", "Norm", ">7", ">8", ">9", "Not Performed"], value="None", label="HbA1c Test" ) diabetes_med = gr.Dropdown(["No", "Yes"], value="No", label="Diabetes Medication") change_diabetes_med = gr.Dropdown( ["No", "Ch", "Up", "Down"], value="No", label="Change in Diabetes Med" ) insulin = gr.Dropdown( ["No", "Down", "Steady", "Up"], value="No", label="Insulin Treatment" ) # Prediction Button predict_btn = gr.Button("🔮 Predict Readmission Risk", variant="primary", size="lg") # Results output = gr.HTML(label="Prediction Results") # Footer gr.Markdown(""" --- **⚠️ Disclaimer:** This tool is for clinical decision support only. Always consult healthcare professionals. **📊 Model Info:** RandomForest + SMOTEENN | 74.71% Accuracy | 62 Features | LACE & HOSPITAL Scores """) # Set up prediction predict_btn.click( fn=predictor.predict_readmission, inputs=[ age, time_in_hospital, n_lab_procedures, n_procedures, n_medications, n_outpatient, n_inpatient, n_emergency, medical_specialty, primary_diagnosis, admission_type, discharge_disposition, glucose_test, a1c_test, diabetes_med, change_diabetes_med, insulin, hemoglobin, sodium ], outputs=output ) return demo # Create and launch the interface if __name__ == "__main__": demo = create_gradio_interface() demo.launch()