Ehr Data Parsing

EHR Data Parsing

Overview

This skill encodes Python pipelines for parsing clinical data from two dominant formats: HL7v2 (pipe-delimited messages) and FHIR R4 (JSON resources). It covers extraction of patient demographics, lab results, diagnoses, and medications, plus format conversion and data quality checks.

Scope:

• HL7v2 message parsing (MSH, PID, PV1, OBR, OBX segments)
• FHIR R4 resource extraction (Patient, Observation, Condition, MedicationRequest)
• HL7v2 → tabular (CSV/DataFrame) conversion
• FHIR Bundle iteration and flattening
• Data quality checks (completeness, date validation, code system verification)

Out of scope: HL7v3/CDA parsing, FHIR server deployment, SMART on FHIR authentication, real-time message routing.

Usage

Invoke when the user asks to:

• Parse an HL7v2 message or batch file
• Extract patient demographics, labs, diagnoses, or medications from FHIR
• Convert HL7v2 messages to CSV or pandas DataFrames
• Flatten FHIR Bundles into tabular format
• Run quality checks on clinical data extracts

The skill emits Python code using python-hl7 for HL7v2 and fhir.resources for FHIR R4. Install dependencies:

bashCopy code
pip install python-hl7 fhir.resources pandas

Response Format

• Lead with the command or code the user needs — explain after
• Structure as: confirm inputs → working code → key parameters explained → gotchas
• One complete working example per task; do not show every alternative
• Keep code comments minimal and functional (what, not why-it-exists)
• Target: 50-100 lines of code with brief surrounding explanation

Core Concepts

Parser Selection Decision Tree

Copy code
What is the input format?
├─ HL7v2 (pipe-delimited, segments start with MSH|PID|OBX...)
│  ├─ Message type?
│  │  ├─ ORU^R01 (lab results) → extract OBR + OBX segments
│  │  ├─ ADT^A01/A08 (admit/update) → extract PID + PV1 segments
│  │  └─ ORM^O01 (orders) → extract ORC + OBR segments
│  └─ Library: python-hl7 (parse → segment → field access)
└─ FHIR R4 (JSON resources)
   ├─ Single resource → fhir.resources model_validate(json)
   └─ Bundle → iterate entry[], check resource_type, follow pagination (link.next)
      ├─ Extract by system URI, not array position
      └─ Check choice types (valueQuantity vs valueCodeableConcept)

HL7v2 Message Structure

HL7v2 messages are pipe-delimited, segment-based. Each segment starts with a 3-character identifier followed by fields separated by |.

Key segments

Common message types

1. Parse HL7v2 Messages

pythonCopy code
import hl7

raw = (
    "MSH|^~\\&|LAB|HOSP|EHR|HOSP|20240115120000||ORU^R01|MSG001|P|2.5\r"
    "PID|1||MRN123^^^HOSP^MR||DOE^JOHN^A||19800315|M|||"
    "123 MAIN ST^^ANYTOWN^CA^90210\r"
    "OBR|1|ORD001||CBC^Complete Blood Count^L|||20240115080000\r"
    "OBX|1|NM|WBC^White Blood Cell Count^L||7.5|10*3/uL|4.5-11.0|N|||F\r"
    "OBX|2|NM|HGB^Hemoglobin^L||14.2|g/dL|12.0-17.5|N|||F\r"
    "OBX|3|NM|PLT^Platelet Count^L||250|10*3/uL|150-400|N|||F\r"
)

msg = hl7.parse(raw)

Extract patient demographics from PID

pythonCopy code
def extract_pid(msg):
    pid = msg.segment("PID")
    return {
        "mrn": str(pid(3)(0)(0)),          # PID-3.1: patient ID
        "last_name": str(pid(5)(0)(0)),     # PID-5.1: family name
        "first_name": str(pid(5)(0)(1)),    # PID-5.2: given name
        "dob": str(pid(7)),                 # PID-7: date of birth
        "sex": str(pid(8)),                 # PID-8: administrative sex
    }

demographics = extract_pid(msg)
# {'mrn': 'MRN123', 'last_name': 'DOE', 'first_name': 'JOHN',
#  'dob': '19800315', 'sex': 'M'}

PID field reference

Extract lab results from OBX segments

pythonCopy code
def extract_obx_results(msg):
    results = []
    for segment in msg:
        if str(segment(0)) == "OBX":
            results.append({
                "set_id": str(segment(1)),
                "value_type": str(segment(2)),       # NM=numeric, ST=string, CE=coded
                "test_code": str(segment(3)(0)(0)),   # OBX-3.1: identifier
                "test_name": str(segment(3)(0)(1)),   # OBX-3.2: text
                "value": str(segment(5)),             # OBX-5: observation value
                "units": str(segment(6)),             # OBX-6: units
                "ref_range": str(segment(7)),         # OBX-7: reference range
                "abnormal_flag": str(segment(8)),     # OBX-8: N/H/L/A
                "status": str(segment(11)),           # OBX-11: F=final, P=preliminary
            })
    return results

