Skip to content

EHR

The EHR modality loads FHIR R4, OMOP and HL7 v2 into a single PatientRecord schema, so everything downstream (timelines, cohorts, guardrails) is source-agnostic.

pip install "openbtk[ehr]"        # fhir.resources, pyarrow, pandas, hl7apy

The example below builds a small FHIR R4 Bundle in the shape Synthea's per-patient export produces (synthetic; no real patient data), then loads it, anchors events to the encounter, selects a cohort and writes the timeline as text.

import contextlib
import io
import json
import pathlib
import tempfile

with contextlib.redirect_stdout(io.StringIO()):  # registry logs, not errors
    from openbtk.data.ehr.cohort import CohortBuilder, has_condition
    from openbtk.data.ehr.fhir import FHIRLoader
    from openbtk.data.ehr.temporal import TemporalNormalizer
    from openbtk.pipelines import PatientTimelineSerializer

bundle = {
    "resourceType": "Bundle",
    "type": "collection",
    "entry": [
        {"resource": {"resourceType": "Patient", "id": "p1", "gender": "female"}},
        {
            "resource": {
                "resourceType": "Encounter",
                "id": "enc-1",
                "status": "finished",
                "class": {"code": "IMP"},
                "subject": {"reference": "Patient/p1"},
                "period": {
                    "start": "2024-03-14T08:00:00Z",
                    "end": "2024-03-16T00:00:00Z",
                },
            }
        },
        {
            "resource": {
                "resourceType": "Condition",
                "id": "cond-1",
                "subject": {"reference": "Patient/p1"},
                "encounter": {"reference": "Encounter/enc-1"},
                "code": {
                    "coding": [
                        {
                            "system": "http://snomed.info/sct",
                            "code": "385093006",
                            "display": "Community-acquired pneumonia",
                        }
                    ]
                },
                "onsetDateTime": "2024-03-14T08:10:00Z",
            }
        },
        {
            "resource": {
                "resourceType": "Observation",
                "id": "obs-1",
                "status": "final",
                "subject": {"reference": "Patient/p1"},
                "encounter": {"reference": "Encounter/enc-1"},
                "code": {
                    "coding": [
                        {
                            "system": "http://loinc.org",
                            "code": "6690-2",
                            "display": "WBC",
                        }
                    ]
                },
                "valueQuantity": {"value": 14.2, "unit": "10*3/uL"},
                "effectiveDateTime": "2024-03-14T08:20:00Z",
                "referenceRange": [{"low": {"value": 4.5}, "high": {"value": 11.0}}],
            }
        },
    ],
}

folder = tempfile.TemporaryDirectory()
pathlib.Path(folder.name, "p1.json").write_text(json.dumps(bundle), encoding="utf-8")

with contextlib.redirect_stdout(io.StringIO()):
    patients = [TemporalNormalizer().process(p) for p in FHIRLoader().load(folder.name)]
    pneumonia = list(CohortBuilder(patients).include(has_condition("385093006")))
    timeline = PatientTimelineSerializer().serialize(pneumonia[0])

assert [p.patient_id for p in pneumonia] == ["p1"]
assert "Community-acquired pneumonia (SNOMED 385093006)" in timeline.text
assert "WBC 14.2 10*3/uL (ref 4.5-11.0) [HIGH]" in timeline.text
folder.cleanup()

