FHIR Vitals Pipeline

Bedside → Broker → Brain → Board.

A production-grade event-streaming platform for hospital wards: ingests HL7 FHIR R4 Observation telemetry, detects sustained clinical anomalies with a sliding-window rules engine, persists time-series vitals in a TimescaleDB hypertable, and fans out CRITICAL alerts to clinician dashboards over WebSocket — with zero-loss Kafka delivery guarantees.

---

📌 Why this exists

Ward monitors emit a firehose of vital-sign readings. Clinicians don't need data — they need the three numbers that mean "come to bed 7, now." This pipeline turns raw device chatter into exactly that:

• 🩺 FHIR R4 ingestion — every payload is validated as a real Observation resource (LOINC-coded, clinically plausible ranges). Garbage never enters the stream: rejected with a FHIR OperationOutcome 422.
• ⚡ Zero-loss streaming — Kafka acks=all, partition-by-patient_id ordering, 3-attempt exponential retry, plus an in-memory local broker for offline dev (USE_LOCAL_BROKER=true).
• 🧠 Sustained-anomaly detection — a 120 s sliding window per patient/metric fires rules only on contiguous breaches (Tachycardia, Bradycardia, Critical Desaturation), with 60 s anti-flapping cooldowns that are bypassed when severity escalates. Alert fatigue is a bug, not a feature.
• 📡 Push, not poll — a dedicated consumer group (healthcare-ws-alert-dispatcher) pumps alerts into a WebSocket hub; dashboards subscribe per ward, per patient, or hospital-wide.
• 🗄️ Audit-first persistence — vitals land in a time-chunked hypertable via a 500 ms / 1 000-row batch writer; every alert is an auditable row with an acknowledgment workflow.

---

🌐 System Architecture

🔍 Interactive version: open docs/architecture.html in any browser — pan/zoom, dark/light themes, PATH & LENS focus mode. Source JSON (validated, showcase profile): docs/archify/architecture.candidate.json.

Component breakdown

Trust & failure boundaries

• Edge → Gateway is untrusted input. Only fully validated FHIR resources are ever produced to Kafka — invalid payloads terminate as 422 OperationOutcome and never touch the stream.
• Gateway → Broker is the durability seam. acks=all + retry means a 202 ACCEPTED implies the record is committed in Kafka, not just accepted into process memory.
• Broker + DB are one private network (Docker Compose). Application services hold the only connections; nothing else is exposed except the gateway's HTTP/WS port.
• Processor is stateless-able: Kafka consumer-group offsets are the replay mechanism; the in-memory window is rebuilt from the stream on restart.

---

🔄 Core Workflow: one observation's journey

🔍 Interactive version: docs/sequence.html · source: docs/archify/sequence.candidate.json

1. Monitor → Gateway — POST /api/v1/fhir/Observation (single, alias route, or /batch array).
2. Validate — Pydantic checks resourceType, LOINC code against the six monitored vitals, and clinically plausible ranges (e.g. HR ∈ [0, 350] bpm). Failure ⇒ immediate 422 OperationOutcome.
3. Produce — the converted internal record is sent to patient.vitals.raw with acks=all, keyed by patient_id (per-patient ordering), 3-attempt exponential backoff before surfacing an error.
4. Acknowledge to device — 202 ACCEPTED { ingested_count, patient_id }.
5. Consume & store — the Stream Processor pulls batches (≤ 100), and the batch writer flushes inserts to the hypertable every 500 ms or 1 000 rows.
6. Evaluate — each record updates its patient's 120 s sliding window; a rule fires only when a threshold breach is contiguous and sustained (see table below).
7. Alert — on trigger: audit row → clinical_alerts, and a message on patient.alerts.critical, subject to a 60 s anti-flapping cooldown per (patient, rule) — escalated severity breaks through the cooldown.
8. Push — the dispatcher consumer group receives the alert and broadcasts it to all subscribed ward/patient/ global WebSocket clients.
9. Close the loop — a nurse acknowledges: POST /api/v1/alerts/{alert_id}/ack updates the audit trail; unacknowledged alerts remain queryable via GET /api/v1/alerts/unacknowledged so nothing is lost across reconnects.

Clinical rules

Supported vitals: heart rate, SpO₂, systolic/diastolic BP, body temperature, respiratory rate (internal/domain/fhir.py — LOINCCode, LOINC_RANGES).

---

🧰 Tech Stack

---

📁 Project Structure

