Cookbook 364: Offline Clinical Cellular Immunotherapy CAR-T CRS & ICANS ASTCT Titrator Engine

"""
Cookbook 364: Offline Cellular Immunotherapy CAR-T CRS & ICANS ASTCT Titrator 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 CARTCellTelemetry:
    patient_id: str
    age_years: float = 61.0
    patient_weight_kg: float = 78.0
    car_t_construct: str = "Axicabtagene ciloleucel (Axi-cel)"
    days_post_infusion: int = 5
    # Vital Signs & Hemodynamics
    temperature_celsius: float = 39.2 # >= 38.0 C = Fever (Prerequisite for CRS)
    systolic_bp_mmhg: float = 84.0
    mean_arterial_pressure_mmhg: float = 58.0
    fluid_boluses_received_ml: float = 2000.0
    vasopressor_count: int = 1 # 1 single pressor (Norepinephrine 0.08 mcg/kg/min) -> Grade 3 CRS!
    # Respiratory Telemetry
    spo2_percent: float = 93.0
    oxygen_delivery_modality: str = "High-Flow Nasal Cannula (FiO2 50% at 35 L/min)" # "Room Air", "Low-Flow NC (<=6L)", "High-Flow NC / Venturi", "Positive Pressure (BiPAP/Intubated)"
    # ICE Score Components (Total 0 - 10)
    ice_orientation_score: int = 2 # 2/4 (Oriented to year & city; missed month & hospital)
    ice_naming_score: int = 2 # 2/3 (Named pen, watch; missed tie)
    ice_following_commands_score: int = 1 # 1/1 (Followed 1-step command)
    ice_writing_score: int = 0 # 0/1 (Moderate-severe dysgraphia; unable to write full sentence)
    ice_calculation_score: int = 0 # 0/1 (Unable to serial count backwards)
    # Total ICE = 2 + 2 + 1 + 0 + 0 = 5 / 10 -> Grade 2 ICANS!
    # Neurological Exam & Seizure Status
    has_seizures: bool = False
    has_cerebral_edema_or_coma: bool = False
    is_on_levetiracetam_prophylaxis: bool = True
    # Hyperinflammatory Biomarkers
    serum_ferritin_ng_ml: float = 8450.0 # Extreme elevation
    c_reactive_protein_mg_l: float = 185.0
    serum_fibrinogen_mg_dl: float = 195.0 # Watch for < 150 mg/dL (sHLH/MAS risk)

@dataclass
class CARTEvaluationReport:
    patient_id: str
    astct_crs_grade: str # "GRADE 3 CYTOKINE RELEASE SYNDROME (ASTCT 2019)"
    astct_icans_grade: str # "GRADE 2 ICANS (ICE Score 5/10)"
    ice_total_score: int # 5 / 10
    tocilizumab_and_steroid_orders: List[str]
    anakinra_and_cellular_rescue: List[str]
    safety_sentinels: List[str]
    clinical_astct_nccn_directive: str

class CARTCellToxicologyDecisionEngine:
    """
    Offline clinical engine for ASTCT 2019 CRS & ICANS consensus staging,
    ICE cognitive evaluation, Tocilizumab/Dexamethasone titration, and MAS/sHLH sentinels.
    """

    def calculate_ice_score(self, d: CARTCellTelemetry) -> int:
        score = (
            min(4, max(0, d.ice_orientation_score)) +
            min(3, max(0, d.ice_naming_score)) +
            min(1, max(0, d.ice_following_commands_score)) +
            min(1, max(0, d.ice_writing_score)) +
            min(1, max(0, d.ice_calculation_score))
        )
        return score

    def stage_astct_crs(self, d: CARTCellTelemetry) -> Tuple[str, str]:
        has_fever = d.temperature_celsius >= 38.0

        if not has_fever:
            return "NO CRS (Temperature < 38.0 C)", "Afebrile; does not meet ASTCT prerequisite for Cytokine Release Syndrome."

        # Fever Present -> Evaluate Hypotension and Hypoxia
        is_ppv = "positive pressure" in d.oxygen_delivery_modality.lower() or "bipap" in d.oxygen_delivery_modality.lower() or "intubated" in d.oxygen_delivery_modality.lower()
        is_high_flow = "high-flow" in d.oxygen_delivery_modality.lower() or "venturi" in d.oxygen_delivery_modality.lower() or "non-rebreather" in d.oxygen_delivery_modality.lower()
        is_low_flow = "low-flow" in d.oxygen_delivery_modality.lower()

        if d.vasopressor_count >= 2 or is_ppv:
            return "GRADE 4 CYTOKINE RELEASE SYNDROME (ASTCT 2019)", "Life-threatening CRS requiring multiple vasopressors and/or positive pressure ventilation (BiPAP/Intubation)."
        elif d.vasopressor_count == 1 or is_high_flow:
            return "GRADE 3 CYTOKINE RELEASE SYNDROME (ASTCT 2019)", "Severe CRS requiring high-flow oxygen delivery and/or single vasopressor infusion."
        elif (d.mean_arterial_pressure_mmhg < 65.0 or d.systolic_bp_mmhg < 90.0 or d.fluid_boluses_received_ml >= 1000.0) or is_low_flow:
            return "GRADE 2 CYTOKINE RELEASE SYNDROME (ASTCT 2019)", "Moderate CRS with fluid-responsive hypotension and/or low-flow oxygen requirements."
        else:
            return "GRADE 1 CYTOKINE RELEASE SYNDROME (ASTCT 2019)", "Mild CRS characterized by fever alone without hypotension or hypoxia."

    def stage_astct_icans(self, d: CARTCellTelemetry, ice_score: int) -> Tuple[str, str]:
        if d.has_cerebral_edema_or_coma or (ice_score == 0 and "unarousable" in d.car_t_construct.lower()):
            return "GRADE 4 ICANS (ASTCT 2019)", "Critical life-threatening neurotoxicity with coma, stupor, or diffuse cerebral edema."
        elif ice_score <= 2 or d.has_seizures:
            return "GRADE 3 ICANS (ASTCT 2019)", "Severe neurotoxicity (ICE 0-2, seizure activity, or elevated intracranial pressure)."
        elif 3 <= ice_score <= 6:
            return "GRADE 2 ICANS (ASTCT 2019)", "Moderate neurotoxicity with intermediate cognitive impairment (ICE 3-6) and expressive dysphasia."
        elif 7 <= ice_score <= 9 or d.ice_writing_score == 0:
            return "GRADE 1 ICANS (ASTCT 2019)", "Mild neurocognitive impairment (ICE 7-9) or isolated early dysgraphia."
        else:
            return "NO ICANS (ICE Score 10/10)", "Normal neurological examination and preserved cognitive orientation."

    def generate_immunotherapy_plan(self, d: CARTCellTelemetry, crs_grade: str, icans_grade: str) -> Tuple[List[str], List[str], List[str]]:
        toci_orders = []
        rescue_orders = []
        sentinels = []

        wt = d.patient_weight_kg
        toci_dose = min(800.0, wt * 8.0)

        # Isolated ICANS Sentinel
        is_crs_present = "NO CRS" not in crs_grade
        is_icans_significant = "GRADE 2" in icans_grade or "GRADE 3" in icans_grade or "GRADE 4" in icans_grade

        if not is_crs_present and is_icans_significant:
            sentinels.append("🚨 ISOLATED ICANS TOCILIZUMAB PARADOXICAL CNS HAZARD: In patients with isolated ICANS without concurrent CRS, Tocilizumab blocks peripheral IL-6 receptors, resulting in an acute transient surge of circulating free IL-6 across the blood-brain barrier, worsening neuro-inflammation and cerebral edema! FIRST-LINE THERAPY IS DEXAMETHASONE, NOT TOCILIZUMAB ALONE!")

        # CRS-Driven Tocilizumab Orders
        if "GRADE 2" in crs_grade or "GRADE 3" in crs_grade or "GRADE 4" in crs_grade:
            toci_orders.append(f"1. STAT TOCILIZUMAB (Anti-IL-6R): Administer {toci_dose:.0f} mg IV ({wt * 8.0:.0f} mg at 8.0 mg/kg, max 800 mg) infused over 60 minutes.")
            toci_orders.append("   • Repeat q8h PRN if no clinical response (Maximum 3-4 doses per 24-hour period).")

        # Corticosteroid Gating
        if "GRADE 4" in crs_grade or "GRADE 4" in icans_grade:
            toci_orders.append("2. STAT HIGH-DOSE METHYLPREDNISOLONE: Administer 1,000 mg IV daily x 3 days, followed by rapid 4-day taper.")
            rescue_orders.append("   • STAT ICU Transfer & Continuous EEG: Order STAT head CT/MRI to assess for diffuse cerebral edema.")
        elif "GRADE 3" in crs_grade or "GRADE 3" in icans_grade:
            toci_orders.append("2. STAT HIGH-DOSE DEXAMETHASONE: Administer 20 mg IV q6h.")
        elif "GRADE 2" in crs_grade or "GRADE 2" in icans_grade:
            toci_orders.append("2. DEXAMETHASONE: Administer 10 mg IV q6h until de-escalation to Grade 1.")

        # Refractory / Anakinra Add-On
        if "GRADE 3" in crs_grade or "GRADE 4" in crs_grade or "GRADE 3" in icans_grade:
            rescue_orders.append("3. REFRACTORY RESCUE (ANAKINRA - Anti-IL-1R): Initiate Anakinra 100-200 mg SC/IV q6-12h (or 8 mg/kg/day) for steroid/tocilizumab-refractory hyper-inflammation.")
            rescue_orders.append("4. SECOND-LINE ALTERNATIVE: Siltuximab 11 mg/kg IV (Direct IL-6 neutralization) if no response to Tocilizumab.")

        # Secondary MAS / sHLH Sentinel
        if d.serum_ferritin_ng_ml >= 10000.0 or d.serum_fibrinogen_mg_dl < 150.0:
            sentinels.append(f"🚨 SECONDARY MAS / sHLH ALERT (Ferritin {d.serum_ferritin_ng_ml:.0f} ng/mL, Fibrinogen {d.serum_fibrinogen_mg_dl:.0f} mg/dL): High risk of Macrophage Activation Syndrome / secondary HLH. Check fasting triglycerides, soluble CD25 (sIL-2R), and bone marrow aspirate. Early high-dose Anakinra indicated!")

        # Seizure Prophylaxis Sentinel
        if not d.is_on_levetiracetam_prophylaxis:
            sentinels.append("🚨 SEIZURE PROPHYLAXIS MANDATE: Patient is not currently receiving Levetiracetam. Order STAT Levetiracetam 750 mg PO/IV BID through Day +30 post CAR-T infusion!")

        return toci_orders, rescue_orders, sentinels

    def evaluate_case(self, data: CARTCellTelemetry) -> CARTEvaluationReport:
        ice = self.calculate_ice_score(data)
        crs_grade, crs_desc = self.stage_astct_crs(data)
        icans_grade, icans_desc = self.stage_astct_icans(data, ice)
        toci_orders, rescue_orders, sentinels = self.generate_immunotherapy_plan(data, crs_grade, icans_grade)

        directives = []
        directives.append(f"CRS: {crs_grade}.")
        directives.append(f"ICANS: {icans_grade} (ICE Score {ice}/10).")
        if toci_orders: directives.append(f"IMMUNOSUPPRESSION: {toci_orders[0]}.")
        directives.append("MONITORING: Oncology ICU telemetry, serial ICE scores q4h, and daily ferritin/CRP/fibrinogen.")

        return CARTEvaluationReport(
            patient_id=data.patient_id,
            astct_crs_grade=crs_grade,
            astct_icans_grade=icans_grade,
            ice_total_score=ice,
            tocilizumab_and_steroid_orders=toci_orders,
            anakinra_and_cellular_rescue=rescue_orders,
            safety_sentinels=sentinels,
            clinical_astct_nccn_directive=" ".join(directives)
        )

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

    print("=" * 80)
    print("OpenPHR Clinical Cellular Immunotherapy CAR-T CRS & ICANS ASTCT Engine")
    print("=" * 80)

    # Test Case 1: 61-year-old male Day +5 post Axi-cel for R/R DLBCL.
    # Telemetry: Temp 39.2 C, MAP 58 mmHg requiring Norepinephrine (1 pressor) + High-Flow NC (FiO2 50%).
    # Neuro: ICE Score 5/10 (Orientation 2, Naming 2, Command 1, Writing 0, Calc 0).
    # Staging: ASTCT Grade 3 CRS (Single Pressor + High-Flow O2) & ASTCT Grade 2 ICANS!
    # Triage: STAT Tocilizumab 624 mg IV (8 mg/kg) + Dexamethasone 20 mg IV q6h + Anakinra Standby!
    # Sentinels: MAS/sHLH monitoring + Seizure Prophylaxis!
    cart1 = CARTCellTelemetry(
        patient_id="CART-AXI-5501",
        age_years=61.0,
        patient_weight_kg=78.0,
        car_t_construct="Axicabtagene ciloleucel (Axi-cel)",
        days_post_infusion=5,
        temperature_celsius=39.2,
        systolic_bp_mmhg=84.0,
        mean_arterial_pressure_mmhg=58.0,
        fluid_boluses_received_ml=2000.0,
        vasopressor_count=1,
        spo2_percent=93.0,
        oxygen_delivery_modality="High-Flow Nasal Cannula (FiO2 50% at 35 L/min)",
        ice_orientation_score=2,
        ice_naming_score=2,
        ice_following_commands_score=1,
        ice_writing_score=0,
        ice_calculation_score=0,
        has_seizures=False,
        has_cerebral_edema_or_coma=False,
        is_on_levetiracetam_prophylaxis=True,
        serum_ferritin_ng_ml=8450.0,
        c_reactive_protein_mg_l=185.0,
        serum_fibrinogen_mg_dl=195.0
    )

    rep1 = engine.evaluate_case(cart1)

    print(f"\n[Patient {rep1.patient_id} - CAR-T Toxicity Assessment]")
    print(f"ASTCT CRS Grade: {rep1.astct_crs_grade}")
    print(f"ASTCT ICANS Grade: {rep1.astct_icans_grade}")
    print(f"ICE Cognitive Score: {rep1.ice_total_score} / 10")
    print("\nTocilizumab & Corticosteroid Orders:")
    for o in rep1.tocilizumab_and_steroid_orders:
        print(f"  {o}")
    print("\nRefractory Rescue & Escalation Orders:")
    for r in rep1.anakinra_and_cellular_rescue:
        print(f"  {r}")
    if rep1.safety_sentinels:
        print("\nSafety Sentinels:")
        for s in rep1.safety_sentinels:
            print(f"  🚨 {s}")
    print(f"\nASTCT / NCCN Consensus Directive:\n{rep1.clinical_astct_nccn_directive}")