🏥
Healthcare: Patient Safety Monitoring
An ICU monitoring system where causal tracing connects vital sign readings, medication infusions, and clinical alerts into a single accountable event graph.
★ Important
In clinical settings, understanding why an alert fired is as important as the alert itself. PyRapide's causal graph lets clinicians trace any alert back through the exact chain of readings and medication events that triggered it.
Architecture Overview
The ICU monitoring architecture consists of three interface types:
- VitalMonitor emits heart rate, SpO2, and blood pressure readings from bedside devices.
- MedicationPump emits dose start, completion, and halt events from IV infusion pumps.
- ClinicalAlert receives events from both sources and generates clinical decision support alerts.
Interfaces
icu_interfaces.py python
1from pyrapide import interface, action, module, when
2from pyrapide import architecture, connect, Pattern, Engine
3from pyrapide import must_match, never, StreamProcessor
4import asyncio
5
6# ── Interfaces ──────────────────────────────────────────
7
8@interface
9class VitalMonitor:
10 """ICU bedside vital signs monitor."""
11 @action
12 async def heart_rate(self, bpm: int, patient_id: str) -> None: ...
13 @action
14 async def spo2(self, level: float, patient_id: str) -> None: ...
15 @action
16 async def blood_pressure(self, systolic: int, diastolic: int, patient_id: str) -> None: ...
17
18@interface
19class MedicationPump:
20 """IV medication infusion pump."""
21 @action
22 async def dose_started(self, drug: str, rate_ml_hr: float, patient_id: str) -> None: ...
23 @action
24 async def dose_completed(self, drug: str, patient_id: str) -> None: ...
25 @action
26 async def dose_halted(self, drug: str, reason: str, patient_id: str) -> None: ...
27
28@interface
29class ClinicalAlert:
30 """Clinical decision support alert system."""
31 @action
32 async def alert(self, message: str, severity: str, patient_id: str) -> None: ...
33 @action
34 async def alert_acknowledged(self, alert_id: str, nurse_id: str) -> None: ...
35 @action
36 async def intervention(self, action: str, patient_id: str) -> None: ...
Module Logic
The ICUAlertSystem module implements the ClinicalAlert interface and reacts to vital monitor and medication pump events.
icu_module.py python
1# ── Modules (behavioral logic) ──────────────────────────
2
3@module(implements=ClinicalAlert)
4class ICUAlertSystem:
5 @when("VitalMonitor.spo2")
6 async def check_hypoxia(self, event):
7 if event.data["level"] < 90.0:
8 await self.alert(
9 message=f"Hypoxia detected: SpO2 {event.data['level']}%",
10 severity="critical",
11 patient_id=event.data["patient_id"]
12 )
13
14 @when("MedicationPump.dose_started")
15 async def check_drug_interaction(self, event):
16 # Cross-reference active medications causally
17 await self.alert(
18 message=f"Verify interaction: {event.data['drug']}",
19 severity="warning",
20 patient_id=event.data["patient_id"]
21 )
Constraints
The constraints encode clinical safety rules directly into the architecture:
- hypoxia_alert: a SpO2 reading below 90% must cause a clinical alert (patient safety requirement).
- double_dose_prevention: two infusion starts for the same drug on the same patient must never be concurrent (medication safety).
- alert_must_be_acked: every clinical alert must be acknowledged by a nurse (accountability).
icu_architecture.py python
1# ── Architecture with Constraints ───────────────────────
2
3@architecture
4class ICUMonitoring:
5 vitals: VitalMonitor
6 pump: MedicationPump
7 alerts: ClinicalAlert
8
9 def connections(self):
10 return [
11 connect(Pattern.match("VitalMonitor.*"), "alerts"),
12 connect(Pattern.match("MedicationPump.*"), "alerts"),
13 ]
14
15 def constraints(self):
16 return [
17 # Hypoxia alert: low SpO2 MUST trigger an alert
18 must_match(
19 trigger="VitalMonitor.spo2",
20 response="ClinicalAlert.alert",
21 condition=lambda e: e.data["level"] < 90.0,
22 name="hypoxia_alert"
23 ),
24
25 # Double-dose prevention: two dose_started events
26 # for the same drug must NEVER happen concurrently
27 never(
28 pattern=("MedicationPump.dose_started",
29 "MedicationPump.dose_started"),
30 condition=lambda e1, e2: (
31 e1.data["drug"] == e2.data["drug"]
32 and e1.data["patient_id"] == e2.data["patient_id"]
33 ),
34 name="double_dose_prevention"
35 ),
36
37 # Every alert must be acknowledged
38 must_match(
39 trigger="ClinicalAlert.alert",
40 response="ClinicalAlert.alert_acknowledged",
41 name="alert_must_be_acked"
42 ),
43 ]
Execution and Analysis
Run the architecture and trace alert causality:
icu_run.py python
1# ── Execute and Analyze ─────────────────────────────────
2
3async def main():
4 engine = Engine()
5 computation = await engine.run(ICUMonitoring)
6
7 # Trace why a specific alert fired
8 from pyrapide import root_causes, backward_slice
9 alerts = [e for e in computation.events if e.name == "ClinicalAlert.alert"]
10
11 for alert in alerts:
12 causes = root_causes(computation, alert)
13 print(f"Alert: {alert.data['message']}")
14 print(f" Root causes: {[c.name for c in causes]}")
15
16 # Full causal chain leading to this alert
17 chain = backward_slice(computation, alert)
18 print(f" Causal chain depth: {len(chain)} events")
19
20asyncio.run(main())
💡 Tip
Use
forward_slice to answer "what happened because of this medication dose?" and backward_slice to answer "what caused this alert to fire?" These are the questions clinical teams need answered during incident review.