.
├── cmd/                        # Service entrypoints (process boundaries)
│   ├── ingestion/              #   FastAPI gateway + WebSocket dispatcher
│   └── processor/              #   Stream processor (rules engine loop)
├── internal/                   # Shared application core
│   ├── domain/                 #   FHIR R4, Observation, ClinicalAlert models
│   ├── engine/                 #   Sliding window + clinical rule evaluation
│   ├── repository/             #   asyncpg pool, batch writer, alert repository
│   ├── stream/                 #   Resilient Kafka producer/consumer + local broker
│   └── transport/              #   REST routes + WebSocket hub/routes
├── migrations/                 # Versioned SQL (hypertable, indexes, audit log)
├── simulator/                  # ICU patient-bed telemetry generator
├── scripts/                    # Migration runner
├── tests/                      # pytest suite (5 files, asyncio mode)
├── docs/                       # Diagrams & documentation
│   ├── architecture.html       #   Interactive Archify architecture
│   ├── sequence.html           #   Interactive Archify sequence
│   ├── archify/                #   Validated diagram source JSON (regenerable)
│   └── assets/                 #   Static PNG exports for this README
├── docker-compose.yml          # Redpanda + TimescaleDB
├── Makefile                    # up / down / migrate / test / simulate / clean
└── pyproject.toml              # Project metadata + pytest config

---

🚀 Quick Start

Prerequisites: Python ≥ 3.11 · Docker Engine · (optional) make

1. Clone & install

git clone https://github.com/<your-org>/healthcare-fhir-vitals-pipeline.git
cd healthcare-fhir-vitals-pipeline

python -m venv .venv
.venv\Scripts\Activate.ps1          # Windows PowerShell
# source .venv/bin/activate         # macOS / Linux

pip install -r requirements.txt

2. Start infrastructure & schema

docker compose up -d                # Redpanda :9092 + TimescaleDB :5432
python scripts/migrate.py           # hypertable + audit log + indexes

Environment overrides (all have sane defaults matching docker-compose.yml; export them in your shell):

# DATABASE_URL=postgresql://postgres:postgres@localhost:5432/healthcare_vitals
# KAFKA_BOOTSTRAP_SERVERS=localhost:9092
# PORT=8000
# KAFKA_MOCK_MODE=false    # true → run without Kafka/DB (dev, tests)
# USE_LOCAL_BROKER=false   # true → in-process broker shim

3. Run the services (two terminals)

# Terminal 1 — ingestion API + WebSocket dispatcher
python -m cmd.ingestion.main

# Terminal 2 — stream processor + clinical rules engine
python -m cmd.processor.main

4. Verify & generate load

curl http://localhost:8000/healthz
# {"status":"healthy","service":"telemetry-and-alert-dispatcher","active_ws_clients":0}

python simulator/patient_simulator.py      # streams ICU beds, induces anomalies

Watch alerts land live:

wscat -c ws://localhost:8000/ws/v1/alerts      # hospital-wide
wscat -c ws://localhost:8000/ws/v1/wards/ICU-1/alerts

Or ingest a single observation manually:

curl -X POST http://localhost:8000/api/v1/fhir/Observation \
  -H "Content-Type: application/json" -d '{
  "resourceType": "Observation",
  "status": "final",
  "code": { "coding": [ { "system": "http://loinc.org", "code": "8867-4", "display": "Heart rate" } ] },
  "subject": { "reference": "Patient/example-001" },
  "effectiveDateTime": "2026-01-01T12:00:00Z",
  "valueQuantity": { "value": 142, "unit": "bpm" }
}'

💡 Makefile shortcuts (paths assume the Windows .venv/Scripts layout): make up · make down · make migrate · make test · make simulate · make clean

⚡ Zero-infra demo (no Docker, no Kafka, no database)

The whole pipeline also runs in a single process using an in-memory broker shim and mock repositories — ideal for a first look or a recruiter sanity check:

python simulate.py --beds 20 --rate 1.0 --duration 60

This spins up the gateway, stream processor, 20 simulated ICU beds (beds 3/7/12 induce anomalies) and a live WebSocket alert console — end to end.

---

📡 API & Event Reference

---

🧪 Testing

pytest -v tests/            # or: make test

Covers: FHIR validation (test_fhir_validation.py), ingestion API (test_ingestion_api.py), sliding-window engine (test_sliding_window.py), alert dispatch + WebSocket delivery (test_alerts_and_websocket.py), and an end-to-end stream integration test (test_stream_integration.py) running on asyncio mode via pyproject.toml.

---

🗺️ Roadmap

• Kubernetes manifests + HA broker/DB topology
• Idempotent upserts + replay-position watermarking
• Grafana dashboard over the hypertable (continuous aggregates)
• Configurable rule packs per unit (NICU vs. general ward)
• OAuth2 / SMART-on-FHIR authorization for ward subscriptions

---

🤝 Contributing

Fork → branch (feat/…, fix/…) → keep pytest -v tests/ green → open a PR describing the clinical or streaming impact. Diagram sources live in docs/archify/*.json — regenerate interactive HTML with Archify: archify finalize architecture docs/archify/architecture.candidate.json docs/architecture.html.

📄 License

Not yet declared — add a LICENSE file before public release.

---

Built for the moment a monitor says what a nurse must hear first. ⭐ Star it if you'd trust it on your ward.