ehr-data-parsing

EHR 数据解析

概述

此技能封装了用于解析两种主流格式临床数据的 Python 流水线：HL7v2（管道符分隔的消息）和 FHIR R4（JSON 资源）。涵盖患者人口统计信息、实验室结果、诊断和药物的提取，以及格式转换和数据质量检查。

范围：

• HL7v2 消息解析（MSH、PID、PV1、OBR、OBX 段）
• FHIR R4 资源提取（Patient、Observation、Condition、MedicationRequest）
• HL7v2 → 表格（CSV/DataFrame）转换
• FHIR Bundle 遍历和扁平化
• 数据质量检查（完整性、日期验证、代码系统验证）

范围之外：HL7v3/CDA 解析、FHIR 服务部署、SMART on FHIR 身份验证、实时消息路由。

用法

当用户要求以下操作时调用：

• 解析 HL7v2 消息或批处理文件
• 从 FHIR 中提取患者人口统计信息、实验室结果、诊断或药物
• 将 HL7v2 消息转换为 CSV 或 pandas DataFrame
• 将 FHIR Bundle 扁平化为表格格式
• 对临床数据提取结果执行质量检查

此技能会使用 python-hl7 处理 HL7v2，并使用 fhir.resources 处理 FHIR R4，生成 Python 代码。安装依赖：

pip install python-hl7 fhir.resources pandas

响应格式

• 先给出用户所需的命令或代码，再进行解释
• 按以下结构组织：确认输入 → 可运行的代码 → 关键参数说明 → 注意事项
• 每项任务提供一个完整可运行的示例；不要展示所有备选方案
• 代码注释应简洁实用（说明做什么，而不是为什么存在）
• 目标：50–100 行代码，并附简短说明

核心概念

解析器选择决策树

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 消息结构

HL7v2 消息以管道符分隔，采用段结构。每个段以 3 个字符的标识符开头，后跟以 | 分隔的字段。

关键段

段	名称	内容
MSH	消息头	发送/接收应用、消息类型、时间戳、版本
PID	患者身份识别	MRN、姓名、出生日期、性别、地址、电话
PV1	患者就诊	就诊编号、患者类别、主治医师
OBR	检查申请	申请编号、申请的检查、申请医师
OBX	检查结果	结果值、单位、参考范围、异常标志
DG1	诊断	诊断代码、类型、描述
AL1	过敏	过敏原、反应、严重程度

常见消息类型

类型	触发事件	用途
ADT^A01	入院	患者入院
ADT^A08	更新	更新患者信息
ORU^R01	观察结果	实验室/放射学结果
ORM^O01	医嘱	新建医嘱
MDM^T02	文档	文档通知

1. 解析 HL7v2 消息

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)

从 PID 中提取患者人口统计信息

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 字段参考

字段	位置	格式
患者 ID	PID-3	CX 数据类型（ID^^^authority^type）
患者姓名	PID-5	XPN：姓^名^中间名
出生日期	PID-7	YYYYMMDD
性别	PID-8	M、F、O、U
地址	PID-11	XAD：街道^城市^州^邮编
电话	PID-13	XTN 数据类型

从 OBX 段提取实验室结果

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 值类型和状态代码

OBX-2	含义	OBX-8	含义	OBX-11	含义
NM	数值	N	正常	F	最终
ST	字符串	H	偏高	P	初步
CE	编码条目	L	偏低	C	已更正

2. 批量处理 HL7v2

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 资源提取

解析 Patient 资源

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

提取 Observation、Condition、MedicationRequest

模式：model_validate(json) → 按系统 URI 提取，而不是按数组位置提取。

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 扁平化

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. 数据质量检查

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

质量检查阈值

检查项	通过	警告	失败
完整性（必填字段）	≥ 95%	80–95%	< 80%
日期有效性	0 个解析错误	—	任意解析错误
日期范围	全部处于预期范围内	—	未来日期或早于 1900 年的日期
代码系统成员资格	所有代码均有效	—	存在无法识别的代码
患者 ID 重复	0 个完全重复项	—	存在人口统计信息冲突的重复 MRN

6. HL7v2 到 FHIR 的概念映射

HL7v2 段	FHIR 资源	关键字段
PID	Patient	identifier, name, birthDate, gender
PV1	Encounter	class, period, participant
OBR + OBX	Observation	code, value, effectiveDateTime, status
DG1	Condition	code, onsetDateTime, clinicalStatus
ORC + RXE	MedicationRequest	medicationCodeableConcept, dosageInstruction
AL1	AllergyIntolerance	code, reaction, criticality

常见错误

• **错误：**未检查空段，就假设所有 HL7v2 版本中的字段位置固定 **正确：**访问子字段前始终检查段是否为空——字段位置稳定，但可选字段可能不存在 **原因：**访问空段的子字段会引发索引错误或返回错误数据

• **错误：**用 \n 而不是 \r 分割 HL7v2 消息 **正确：**解析前将段分隔符规范化为 \r——许多文件使用 \r\n 或 \n **原因：**HL7v2 标准的段分隔符是回车符（\r）；分割方式不正确会生成格式错误的段

• **错误：**未检查 OBX-2 中的值类型就解析 OBX-5 **正确：**始终先检查 OBX-2——数值型（NM）和编码型（CE）结果需要不同的解析逻辑 **原因：**将编码条目误读为数值型（反之亦然）会导致数据损坏

• **错误：**将所有 OBX 段都视为实验室结果 **正确：**按 OBR 上下文或 OBX-3 代码系统筛选，以区分实验室结果、生命体征、影像学印象和临床记录 **原因：**OBX 包含多种观察类型；假设它们全是实验室结果会导致数据集混杂且无法使用

• **错误：**硬编码 FHIR 资源字段路径（例如始终访问 effectiveDateTime） **正确：**检查实际存在的选择类型变体（例如 effectiveDateTime 与 effectivePeriod，valueQuantity 与 valueCodeableConcept） **原因：**FHIR 允许选择类型——硬编码路径会引发 AttributeError，或者在使用其他变体时漏掉数据

• **错误：**将初步 Observation（status: "preliminary"）纳入分析数据集 **正确：**分析时按 status: "final" 筛选——初步结果可能会被后续结果取代 **原因：**初步结果可能会被更正或替换，导致分析中出现重复或不准确的数据

• **错误：**按数组位置（例如 coding[0]）提取编码值，而不是按系统 URI 提取 **正确：**按系统 URI 提取（例如筛选 "http://snomed.info/sct"）——Condition 可能同时包含 SNOMED 和 ICD-10 编码 **原因：**数组顺序无法保证；按位置访问会不可预测地返回错误的代码系统

• **错误：**处理 FHIR Bundle 时，只提取第一页而不检查分页 **正确：**检查 bundle.link 中是否有 relation: "next"，并继续获取所有页面，直到没有下一页 **原因：**FHIR 服务会返回分页的 Bundle——停在第一页会漏掉大部分数据

• **错误：**假设 HL7v2 和 FHIR 中的所有日期都采用 ISO 8601 格式 **正确：**将 HL7v2 日期解析为 YYYYMMDD[HHmmss]，将 FHIR 日期按不同精度解析（2024、2024-01、2024-01-15） **原因：**格式不匹配会导致解析错误或数据静默损坏

• **错误：**未验证代码系统 URI 就接受编码值 **正确：**确认 system 字段与预期 URI 相符（例如，LOINC 代码应使用 http://loinc.org） **原因：**没有系统 URI 的代码含义不明确——未经系统确认的 "6690-2" 不一定是 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/