labs = extract_obx_results(msg)

OBX value types and status codes

2. Batch HL7v2 Processing

pythonCopy code
import hl7
import pandas as pd

def parse_hl7_file(filepath):
    """Parse a file containing multiple HL7v2 messages separated by blank lines."""
    with open(filepath, "r") as f:
        content = f.read()

    raw_messages = content.strip().split("\nMSH|")
    raw_messages = [raw_messages[0]] + ["MSH|" + m for m in raw_messages[1:]]

    all_results = []
    for raw in raw_messages:
        raw = raw.strip().replace("\n", "\r")
        if not raw:
            continue
        msg = hl7.parse(raw)
        demo = extract_pid(msg)
        for lab in extract_obx_results(msg):
            all_results.append({**demo, **lab})

    return pd.DataFrame(all_results)

df = parse_hl7_file("lab_results.hl7")

3. FHIR R4 Resource Extraction

Parse a Patient resource

pythonCopy code
from fhir.resources.patient import Patient

patient_json = {
    "resourceType": "Patient",
    "id": "example-001",
    "identifier": [
        {"system": "http://hospital.example.org/mrn", "value": "MRN123"}
    ],
    "name": [{"family": "Doe", "given": ["John", "A"]}],
    "gender": "male",
    "birthDate": "1980-03-15",
    "address": [
        {"line": ["123 Main St"], "city": "Anytown", "state": "CA", "postalCode": "90210"}
    ],
}

patient = Patient.model_validate(patient_json)

demographics = {
    "id": patient.id,
    "mrn": patient.identifier[0].value if patient.identifier else None,
    "last_name": patient.name[0].family if patient.name else None,
    "first_name": patient.name[0].given[0] if patient.name and patient.name[0].given else None,
    "gender": patient.gender,
    "birth_date": str(patient.birthDate) if patient.birthDate else None,
}

Extract Observation, Condition, MedicationRequest

Pattern: model_validate(json) → extract by system URI, not array position.

pythonCopy code
from fhir.resources.observation import Observation
from fhir.resources.condition import Condition
from fhir.resources.medicationrequest import MedicationRequest

def extract_observation(obs_json):
    obs = Observation.model_validate(obs_json)
    return {
        "id": obs.id, "status": obs.status,
        "loinc_code": next((c.code for c in (obs.code.coding or []) if c.system == "http://loinc.org"), None),
        "display": obs.code.coding[0].display if obs.code.coding else None,
        "value": obs.valueQuantity.value if obs.valueQuantity else None,
        "unit": obs.valueQuantity.unit if obs.valueQuantity else None,
        "effective_date": str(obs.effectiveDateTime) if obs.effectiveDateTime else None,
        "patient_ref": obs.subject.reference if obs.subject else None,
    }

def extract_condition(cond_json):
    cond = Condition.model_validate(cond_json)
    return {
        "id": cond.id,
        "status": cond.clinicalStatus.coding[0].code if cond.clinicalStatus and cond.clinicalStatus.coding else None,
        "snomed_code": next((c.code for c in (cond.code.coding or []) if c.system and "snomed" in c.system), None),
        "icd10_code": next((c.code for c in (cond.code.coding or []) if c.system and "icd-10" in c.system), None),
        "display": cond.code.coding[0].display if cond.code.coding else None,
        "onset": str(cond.onsetDateTime) if cond.onsetDateTime else None,
        "patient_ref": cond.subject.reference if cond.subject else None,
    }

def extract_medication_request(med_json):
    med = MedicationRequest.model_validate(med_json)
    cc = med.medicationCodeableConcept
    return {
        "id": med.id, "status": med.status,
        "rxnorm_code": cc.coding[0].code if cc and cc.coding else None,
        "display": cc.coding[0].display if cc and cc.coding else None,
        "dosage_text": med.dosageInstruction[0].text if med.dosageInstruction else None,
        "authored_on": str(med.authoredOn) if med.authoredOn else None,
        "patient_ref": med.subject.reference if med.subject else None,
    }

4. FHIR Bundle Flattening

pythonCopy code
import json
import pandas as pd
from fhir.resources.bundle import Bundle

def flatten_observation_bundle(bundle_json):
    """Extract all Observation resources from a FHIR Bundle into a DataFrame."""
    bundle = Bundle.model_validate(bundle_json)
    rows = [extract_observation(entry.resource.model_dump())
            for entry in (bundle.entry or [])
            if entry.resource.resource_type == "Observation"]
    return pd.DataFrame(rows)

with open("observations_bundle.json") as f:
    df = flatten_observation_bundle(json.load(f))

5. Data Quality Checks

