Cookbook 368: Offline Clinical Critical Care Mechanically Ventilated Status Asthmaticus & Auto-PEEP Engine

"""
Cookbook 368: Offline Clinical Critical Care Ventilated Status Asthmaticus & Auto-PEEP Engine
OpenPHR Clinical AI Working Group (https://openphr.org)
"""

import math
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass

@dataclass
class VentilatedAsthmaTelemetry:
    patient_id: str
    age_years: float = 32.0
    sex: str = "Female"
    height_cm: float = 165.0
    # Ventilator Settings
    vent_mode: str = "Volume Assist-Control (V-A/C)"
    set_tidal_volume_ml: float = 480.0
    set_respiratory_rate_bpm: float = 18.0 # Too high for severe asthma! (T_e too short)
    set_peak_flow_rate_l_min: float = 55.0 # Low flow -> prolonged T_i -> shortened T_e
    set_peep_cmh2o: float = 5.0
    # Airway Pressures (from Hold Maneuvers)
    peak_inspiratory_pressure_cmh2o: float = 54.0 # High resistive load
    plateau_pressure_cmh2o: float = 28.0 # Safe if < 30 cmH2O
    total_peep_measured_cmh2o: float = 17.5 # Measured during End-Expiratory Hold!
    expiratory_flow_returns_to_baseline: bool = False # Confirms Dynamic Hyperinflation on screen
    # Arterial Blood Gas (ABG) & Hemodynamics
    arterial_ph: float = 7.22 # Permissive hypercapnia window
    arterial_pco2_mmhg: float = 68.0
    arterial_pao2_mmhg: float = 94.0
    systolic_bp_mmhg: float = 88.0 # Borderline hypotension due to decreased venous return
    mean_arterial_pressure_mmhg: float = 60.0
    is_deeply_sedated_and_paralyzed: bool = True

@dataclass
class AsthmaVentEvaluationReport:
    patient_id: str
    predicted_body_weight_kg: float
    auto_peep_cmh2o: float # 12.5 cmH2O (Severe)
    airway_resistance_cmh2o_l_s: float
    static_compliance_ml_cmh2o: float
    dynamic_hyperinflation_tier: str # "CRITICAL DYNAMIC HYPERINFLATION (Auto-PEEP >= 10 cmH2O)"
    ventilator_optimization_prescription: List[str]
    permissive_hypercapnia_status: str
    safety_sentinels: List[str]
    clinical_ats_gina_directive: str

