pip install python-hl7 fhir.resources pandas

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)

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)

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)

undefined

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)

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")

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,
}

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,
    }

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))

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",
    }