What each step does

  • FHIRLoader reads a directory of R4 Bundle JSON files (one patient per file) and validates each resource with fhir.resources. OMOPLoader reads the OMOP CDM v5.4 core tables (person, visit, condition, drug, measurement, procedure, observation) from Parquet files in pyarrow batches. Both yield PatientRecords. OMOP's limits are real: Parquet only (no CSV or SQL), codes come from each table's *_source_value column (the concept vocabulary is not bundled, so you declare which code system a table's source values use), the death table is not read, and because OMOP does not group a patient's rows together, each event table is scanned in full before the first record is yielded, so memory scales with event count rather than batch size.
  • TemporalNormalizer anchors events to their encounter so a timeline has a usable order even where a resource lacks its own timestamp.
  • CohortBuilder composes include / exclude predicates (has_condition, has_medication, has_procedure, age_between, or your own function) over a stream. It never materialises the source.
  • PatientTimelineSerializer turns a record into a normal ClinicalTextRecord. That is the seam between the two modalities: the output goes through the same de-identification, segmentation and chunking as any note.

HL7 v2 messages

Many hospitals still send admissions and results as HL7 v2 messages. HL7v2Loader reads a directory of *.hl7 files (each may hold many messages) and folds the messages about one patient into one PatientRecord:

import contextlib
import io
import pathlib
import tempfile

with contextlib.redirect_stdout(io.StringIO()):
    from openbtk.data.ehr.hl7v2 import HL7v2Loader

messages = [
    r"MSH|^~\&|APP|FAC|RCV|FAC|20240314101500+0000||ADT^A01|M1|P|2.5",
    "PID|1||PT0001^^^HOSP^MR||DOE^JANE||19800601|F",
    "DG1|1||I10^Essential hypertension^I10C||20240314",
]
folder = tempfile.TemporaryDirectory()
(pathlib.Path(folder.name) / "adt.hl7").write_text(
    "\r".join(messages), encoding="utf-8"
)

with contextlib.redirect_stdout(io.StringIO()):
    (record,) = HL7v2Loader().load(folder.name)
assert record.patient_id == "PT0001" and record.demographics.gender == "female"
assert record.conditions[0].display == "Essential hypertension"
folder.cleanup()

What it reads and how, so you can judge whether it fits your feed:

  • It reads PID (identifier, birth date, sex, race, ethnic group, death), PV1 (visit number, class, admit and discharge times), DG1, OBX, RXE/RXA and PR1. Everything else is ignored, and names, addresses and phone numbers are never read.
  • A code is mapped only through coding-system names HL7 defines for the code sets Openbtk models: SCT (SNOMED CT), LN (LOINC), I10C (ICD-10-CM), RXNORM, C4 (CPT-4). Plain I10 is the WHO ICD-10, a different code set, and is not treated as ICD-10-CM. An event with no mappable code is skipped and counted in a log line, not raised.
  • A timestamp with no UTC offset is assumed to be UTC. HL7 v2 does not label local time, so if your interface engine emits it, pass default_utc_offset_hours=.
  • The patient identifier (PID-3) is PHI. De-identify the records before they leave your environment, as for any other source.

Joining notes to structured data

A note and a patient's codes and measurements often belong together: the discharge summary, plus the labs drawn the day before it. join_notes_to_events attaches a patient's events to each note, by shared encounter, by a time window, or either:

import contextlib
import io
from datetime import UTC, datetime, timedelta

with contextlib.redirect_stdout(io.StringIO()):
    from openbtk.data.clinical_text.schemas import ClinicalTextRecord
    from openbtk.data.ehr.schemas import Demographics, Measurement, PatientRecord
    from openbtk.pipelines import index_patients, join_notes_to_events

noon = datetime(2024, 3, 14, 12, tzinfo=UTC)
patient = PatientRecord(
    patient_id="pt-1",
    demographics=Demographics(),
    observations=[
        Measurement(code="718-7", value=13.5, timestamp=noon - timedelta(hours=6)),
        Measurement(code="718-7", value=12.0, timestamp=noon - timedelta(days=9)),
    ],
    source_system="fhir-r4",
)
note = ClinicalTextRecord(
    record_id="n1",
    source="synthetic",
    text="Discharge summary.",
    patient_ref="pt-1",
    timestamp=noon,
)

(joined,) = join_notes_to_events(
    [note], index_patients([patient]), before=timedelta(days=1)
)
assert [e.event.value for e in joined.events] == [13.5]  # the older result is outside

The join is on identifiers only: patient_ref on the note must equal the record's patient_id, and it never links by name or date of birth. A note with no patient_ref, or one that matches no patient, comes back with patient_found=False rather than being guessed at, so a join across sources that pseudonymise differently is visible as a run of False, not as silent emptiness. Notes stream; the patients are held as a lookup (one entry per patient) or fetched by a function you pass.

From EHR to retrieval-ready text

Because the timeline is a ClinicalTextRecord, de-identify it exactly like a note before it goes anywhere else:

import contextlib
import io

with contextlib.redirect_stdout(io.StringIO()):
    from openbtk.data.clinical_text.preprocessing import DeidPreprocessor
    from openbtk.deid.schemas import DeidMode

    clean = DeidPreprocessor(mode=DeidMode.REDACT).process(timeline)

assert clean.deid_status.value == "deidentified"

Guardrails for EHR data

guardrail.ehr.code_validity checks that coded events use codes their system recognises, guardrail.ehr.referential finds events that point at a missing encounter or fall outside its window, and guardrail.ehr.units flags implausible lab units and values. guardrail.ehr.k_anonymity checks that an exported cohort cannot be narrowed to fewer than k people by its quasi-identifiers (see k-anonymity). They return results rather than raising; see the guardrails reference.

Limits worth knowing

  • The FHIR loader is exercised on Synthea-shaped bundles and resource-by-resource against fhir.resources validation. It has not been run against a full real Synthea export or a real hospital's FHIR server.
  • The code-validity check consults the bundled ICD-10-CM subset unless you supply a fuller terminology service; a code it cannot confirm is a warning, never silently valid.