class VentilatedAsthmaDecisionEngine:
    """
    Offline clinical engine for Auto-PEEP quantification, airway resistance dissection,
    ventilator timing optimization (I:E ratio maximization), and apnea rescue gating.
    """

    def calculate_pbw(self, height_cm: float, sex: str) -> float:
        height_inches = height_cm / 2.54
        inches_over_5ft = max(0.0, height_inches - 60.0)

        if sex.lower() == "male":
            pbw = 50.0 + 2.3 * inches_over_5ft
        else:
            pbw = 45.5 + 2.3 * inches_over_5ft
        return round(pbw, 1)

    def dissect_respiratory_mechanics(self, d: VentilatedAsthmaTelemetry) -> Tuple[float, float, float, str]:
        # Auto-PEEP = Total PEEP - Set PEEP
        auto_peep = max(0.0, d.total_peep_measured_cmh2o - d.set_peep_cmh2o)

        # Flow in L/s
        flow_l_s = d.set_peak_flow_rate_l_min / 60.0

        # Airway Resistance Raw = (Ppeak - Pplat) / Flow
        raw = (d.peak_inspiratory_pressure_cmh2o - d.plateau_pressure_cmh2o) / max(0.1, flow_l_s)

        # Static Compliance Crs = Vt / (Pplat - Total PEEP)
        driving_pressure = max(1.0, d.plateau_pressure_cmh2o - d.total_peep_measured_cmh2o)
        crs = d.set_tidal_volume_ml / driving_pressure

        if auto_peep >= 10.0:
            tier = f"CRITICAL DYNAMIC HYPERINFLATION (Auto-PEEP {auto_peep:.1f} cmH2O >= 10 cmH2O - Extreme Hemodynamic & Barotrauma Hazard)"
        elif auto_peep >= 5.0:
            tier = f"MODERATE DYNAMIC HYPERINFLATION (Auto-PEEP {auto_peep:.1f} cmH2O)"
        else:
            tier = f"MILD / ACCEPTABLE AUTO-PEEP ({auto_peep:.1f} cmH2O)"

        return round(auto_peep, 1), round(raw, 1), round(crs, 1), tier

    def optimize_ventilator_timing(self, d: VentilatedAsthmaTelemetry, pbw: float, auto_peep: float) -> List[str]:
        orders = []

        # Target Vt: 6-7 mL/kg PBW
        target_vt = round(pbw * 6.5, 0)
        # Target RR: 10-12 bpm
        target_rr = 10.0
        # Target Flow: 75-85 L/min square wave
        target_flow = 80.0

        # Calculated Insp time: Vt (L) / Flow (L/s)
        t_i = (target_vt / 1000.0) / (target_flow / 60.0)
        # Total cycle time: 60 / RR
        t_tot = 60.0 / target_rr
        t_e = t_tot - t_i
        ie_ratio = t_e / t_i

        orders.append("1. VENTILATOR RE-CONFIGURATION PRESCRIPTION (To Eliminate Dynamic Hyperinflation):")
        orders.append(f"   • DECREASE RESPIRATORY RATE: Lower Set Rate from {d.set_respiratory_rate_bpm:.0f} bpm to {target_rr:.0f} bpm (Expands total cycle time to {t_tot:.1f}s).")
        orders.append(f"   • INCREASE PEAK FLOW RATE: Elevate Square Flow from {d.set_peak_flow_rate_l_min:.0f} L/min to {target_flow:.0f} L/min (Shortens T_i to {t_i:.2f}s).")
        orders.append(f"   • ADJUST TIDAL VOLUME: Set Vt to {target_vt:.0f} mL ({target_vt / pbw:.1f} mL/kg PBW).")
        orders.append(f"   • RESULTING TIMING: Expiratory Time (T_e) = {t_e:.2f}s -> Resulting I:E Ratio = 1:{ie_ratio:.1f} (Optimal >= 1:4).")
        orders.append(f"   • SET PEEP: Maintain Set PEEP at 0 - 5 cmH2O (Current: {d.set_peep_cmh2o:.0f} cmH2O). DO NOT INCREASE SET PEEP.")

        return orders

    def evaluate_permissive_hypercapnia_and_safety(self, d: VentilatedAsthmaTelemetry, auto_peep: float) -> Tuple[str, List[str]]:
        sentinels = []

        ph = d.arterial_ph
        pco2 = d.arterial_pco2_mmhg

        if ph >= 7.20:
            status = f"ACCEPTABLE PERMISSIVE HYPERCAPNIA (pH {ph:.2f} >= 7.20, pCO2 {pco2:.1f} mmHg). Do not increase ventilator rate to normalize pCO2!"
        elif 7.15 <= ph < 7.20:
            status = f"MODERATE RESPIRATORY ACIDEMIA (pH {ph:.2f}, pCO2 {pco2:.1f} mmHg). Consider slow IV Sodium Bicarbonate buffer (1-2 ampules) if hemodynamic instability ensues."
        else:
            status = f"SEVERE REFRACTORY ACIDEMIA (pH {ph:.2f} < 7.15). Administer IV Sodium Bicarbonate / THAM infusion; evaluate for inhaled Sevoflurane / Heliox rescue."

        # Plateau Pressure Sentinel
        if d.plateau_pressure_cmh2o >= 30.0:
            sentinels.append(f"🚨 ALVEOLAR OVERDISTENSION HAZARD (Pplat {d.plateau_pressure_cmh2o:.1f} cmH2O >= 30): High risk of barotrauma, pneumothorax, and pneumomediastinum. Reduce Tidal Volume by 1 mL/kg PBW immediately!")

        # Disconnect the Vent Rescue Protocol Sentinel
        sentinels.append("🚨 'DISCONNECT THE VENTILATOR' APNEA RESCUE PROTOCOL: If the patient develops sudden severe hypotension, bradycardia, or PEA cardiac arrest: (1) IMMEDIATELY DISCONNECT PATIENT FROM VENTILATOR; (2) Allow 30-60 seconds of passive apnea with manual chest compression to decompress trapped air; (3) If BP recovers, arrest was dynamic hyperinflation; if hypotension persists, perform STAT bilateral needle/finger thoracostomy for tension pneumothorax!")

        return status, sentinels

    def evaluate_case(self, data: VentilatedAsthmaTelemetry) -> AsthmaVentEvaluationReport:
        pbw = self.calculate_pbw(data.height_cm, data.sex)
        auto_peep, raw, crs, tier = self.dissect_respiratory_mechanics(data)
        vent_orders = self.optimize_ventilator_timing(data, pbw, auto_peep)
        hcap_status, sentinels = self.evaluate_permissive_hypercapnia_and_safety(data, auto_peep)

        directives = []
        directives.append(f"DHI STATUS: {tier}.")
        directives.append(f"MECHANICS: Raw {raw:.1f} cmH2O/L/s, Crs {crs:.1f} mL/cmH2O, Pplat {data.plateau_pressure_cmh2o:.1f} cmH2O.")
        directives.append(f"VENT GOAL: Rate 10 bpm, Flow 80 L/min (I:E >= 1:4).")
        directives.append(f"HYPERCAPNIA: {hcap_status}.")

        return AsthmaVentEvaluationReport(
            patient_id=data.patient_id,
            predicted_body_weight_kg=pbw,
            auto_peep_cmh2o=auto_peep,
            airway_resistance_cmh2o_l_s=raw,
            static_compliance_ml_cmh2o=crs,
            dynamic_hyperinflation_tier=tier,
            ventilator_optimization_prescription=vent_orders,
            permissive_hypercapnia_status=hcap_status,
            safety_sentinels=sentinels,
            clinical_ats_gina_directive=" ".join(directives)
        )

