""" VitalLink Triage Engine Centralized triage assessment algorithm based on hospital emergency standards Combines elements from US ESI, UK Manchester, and Canadian CTAS systems """ from datetime import datetime from enum import IntEnum from typing import List, Optional, Dict from dataclasses import dataclass # ============================================================================ # TRIAGE LEVELS (Lower number = Higher priority) # ============================================================================ class TriageLevel(IntEnum): """ Four-tier triage system Based on Emergency Severity Index (ESI) and similar systems """ LEVEL_1_EMERGENCY = 1 # Immediate life threat - needs resuscitation LEVEL_2_WARNING = 2 # High risk - deteriorating/unstable LEVEL_3_ALERT = 3 # Urgent - abnormal vitals or moderate risk LEVEL_4_NORMAL = 4 # Standard - stable with minor complaints # ============================================================================ # VITAL SIGN THRESHOLDS # ============================================================================ class VitalThresholds: """Clinical thresholds for vital sign assessment with hysteresis""" # Heart Rate (BPM) - Upgrade thresholds HR_CRITICAL_LOW = 40 HR_CRITICAL_HIGH = 140 HR_ABNORMAL_LOW = 50 HR_ABNORMAL_HIGH = 110 # Heart Rate - Downgrade thresholds (with hysteresis gap) HR_CRITICAL_LOW_RECOVERY = 45 # Must go above 45 to leave critical HR_CRITICAL_HIGH_RECOVERY = 135 # Must go below 135 to leave critical HR_ABNORMAL_LOW_RECOVERY = 55 HR_ABNORMAL_HIGH_RECOVERY = 105 # SpO2 (%) - Upgrade thresholds SPO2_CRITICAL = 88 SPO2_ABNORMAL = 92 # SpO2 - Downgrade thresholds SPO2_CRITICAL_RECOVERY = 90 # Must go above 90 to leave critical SPO2_ABNORMAL_RECOVERY = 94 # Must go above 94 to leave abnormal # Temperature (°C) - Upgrade thresholds TEMP_CRITICAL_LOW = 35.0 TEMP_CRITICAL_HIGH = 39.5 TEMP_ABNORMAL_LOW = 35.5 TEMP_ABNORMAL_HIGH = 38.3 # Temperature - Downgrade thresholds TEMP_CRITICAL_LOW_RECOVERY = 35.5 TEMP_CRITICAL_HIGH_RECOVERY = 39.0 TEMP_ABNORMAL_LOW_RECOVERY = 36.0 TEMP_ABNORMAL_HIGH_RECOVERY = 38.0 # Respiratory Rate (breaths/min) RR_CRITICAL_LOW = 10 RR_CRITICAL_HIGH = 30 RR_ABNORMAL_LOW = 12 RR_ABNORMAL_HIGH = 20 @dataclass class TierStabilityTracker: """Tracks tier changes to prevent oscillation""" current_tier: str = "NORMAL" tier_since: float = 0.0 # Timestamp when tier was set consecutive_readings: int = 0 tier_change_count: int = 0 # Confirmation requirements CONFIRMATION_READINGS_UPGRADE = 3 # Need 3 consecutive readings to upgrade CONFIRMATION_READINGS_DOWNGRADE = 5 # Need 5 consecutive readings to downgrade MIN_TIME_IN_TIER = 30.0 # Minimum 30 seconds before tier can change # ============================================================================ # CLINICAL KNOWLEDGE BASE # ============================================================================ CRITICAL_CONDITIONS = [ "cardiac arrest", "respiratory arrest", "major trauma", "severe hemorrhage", "stroke", "seizure ongoing", "anaphylaxis", "severe chest pain", "unresponsive", ] HIGH_RISK_SYMPTOMS = [ "chest pain", "difficulty breathing", "altered mental status", "severe headache", "excessive bleeding", "abdominal pain", "head injury", "suspected sepsis", ] MODERATE_RISK_SYMPTOMS = [ "fever", "nausea/vomiting", "dizziness", "moderate pain", "injury/trauma", "infection signs", ] PREEXISTING_CONDITIONS = [ "diabetes", "copd", "asthma", "cancer", "hypertension", "heart disease", "kidney disease", "immunocompromised", ] # ============================================================================ # VITAL SIGNS DATA CLASS # ============================================================================ @dataclass class VitalSigns: """Current vital signs for a patient""" heart_rate: int spo2: int temperature: float activity: float = 0.0 respiratory_rate: Optional[int] = None blood_pressure_systolic: Optional[int] = None consciousness_level: str = "Alert" pain_level: int = 0 # 0-10 scale def is_critical(self) -> bool: """Check if any vital is in critical range""" return ( self.heart_rate < VitalThresholds.HR_CRITICAL_LOW or self.heart_rate > VitalThresholds.HR_CRITICAL_HIGH or self.spo2 < VitalThresholds.SPO2_CRITICAL or self.temperature < VitalThresholds.TEMP_CRITICAL_LOW or self.temperature > VitalThresholds.TEMP_CRITICAL_HIGH or ( self.respiratory_rate and ( self.respiratory_rate < VitalThresholds.RR_CRITICAL_LOW or self.respiratory_rate > VitalThresholds.RR_CRITICAL_HIGH ) ) ) def is_abnormal(self) -> bool: """Check if any vital is abnormal but not critical""" return ( self.heart_rate < VitalThresholds.HR_ABNORMAL_LOW or self.heart_rate > VitalThresholds.HR_ABNORMAL_HIGH or self.spo2 < VitalThresholds.SPO2_ABNORMAL or self.temperature < VitalThresholds.TEMP_ABNORMAL_LOW or self.temperature > VitalThresholds.TEMP_ABNORMAL_HIGH ) def get_abnormalities(self) -> List[str]: """Get list of abnormal vital signs""" abnormalities = [] if self.heart_rate < VitalThresholds.HR_CRITICAL_LOW: abnormalities.append(f"Severe Bradycardia ({self.heart_rate} bpm)") elif self.heart_rate > VitalThresholds.HR_CRITICAL_HIGH: abnormalities.append(f"Severe Tachycardia ({self.heart_rate} bpm)") elif self.heart_rate < VitalThresholds.HR_ABNORMAL_LOW: abnormalities.append(f"Bradycardia ({self.heart_rate} bpm)") elif self.heart_rate > VitalThresholds.HR_ABNORMAL_HIGH: abnormalities.append(f"Tachycardia ({self.heart_rate} bpm)") if self.spo2 < VitalThresholds.SPO2_CRITICAL: abnormalities.append(f"Critical Hypoxia ({self.spo2}%)") elif self.spo2 < VitalThresholds.SPO2_ABNORMAL: abnormalities.append(f"Hypoxia ({self.spo2}%)") if self.temperature > VitalThresholds.TEMP_CRITICAL_HIGH: abnormalities.append(f"High Fever ({self.temperature:.1f}°C)") elif self.temperature > VitalThresholds.TEMP_ABNORMAL_HIGH: abnormalities.append(f"Fever ({self.temperature:.1f}°C)") elif self.temperature < VitalThresholds.TEMP_CRITICAL_LOW: abnormalities.append(f"Hypothermia ({self.temperature:.1f}°C)") return abnormalities def is_critical_with_hysteresis(self, current_tier: str) -> bool: """Check if critical, with hysteresis based on current tier""" if current_tier == "EMERGENCY": # Already in EMERGENCY - use recovery thresholds (harder to leave) return ( self.heart_rate < VitalThresholds.HR_CRITICAL_LOW_RECOVERY or self.heart_rate > VitalThresholds.HR_CRITICAL_HIGH_RECOVERY or self.spo2 < VitalThresholds.SPO2_CRITICAL_RECOVERY or self.temperature < VitalThresholds.TEMP_CRITICAL_LOW_RECOVERY or self.temperature > VitalThresholds.TEMP_CRITICAL_HIGH_RECOVERY ) else: # Not in EMERGENCY - use standard thresholds (easier to enter) return self.is_critical() def is_abnormal_with_hysteresis(self, current_tier: str) -> bool: """Check if abnormal, with hysteresis based on current tier""" if current_tier in ["ALERT", "EMERGENCY"]: # Already elevated - use recovery thresholds return ( self.heart_rate < VitalThresholds.HR_ABNORMAL_LOW_RECOVERY or self.heart_rate > VitalThresholds.HR_ABNORMAL_HIGH_RECOVERY or self.spo2 < VitalThresholds.SPO2_ABNORMAL_RECOVERY or self.temperature < VitalThresholds.TEMP_ABNORMAL_LOW_RECOVERY or self.temperature > VitalThresholds.TEMP_ABNORMAL_HIGH_RECOVERY ) else: # Normal tier - use standard thresholds return self.is_abnormal() # ============================================================================ # TRIAGE ENGINE # ============================================================================ class TriageEngine: """ Central triage assessment engine Implements hospital-standard triage algorithms """ @staticmethod def assess_patient( vitals: VitalSigns, symptoms: List[str], severity: str, age: Optional[int] = None, preexisting: List[str] = None, wait_time_minutes: int = 0, ) -> Dict: """ Comprehensive triage assessment Returns triage level, priority score, and reasoning """ preexisting = preexisting or [] # Step 1: Check for immediate life threats if TriageEngine._is_life_threatening(vitals, symptoms): return { "triage_level": TriageLevel.LEVEL_1_EMERGENCY, "tier_name": "EMERGENCY", "priority_score": 200.0, "reasoning": "Life-threatening condition detected", "abnormalities": vitals.get_abnormalities(), "recommended_action": "Immediate intervention required", } # Step 2: Assess high-risk situations if TriageEngine._is_high_risk(vitals, symptoms, severity, age, preexisting): base_score = 100.0 # Add points for vital abnormalities if vitals.is_abnormal(): base_score += 30.0 # Add wait time urgency if wait_time_minutes > 30: base_score += (wait_time_minutes - 30) * 0.5 return { "triage_level": TriageLevel.LEVEL_2_WARNING, "tier_name": "ALERT", "priority_score": base_score, "reasoning": "High-risk patient with concerning symptoms", "abnormalities": vitals.get_abnormalities(), "recommended_action": "Rapid assessment needed", } # Step 3: Evaluate moderate urgency resources_needed = TriageEngine._estimate_resources(symptoms, severity) if resources_needed >= 2 or vitals.is_abnormal(): base_score = 50.0 if vitals.is_abnormal(): base_score += 20.0 if wait_time_minutes > 45: base_score += (wait_time_minutes - 45) * 0.8 # Check for preexisting conditions if any( cond.lower() in [p.lower() for p in preexisting] for cond in PREEXISTING_CONDITIONS ): base_score += 15.0 return { "triage_level": TriageLevel.LEVEL_3_ALERT, "tier_name": "ALERT", "priority_score": base_score, "reasoning": "Urgent care needed - abnormal vitals or complex case", "abnormalities": vitals.get_abnormalities(), "recommended_action": "See within 30 minutes", } # Step 4: Standard care base_score = 20.0 # Severity modifier if severity == "severe": base_score += 15.0 elif severity == "moderate": base_score += 10.0 else: base_score += 5.0 # Long wait compensation if wait_time_minutes > 60: base_score += (wait_time_minutes - 60) * 1.0 return { "triage_level": TriageLevel.LEVEL_4_NORMAL, "tier_name": "NORMAL", "priority_score": base_score, "reasoning": "Stable patient - standard care", "abnormalities": [], "recommended_action": "See within 120 minutes", } @staticmethod def _is_life_threatening(vitals: VitalSigns, symptoms: List[str]) -> bool: """Detect immediate life threats""" # Critical vital signs if vitals.is_critical(): return True # Unresponsive patient if vitals.consciousness_level.lower() in ["unresponsive", "unconscious"]: return True # Critical conditions in symptoms return any( cond.lower() in " ".join(symptoms).lower() for cond in CRITICAL_CONDITIONS ) @staticmethod def _is_high_risk( vitals: VitalSigns, symptoms: List[str], severity: str, age: Optional[int], preexisting: List[str], ) -> bool: """Detect high-risk patients""" # High-risk age groups if age and (age < 1 or age > 65): if vitals.is_abnormal() or severity == "severe": return True # High-risk symptoms symptom_text = " ".join(symptoms).lower() if any(symptom.lower() in symptom_text for symptom in HIGH_RISK_SYMPTOMS): return True # Severe self-reported severity with abnormal vitals if severity == "severe" and vitals.is_abnormal(): return True # Preexisting conditions with concerning vitals if preexisting and vitals.is_abnormal(): return True return False @staticmethod def _estimate_resources(symptoms: List[str], severity: str) -> int: """ Estimate number of resources needed Resources: labs, imaging, IV access, procedures, consultations """ resource_count = 0 symptom_text = " ".join(symptoms).lower() # Likely needs labs if any(s in symptom_text for s in ["fever", "infection", "pain", "bleeding"]): resource_count += 1 # Likely needs imaging if any( s in symptom_text for s in ["head injury", "chest pain", "abdominal", "trauma"] ): resource_count += 1 # Likely needs IV access if any( s in symptom_text for s in ["difficulty breathing", "severe", "bleeding", "dehydration"] ): resource_count += 1 # Complex intervention likely if severity == "severe": resource_count += 1 return resource_count @staticmethod def calculate_priority_score( triage_level: TriageLevel, vitals: VitalSigns, severity: str, wait_time_minutes: int, preexisting: List[str] = None, ) -> float: """ Calculate final priority score for queue ordering Higher score = higher priority """ preexisting = preexisting or [] # Base score from triage level level_scores = { TriageLevel.LEVEL_1_EMERGENCY: 200.0, TriageLevel.LEVEL_2_WARNING: 100.0, TriageLevel.LEVEL_3_ALERT: 50.0, TriageLevel.LEVEL_4_NORMAL: 20.0, } score = level_scores.get(triage_level, 0.0) # Vital sign modifiers if vitals.is_critical(): score += 50.0 elif vitals.is_abnormal(): score += 20.0 # Specific vital concerns if vitals.heart_rate > 140 or vitals.heart_rate < 40: score += 30.0 if vitals.spo2 < 88: score += 40.0 if vitals.temperature > 39.5: score += 25.0 # Severity modifier severity_scores = {"severe": 25, "moderate": 15, "mild": 5} score += severity_scores.get(severity, 0) # Wait time escalation if wait_time_minutes > 30: score += (wait_time_minutes - 30) * 0.5 if wait_time_minutes > 60: score += (wait_time_minutes - 60) * 1.0 # Accelerate if wait_time_minutes > 120: score += (wait_time_minutes - 120) * 2.0 # Major escalation # Preexisting conditions if preexisting and triage_level != TriageLevel.LEVEL_1_EMERGENCY: score += 10.0 * len(preexisting) return round(score, 1) @staticmethod def detect_deterioration( current_vitals: VitalSigns, previous_vitals: List[VitalSigns], time_window_minutes: int = 10, ) -> Dict: """ Detect if patient is deteriorating based on vital sign trends Returns trend analysis and recommendation """ if len(previous_vitals) < 3: return { "deteriorating": False, "reason": "Insufficient data for trend analysis", } # Take last 3 readings recent = previous_vitals[-3:] # Calculate trends hr_trend = current_vitals.heart_rate - sum(v.heart_rate for v in recent) / len( recent ) spo2_trend = sum(v.spo2 for v in recent) / len(recent) - current_vitals.spo2 temp_trend = current_vitals.temperature - sum( v.temperature for v in recent ) / len(recent) # Detect concerning trends concerns = [] if hr_trend > 15: concerns.append(f"Increasing HR (+{hr_trend:.0f} bpm)") if spo2_trend > 3: concerns.append(f"Decreasing SpO2 (-{spo2_trend:.0f}%)") if temp_trend > 0.5: concerns.append(f"Rising temperature (+{temp_trend:.1f}°C)") # Crossed into abnormal range if current_vitals.is_critical() and not any(v.is_critical() for v in recent): concerns.append("Vitals crossed into CRITICAL range") elif current_vitals.is_abnormal() and not any(v.is_abnormal() for v in recent): concerns.append("Vitals crossed into ABNORMAL range") deteriorating = len(concerns) > 0 return { "deteriorating": deteriorating, "concerns": concerns, "hr_trend": hr_trend, "spo2_trend": spo2_trend, "temp_trend": temp_trend, "recommendation": "Escalate care immediately" if deteriorating else "Continue monitoring", } @staticmethod def should_escalate_tier( current_tier: TriageLevel, vitals: VitalSigns, wait_time_minutes: int ) -> tuple[bool, Optional[TriageLevel], str]: """ Determine if patient tier should be escalated Returns (should_escalate, new_tier, reason) """ # Always escalate if vitals become critical if vitals.is_critical() and current_tier != TriageLevel.LEVEL_1_EMERGENCY: return ( True, TriageLevel.LEVEL_1_EMERGENCY, "Vitals entered critical range", ) # Escalate NORMAL to ALERT if vitals abnormal if current_tier == TriageLevel.LEVEL_4_NORMAL and vitals.is_abnormal(): return (True, TriageLevel.LEVEL_3_ALERT, "Vitals became abnormal") # Escalate ALERT to WARNING if vitals worsening if current_tier == TriageLevel.LEVEL_3_ALERT and vitals.is_critical(): return ( True, TriageLevel.LEVEL_2_WARNING, "Vitals worsening toward critical", ) # Time-based escalation for long waits if wait_time_minutes > 120 and current_tier == TriageLevel.LEVEL_4_NORMAL: return (True, TriageLevel.LEVEL_3_ALERT, "Excessive wait time (>2 hours)") if wait_time_minutes > 180 and current_tier == TriageLevel.LEVEL_3_ALERT: return (True, TriageLevel.LEVEL_2_WARNING, "Critical wait time (>3 hours)") return (False, None, "No escalation needed") # ============================================================================ # CONVENIENCE FUNCTIONS # ============================================================================ def triage_from_vitals(hr: int, spo2: int, temp: float) -> str: """ Quick triage tier determination from vital signs only Returns: "EMERGENCY", "ALERT", or "NORMAL" """ vitals = VitalSigns(heart_rate=hr, spo2=spo2, temperature=temp) if vitals.is_critical(): return "EMERGENCY" elif vitals.is_abnormal(): return "ALERT" else: return "NORMAL" def assess_new_patient( hr: int, spo2: int, temp: float, symptoms: List[str], severity: str, age: Optional[int] = None, preexisting: List[str] = None, ) -> Dict: """ Complete assessment for new patient Returns full triage result with score and recommendations """ vitals = VitalSigns(heart_rate=hr, spo2=spo2, temperature=temp) result = TriageEngine.assess_patient( vitals=vitals, symptoms=symptoms, severity=severity, age=age, preexisting=preexisting, wait_time_minutes=0, ) return result def reassess_patient( patient_data: Dict, current_vitals: Dict, wait_time_minutes: int ) -> Dict: """ Reassess existing patient with updated vitals and wait time Returns updated triage assessment """ vitals = VitalSigns( heart_rate=current_vitals.get("hr_bpm", 75), spo2=current_vitals.get("spo2", 98), temperature=current_vitals.get("temp_c", 37.0), activity=current_vitals.get("activity", 0.0), ) result = TriageEngine.assess_patient( vitals=vitals, symptoms=patient_data.get("symptoms", []), severity=patient_data.get("severity", "moderate"), age=patient_data.get("age"), preexisting=patient_data.get("preexisting", []), wait_time_minutes=wait_time_minutes, ) return result # ============================================================================ # EXAMPLE USAGE # ============================================================================ if __name__ == "__main__": print("=" * 80) print("VitalLink Triage Engine - Test Cases") print("=" * 80 + "\n") # Test Case 1: Critical patient print("Test 1: Critical Emergency") result = assess_new_patient( hr=155, spo2=85, temp=40.2, symptoms=["Severe Chest Pain", "Difficulty Breathing"], severity="severe", ) print(f" Level: {result['triage_level'].name}") print(f" Tier: {result['tier_name']}") print(f" Score: {result['priority_score']}") print(f" Reason: {result['reasoning']}") print(f" Abnormalities: {result['abnormalities']}") print() # Test Case 2: Deteriorating patient print("Test 2: Deteriorating Patient") result = assess_new_patient( hr=118, spo2=93, temp=38.5, symptoms=["Fever", "Difficulty Breathing"], severity="moderate", ) print(f" Level: {result['triage_level'].name}") print(f" Tier: {result['tier_name']}") print(f" Score: {result['priority_score']}") print() # Test Case 3: Stable patient print("Test 3: Stable Patient") result = assess_new_patient( hr=75, spo2=98, temp=36.9, symptoms=["Minor Pain"], severity="mild" ) print(f" Level: {result['triage_level'].name}") print(f" Tier: {result['tier_name']}") print(f" Score: {result['priority_score']}") print() # Test Case 4: Long wait escalation print("Test 4: Long Wait Time Escalation") vitals = VitalSigns(heart_rate=78, spo2=97, temperature=37.1) should_escalate, new_tier, reason = TriageEngine.should_escalate_tier( current_tier=TriageLevel.LEVEL_4_NORMAL, vitals=vitals, wait_time_minutes=130 ) print(f" Should escalate: {should_escalate}") print(f" New tier: {new_tier.name if new_tier else 'N/A'}") print(f" Reason: {reason}") print() print("=" * 80)