pythonCopy code
import pandas as pd
from datetime import datetime

def check_completeness(df, required_columns):
    """Report missing value rates for required columns."""
    report = {}
    for col in required_columns:
        if col not in df.columns:
            report[col] = {"status": "MISSING_COLUMN", "pct_complete": 0.0}
        else:
            pct = (df[col].notna().sum() / len(df)) * 100
            report[col] = {
                "status": "OK" if pct >= 95 else "WARNING" if pct >= 80 else "FAIL",
                "pct_complete": round(pct, 1),
            }
    return report

def validate_dates(series, fmt="%Y-%m-%d", min_date="1900-01-01", max_date=None):
    """Validate date strings. Returns indices of invalid entries."""
    max_dt = datetime.strptime(max_date, fmt) if max_date else datetime.now()
    min_dt = datetime.strptime(min_date, fmt)
    invalid = []
    for idx, val in series.items():
        if pd.isna(val):
            continue
        try:
            dt = datetime.strptime(str(val)[:10], fmt)
            if dt < min_dt or dt > max_dt:
                invalid.append({"index": idx, "value": val, "reason": "out_of_range"})
        except ValueError:
            invalid.append({"index": idx, "value": val, "reason": "parse_error"})
    return invalid

def verify_code_system(df, code_col, valid_codes, system_name="unknown"):
    """Check that coded values belong to an expected set."""
    if code_col not in df.columns:
        return {"status": "MISSING_COLUMN", "invalid_count": 0}
    codes = df[code_col].dropna().unique()
    invalid = [c for c in codes if c not in valid_codes]
    return {
        "system": system_name, "total_unique": len(codes),
        "invalid_codes": invalid, "invalid_count": len(invalid),
        "status": "OK" if len(invalid) == 0 else "FAIL",
    }

Quality check thresholds

6. HL7v2 to FHIR Conceptual Mapping

Common Mistakes

• Wrong: Assuming fixed field positions across all HL7v2 versions without checking for empty segments Right: Always check for empty segments before accessing subfields — field positions are stable but optional fields may be absent Why: Accessing a subfield of an empty segment raises index errors or returns incorrect data

• Wrong: Splitting HL7v2 messages on \n instead of \r Right: Normalize segment delimiters to \r before parsing — many files use \r\n or \n Why: The HL7v2 standard segment delimiter is carriage return (\r); incorrect splitting produces malformed segments

• Wrong: Parsing OBX-5 without checking the value type in OBX-2 Right: Always check OBX-2 first — numeric (NM) and coded (CE) results require different parsing logic Why: Misinterpreting a coded entry as numeric (or vice versa) produces corrupt data

• Wrong: Treating all OBX segments as lab results Right: Filter by OBR context or OBX-3 code system to distinguish labs from vitals, imaging impressions, and clinical notes Why: OBX carries many observation types; assuming all are labs leads to mixed, unusable datasets

• Wrong: Hardcoding FHIR resource field paths (e.g., always accessing effectiveDateTime) Right: Check which choice type variant is present (e.g., effectiveDateTime vs effectivePeriod, valueQuantity vs valueCodeableConcept) Why: FHIR allows choice types — hardcoded paths raise AttributeError or miss data when the alternate variant is used

• Wrong: Including preliminary Observations (status: "preliminary") in analysis datasets Right: Filter on status: "final" for analysis — preliminary results may be superseded Why: Preliminary results can be corrected or replaced, leading to duplicate or inaccurate data in analyses

• Wrong: Extracting coded values by array position (e.g., coding[0]) instead of by system URI Right: Extract by system URI (e.g., filter for "http://snomed.info/sct") — a Condition may carry both SNOMED and ICD-10 codings Why: Array order is not guaranteed; position-based access returns the wrong code system unpredictably

• Wrong: Processing only the first page of a FHIR Bundle without checking for pagination Right: Check bundle.link for relation: "next" and follow all pages until exhausted Why: FHIR servers return paginated Bundles — stopping at page one misses most of the data

• Wrong: Assuming all dates are ISO 8601 format across HL7v2 and FHIR Right: Parse HL7v2 dates as YYYYMMDD[HHmmss] and FHIR dates with variable precision (2024, 2024-01, 2024-01-15) Why: Format mismatches cause parse errors or silent data corruption

• Wrong: Accepting coded values without validating the code system URI Right: Verify that the system field matches the expected URI (e.g., http://loinc.org for LOINC codes) Why: Codes without system URIs are ambiguous — a "6690-2" without system confirmation may not be LOINC

References

• python-hl7: https://python-hl7.readthedocs.io/
• fhir.resources (Python): https://pypi.org/project/fhir.resources/
• FHIR R4 Specification: https://hl7.org/fhir/R4/
• LOINC: https://loinc.org/
• SNOMED CT: https://www.snomed.org/