# Example Execution & Verification
if __name__ == "__main__":
    engine = VentilatedAsthmaDecisionEngine()

    print("=" * 80)
    print("OpenPHR Clinical Critical Care Mechanically Ventilated Status Asthmaticus Engine")
    print("=" * 80)

    # Test Case 1: 32-year-old female (165 cm, PBW 57.0 kg) intubated for near-fatal status asthmaticus.
    # Vent: Rate 18 bpm, Flow 55 L/min, Vt 480 mL.
    # Mechanics: Ppeak 54 cmH2O, Pplat 28 cmH2O, Total PEEP 17.5 cmH2O (Set PEEP 5.0 cmH2O).
    # Dissection: Auto-PEEP = 12.5 cmH2O (Critical DHI!), Raw = 31.6 cmH2O/L/s (Severe Bronchospasm).
    # ABG: pH 7.22, pCO2 68 mmHg (Acceptable Permissive Hypercapnia).
    # Triage: Reduce RR to 10 bpm + Increase Flow to 80 L/min -> Expands I:E to 1:4.8 + Disconnect-the-Vent Sentinel!
    sav1 = VentilatedAsthmaTelemetry(
        patient_id="MICU-SAV-4401",
        age_years=32.0,
        sex="Female",
        height_cm=165.0,
        vent_mode="Volume Assist-Control (V-A/C)",
        set_tidal_volume_ml=480.0,
        set_respiratory_rate_bpm=18.0,
        set_peak_flow_rate_l_min=55.0,
        set_peep_cmh2o=5.0,
        peak_inspiratory_pressure_cmh2o=54.0,
        plateau_pressure_cmh2o=28.0,
        total_peep_measured_cmh2o=17.5,
        expiratory_flow_returns_to_baseline=False,
        arterial_ph=7.22,
        arterial_pco2_mmhg=68.0,
        arterial_pao2_mmhg=94.0,
        systolic_bp_mmhg=88.0,
        mean_arterial_pressure_mmhg=60.0,
        is_deeply_sedated_and_paralyzed=True
    )

    rep1 = engine.evaluate_case(sav1)

    print(f"\n[Patient {rep1.patient_id} - Ventilated Asthma Assessment]")
    print(f"Predicted Body Weight (PBW): {rep1.predicted_body_weight_kg:.1f} kg")
    print(f"Auto-PEEP: {rep1.auto_peep_cmh2o:.1f} cmH2O")
    print(f"Airway Resistance (Raw): {rep1.airway_resistance_cmh2o_l_s:.1f} cmH2O/L/s (Normal < 5.0)")
    print(f"Static Compliance (Crs): {rep1.static_compliance_ml_cmh2o:.1f} mL/cmH2O")
    print(f"Dynamic Hyperinflation Status:\n  {rep1.dynamic_hyperinflation_tier}")
    print("\nVentilator Optimization Prescription:")
    for o in rep1.ventilator_optimization_prescription:
        print(f"  {o}")
    print(f"\nPermissive Hypercapnia Protocol:\n  {rep1.permissive_hypercapnia_status}")
    if rep1.safety_sentinels:
        print("\nSafety Sentinels:")
        for s in rep1.safety_sentinels:
            print(f"  🚨 {s}")
    print(f"\nATS / GINA Consensus Directive:\n{rep1.clinical_ats_gina_directive}")