Spaces:
Sleeping
Sleeping
File size: 6,240 Bytes
d80edbe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | """
Vitals Guard AI — Comprehensive Dashboard Predictor
Evaluates a patient's vitals across all 8 AI models to generate a unified health report.
This acts as the backend controller for the "Digital Twin" dashboard.
"""
import sys
import os
import json
import numpy as np
# Import predict functions from all 8 models
try:
# 01 Health Predictor
from importlib import import_module
m_01 = import_module("01_health_predictor.predict")
# 02 Smart Alert
m_02 = import_module("02_smart_alert.predict")
# 03 Trend Predictor
m_03 = import_module("03_trend_prediction.predict")
# 04 Silent Risk Detector
m_04 = import_module("04_silent_risk_detector.predict")
# 05 Anomaly Fingerprint
m_05 = import_module("05_anomaly_fingerprint.predict")
# 06 ECG Pattern Detection
m_06 = import_module("06_ecg_lstm.predict")
# 07 What-If Simulator
m_07 = import_module("07_whatif_simulator.predict")
# 08 Behavior-Aware Prediction
m_08 = import_module("08_behavior_health.predict")
except ImportError as e:
print(f"Error loading models. Are they trained? Details: {e}")
sys.exit(1)
from shared.utils import generate_ecg_wave
def evaluate_patient(patient_data):
"""
Run the patient data through all 8 models and compile a multi-agent consensus report.
"""
print("\n" + "="*80)
print(f" 🩺 ANALYZING PATIENT: {patient_data.get('name', 'Unknown')}")
print("="*80)
vitals = patient_data["vitals"]
vitals_sequence = patient_data.get("vitals_sequence", [vitals]*24) # Fallback if seq is missing
ecg_signal = patient_data.get("ecg_signal", np.zeros(500)) # Fallback
print(f"[CURRENT VITALS] HR: {vitals['heart_rate']} | SpO2: {vitals['spo2']}% | Temp: {vitals['temperature']}°C | RR: {vitals['respiratory_rate']} | BP: {vitals['systolic_bp']}/{vitals['diastolic_bp']}\n")
# 1. Core Health Prediction
print("▶ MODEL 1: Core Health Condition")
res_1 = m_01.predict_condition(vitals)
print(f" Predicted Condition: {res_1['predicted_condition']} (Confidence: {res_1['confidence']:.1%})")
# 2. Smart Alert System
print("\n▶ MODEL 2: Smart Alert & Emergency Triage")
res_2 = m_02.predict_alert(vitals)
severity = res_2['severity'].upper()
color = "🔴" if severity == "CRITICAL" else "🟡" if severity == "WARNING" else "🟢"
print(f" Severity Level: {color} {severity}")
print(f" Action Required: {res_2['action']}")
# 3. Trend Prediction
print("\n▶ MODEL 3: Trend-Based Forecasting")
res_3 = m_03.predict_trend(vitals_sequence)
trend_arrow = "📈" if res_3['trend'] == "deteriorating" else "📉" if res_3['trend'] == "improving" else "➡️"
print(f" Predicted Trend: {trend_arrow} {res_3['trend'].capitalize()} ({res_3['confidence']:.1%})")
# 4. Silent Risk Detector
print("\n▶ MODEL 4: Silent Risk Detector (Hidden Pattern Finder)")
res_4 = m_04.detect_silent_risk(vitals)
print(f" Risk Score: {res_4['risk_score']:.2f}/1.00")
print(f" Insight: {res_4['description']}")
# 5. Health Anomaly Fingerprint
print("\n▶ MODEL 5: Disease Signature Fingerprint")
res_5 = m_05.match_fingerprint(vitals)
print(f" {res_5['description']}")
matches_str = ', '.join([f"{m['disease']} ({m['similarity']:.0%})" for m in res_5['top_3_matches']])
print(f" Matches: {matches_str}")
# 6. ECG LSTM
print("\n▶ MODEL 6: Real-time ECG Pattern Analysis")
res_6 = m_06.detect_ecg_anomalies(ecg_signal)
print(f" Result: {res_6['summary']}")
# 7. What-If Health Simulator
print("\n▶ MODEL 7: 24-Hour Risk Progression Simulator")
res_7 = m_07.simulate_whatif(vitals)
print(" Progression Timeline:")
for ts in ["2h", "6h", "12h", "24h"]:
print(f" - {ts:3} > Risk [{res_7[ts]['risk_level']}] : {res_7[ts]['condition']}")
# 8. Behavior-Aware Health Prediction
print("\n▶ MODEL 8: Behavior-Aware Health Prediction")
res_8 = m_08.predict_behavior_risk(vitals)
print(f" Behavior Risk: {res_8['risk_level'].upper()}")
for rec in res_8['behavioral_insights']:
print(f" - {rec}")
print("\n" + "="*80)
if __name__ == "__main__":
# Generate some ECG signals for the tests
_, normal_ecg = generate_ecg_wave(duration_sec=2.0, sampling_rate=250, heart_rate=72)
_, abnormal_ecg = generate_ecg_wave(duration_sec=2.0, sampling_rate=250, heart_rate=135)
# Inject anomaly for testing
for i in range(200, 250): abnormal_ecg[i] += np.random.uniform(0.5, 1.5) * np.sin(i * 0.5)
test_cases = [
{
"name": "Patient A (The 'Silent Risk' Case - Vitals Normal, but Pattern Dangerous)",
"vitals": {
"heart_rate": 78, "spo2": 97, "temperature": 36.8, "respiratory_rate": 17,
"systolic_bp": 128, "diastolic_bp": 80, "hr_variability": 35,
"spo2_variability": 1.5, "hr_spo2_corr": 0.7, "rr_irregularity": 0.25,
"bp_pulse_pressure": 48, "sleep_hours": 6, "stress_level": 8, "activity_level": 3
},
# Generate stable past sequence
"vitals_sequence": [{"heart_rate": 75, "spo2": 97, "temperature": 36.8, "respiratory_rate": 17} for _ in range(24)],
"ecg_signal": normal_ecg
},
{
"name": "Patient B (Critical COVID/Respiratory Distress)",
"vitals": {
"heart_rate": 120, "spo2": 85, "temperature": 39.5, "respiratory_rate": 32,
"systolic_bp": 95, "diastolic_bp": 60, "hr_variability": 15,
"spo2_variability": 2.5, "hr_spo2_corr": 0.3, "rr_irregularity": 0.15,
"bp_pulse_pressure": 35, "sleep_hours": 3, "stress_level": 9, "activity_level": 0,
"spo2_trend": -1.5, "temp_trend": 0.5, "rr_trend": 1.5
},
# Generate deteriorating sequence
"vitals_sequence": [{"heart_rate": 90+i, "spo2": 95-(i*0.4), "temperature": 37.5+(i*0.1), "respiratory_rate": 20+(i*0.5)} for i in range(24)],
"ecg_signal": abnormal_ecg
}
]
for patient in test_cases:
evaluate_patient(patient)
|