ehr-data-parsing

Name: ehr-data-parsing
Author: awslabs

github.com/awslabsデータ・データベース6881 回インストール4 週間前に更新

リポジトリ

ehr-data-parsing について

ehr-data-parsingは、HL7v2メッセージとFHIR R4リソースを解析するPythonパイプラインを作成するスキルです。臨床データの抽出、表形式への変換、Bundleのフラット化、品質チェックを行う分析担当者向けです。

主な機能

01HL7v2メッセージセグメントの解析
02FHIR R4リソースの抽出
03HL7v2をCSVやDataFrameに変換
04FHIR Bundleのフラット化
05臨床データの品質チェック

活用シーン

01患者属性や検査結果の抽出
02HL7v2バッチを表形式データに変換
03分析用にFHIR Bundleをフラット化
04臨床データ抽出結果の検証

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/awslabs/hcls-agent-skills --skill ehr-data-parsing

インストール後、Claude Code で/ehr-data-parsingを入力して読み込まれたか確認します。

詳細

EHR データの解析

概要

このスキルでは、主要な 2 つの形式である HL7v2（パイプ区切りのメッセージ）と FHIR R4（JSON リソース）から臨床データを解析する Python パイプラインを提供します。患者の人口統計情報、検査結果、診断、投薬情報の抽出に加え、形式変換とデータ品質チェックにも対応しています。

対象範囲：

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 を表形式にフラット化
臨床データ抽出結果の品質チェックを実行

このスキルでは、HL7v2 に python-hl7、FHIR R4 に fhir.resources を使用した Python コードを生成します。依存関係をインストールします：

pip install python-hl7 fhir.resources pandas

応答形式

まずユーザーに必要なコマンドまたはコードを提示し、その後に説明します
次の構成にします：入力の確認 → 動作するコード → 主なパラメーターの説明 → 注意点
タスクごとに、完全に動作する例を 1 つ示します。すべての代替案は示しません
コードコメントは最小限かつ実用的にします（存在理由ではなく、何をするかを説明）
目安：コード 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 バージョンでフィールド位置が固定されていると仮定する 正しい方法： サブフィールドにアクセスする前に、必ず空のセグメントを確認します。フィールド位置は安定していますが、オプションのフィールドが存在しない場合があります 理由： 空のセグメントのサブフィールドにアクセスすると、インデックスエラーが発生するか、誤ったデータが返されます
誤り： HL7v2 メッセージを \r ではなく \n で分割する 正しい方法： 解析する前にセグメント区切り文字を \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" でフィルタリングします。予備的な結果は後から置き換えられる場合があります 理由： 予備的な結果は修正または置換される可能性があり、分析に重複データや不正確なデータが含まれる原因になります
誤り： システム URI ではなく配列の位置（例：coding[0]）でコード値を抽出する 正しい方法： システム 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/

Markdown 対応

ログインしてコメント

基本情報

作者: awslabs
リポジトリ: awslabs/hcls-agent-skills
カテゴリ: データ・データベース
インストール数: 81
GitHub スター: 68

問題を報告