医療データは、これまでに扱ってきたどのデータとも異なります。この記事では、その理由 と、初日から適切に使用する方法について説明します。
1. DICOM — 医療画像規格
DICOM (Digital Imaging and Communications in Medicine) は、医療画像の保存と送信のための国際標準です。 1993 年に開始された DICOM は、現在、世界中のすべての病院で使用されています。
＃＃＃１．１． DICOM ファイル構造
DICOM は単なる画像ではなく、画像 + メタデータです。
File DICOM (.dcm)
├── File Meta Information
│ ├── Transfer Syntax UID (cách encode dữ liệu pixel)
│ └── SOP Class UID (loại hình ảnh: CT, MRI, X-ray...)
│
└── Data Set (hàng trăm tags)
├── Patient Info
│ ├── (0010,0010) PatientName = "Nguyen^Van^A"
│ ├── (0010,0020) PatientID = "BN-2024-001"
│ ├── (0010,0030) PatientBirthDate = "19650315"
│ └── (0010,0040) PatientSex = "M"
│
├── Study Info
│ ├── (0020,000D) StudyInstanceUID
│ ├── (0008,0020) StudyDate = "20240115"
│ └── (0008,1030) StudyDescription = "CHEST AP"
│
├── Image Info
│ ├── (0028,0010) Rows = 2048
│ ├── (0028,0011) Columns = 2048
│ ├── (0028,0030) PixelSpacing = [0.175, 0.175] mm/pixel
│ ├── (0028,1050) WindowCenter = 40
│ ├── (0028,1051) WindowWidth = 400
│ └── (0028,0103) PixelRepresentation = 1 (signed integers)
│
└── Pixel Data
└── (7FE0,0010) PixelData = [raw 12-bit pixel values...]
＃＃＃１．２． pydicom で DICOM を読み取る
import pydicom
import numpy as np
import matplotlib.pyplot as plt
# Đọc file DICOM
dcm = pydicom.dcmread("chest_xray.dcm")
# Xem metadata
print(f"Patient: {dcm.PatientName}")
print(f"Study Date: {dcm.StudyDate}")
print(f"Modality: {dcm.Modality}") # CR (X-ray), CT, MR, US...
print(f"Image size: {dcm.Rows} x {dcm.Columns}")
print(f"Pixel spacing: {dcm.PixelSpacing} mm/pixel")
# Lấy pixel array (giá trị thô — chưa phải HU units!)
raw_pixels = dcm.pixel_array
print(f"Pixel shape: {raw_pixels.shape}")
print(f"Pixel dtype: {raw_pixels.dtype}") # Thường là int16 cho CT
print(f"Min/Max values: {raw_pixels.min()} / {raw_pixels.max()}")
# Chuyển sang Hounsfield Units (HU) cho CT
# Công thức: HU = pixel_value * RescaleSlope + RescaleIntercept
def to_hounsfield_units(dcm_file):
pixels = dcm_file.pixel_array.astype(np.float32)
slope = float(dcm_file.RescaleSlope) if hasattr(dcm_file, 'RescaleSlope') else 1.0
intercept = float(dcm_file.RescaleIntercept) if hasattr(dcm_file, 'RescaleIntercept') else 0.0
return pixels * slope + intercept
hu_image = to_hounsfield_units(dcm)
# HU reference values:
# Air: -1000 HU | Lung: -500 HU | Fat: -100 HU
# Water/Soft tissue: 0-80 HU | Blood: 40-80 HU | Bone: 400-1000 HU
＃＃＃１．３．ウィンドウ処理 — 医用画像表示の中核
画面には 256 グレー レベルしか表示されませんが、CT には約 4000 HU レベルがあります。 ウィンドウ表示 表示する HU 範囲を選択します。
def apply_windowing(image_hu: np.ndarray, window_center: int, window_width: int) -> np.ndarray:
"""
Window Center (WC) và Window Width (WW) cho các mô khác nhau:
Phổi: WC=-600, WW=1500 (thấy cấu trúc phổi chi tiết)
Ổ bụng: WC=40, WW=400 (thấy gan, lách, thận)
Não: WC=40, WW=80 (thấy hematoma)
Xương: WC=400, WW=1800 (thấy gãy xương)
"""
lower = window_center - window_width / 2
upper = window_center + window_width / 2
# Clip và normalize về [0, 1]
windowed = np.clip(image_hu, lower, upper)
normalized = (windowed - lower) / (upper - lower)
return normalized
# Hiển thị cùng một CT scan với 3 window settings khác nhau
fig, axes = plt.subplots(1, 3, figsize=(15, 5))
windows = [
("Lung Window", -600, 1500),
("Soft Tissue", 40, 400),
("Bone Window", 400, 1800),
]
for ax, (title, wc, ww) in zip(axes, windows):
windowed = apply_windowing(hu_image, wc, ww)
ax.imshow(windowed, cmap='gray')
ax.set_title(f"{title}\nWC={wc}, WW={ww}")
ax.axis('off')
plt.tight_layout()
plt.savefig("windowing_comparison.png", dpi=150)
2. HL7 FHIR — 臨床データ交換標準
HL7 FHIR (Fast Healthcare Interoperability Resources) は、医療システム間のデータ交換のための最新の標準です。画像ではありません。これは 構造化された臨床データです。
＃＃＃２．１．コアリソース
FHIR はデータを「リソース」に編成します。各エンティティ タイプには独自のスキーマがあります。
FHIR Resources (chỉ các resource quan trọng nhất cho AI)
│
├── Patient — thông tin bệnh nhân
├── Observation — vital signs, lab results, measurements
├── Condition — diagnoses, problems (mã ICD-10/11)
├── MedicationRequest— đơn thuốc
├── Procedure — phẫu thuật, can thiệp (mã CPT/ICD)
├── DiagnosticReport — báo cáo xét nghiệm, hình ảnh
├── Encounter — lần khám/nhập viện
└── AllergyIntolerance — dị ứng
＃＃＃２．２． FHIR APIで動作します
import requests
from datetime import datetime, timedelta
# FHIR R4 RESTful API
FHIR_SERVER = "https://r4.smarthealthit.org" # Public test server
def get_patient_observations(patient_id: str, code: str) -> list:
"""
Lấy lab results của bệnh nhân
code: LOINC code — chuẩn mã cho lab tests
Ví dụ LOINC codes:
- 2339-0: Glucose [Mass/volume] in Blood
- 4548-4: Hemoglobin A1c/Hemoglobin.total in Blood (HbA1c)
- 2160-0: Creatinine [Mass/volume] in Serum or Plasma
- 718-7: Hemoglobin [Mass/volume] in Blood
"""
response = requests.get(
f"{FHIR_SERVER}/Observation",
params={
"patient": patient_id,
"code": code,
"_sort": "-date", # Mới nhất trước
"_count": 100,
},
headers={"Accept": "application/fhir+json"}
)
response.raise_for_status()
bundle = response.json()
observations = []
for entry in bundle.get("entry", []):
obs = entry["resource"]
observations.append({
"date": obs.get("effectiveDateTime"),
"value": obs.get("valueQuantity", {}).get("value"),
"unit": obs.get("valueQuantity", {}).get("unit"),
"status": obs.get("status"),
})
return observations
# Ví dụ: lấy lịch sử HbA1c của bệnh nhân để dự đoán tiểu đường
hba1c_history = get_patient_observations("patient-123", "4548-4")
# [{"date": "2024-01-15", "value": 8.2, "unit": "%", "status": "final"},
# {"date": "2023-10-10", "value": 7.9, "unit": "%", "status": "final"}, ...]
def build_patient_ai_features(patient_id: str) -> dict:
"""Tổng hợp features từ FHIR cho ML model"""
# Latest vitals
bp_readings = get_patient_observations(patient_id, "55284-4") # Blood pressure
glucose = get_patient_observations(patient_id, "2339-0")
hba1c = get_patient_observations(patient_id, "4548-4")
creatinine = get_patient_observations(patient_id, "2160-0")
return {
"latest_systolic": bp_readings[0]["value"] if bp_readings else None,
"latest_glucose": glucose[0]["value"] if glucose else None,
"latest_hba1c": hba1c[0]["value"] if hba1c else None,
"hba1c_trend": (
hba1c[0]["value"] - hba1c[-1]["value"]
if len(hba1c) >= 2 else 0
),
"latest_creatinine": creatinine[0]["value"] if creatinine else None,
# ... thêm nhiều features khác
}
3. 匿名化 — 医療データの匿名化
＃＃＃３．１．匿名化が思っているよりも難しい理由
よくある間違い: 「名前と生年月日を削除するだけで完了です。」
事実: 再識別攻撃は次のものから再識別される可能性があります。
- 年齢 + 性別 + 郵便番号の組み合わせ → アメリカ人の 87% が識別可能 (Sweeney、2000)
- 珍しい診断の組み合わせ：「50歳+唾液腺がん+軽度の脳卒中」→ほぼユニーク
- ウェアラブルデータパターン → 歩き方から人を認識
- 医用画像メタデータ（撮影日、デバイス）
＃＃＃３．２． HIPAA セーフハーバー — 削除する 18 個の識別子
# HIPAA Safe Harbor method: xóa 18 loại thông tin nhận dạng
HIPAA_PHI_IDENTIFIERS = [
"name", # Tên
"geographic_data", # Địa chỉ (chỉ giữ 3 digits zip code)
"dates", # Ngày (chỉ giữ năm — QUAN TRỌNG!)
"phone_numbers", # Số điện thoại
"fax_numbers", # Fax
"email_addresses", # Email
"ssn", # Số an sinh xã hội
"medical_record_numbers", # Mã hồ sơ bệnh án
"health_plan_numbers", # Số bảo hiểm
"account_numbers", # Số tài khoản
"certificate_numbers", # Số chứng chỉ
"vehicle_identifiers", # Biển số xe
"device_identifiers", # Serial number thiết bị
"urls", # Website URLs
"ip_addresses", # Địa chỉ IP
"biometric_identifiers", # Vân tay, giọng nói, mống mắt
"full_face_photos", # Ảnh mặt
"unique_identifiers", # Bất kỳ ID duy nhất nào khác
]
import re
from dateutil import parser as dateparser
def deidentify_clinical_note(text: str) -> str:
"""
De-identify text sử dụng NLP + regex
Đây là ví dụ đơn giản — thực tế dùng tools như:
- Microsoft Presidio (open source)
- AWS Comprehend Medical + AWS Macie
- Google Cloud Healthcare DLP API
"""
# Xóa các patterns thông thường với regex
patterns = {
# Tên (giả sử đại tự hóa)
r'\b[A-Z][a-z]+ [A-Z][a-z]+\b': '[NAME]',
# Số điện thoại VN
r'\b(0[3-9]\d{8}|\+84[3-9]\d{8})\b': '[PHONE]',
# Email
r'\b[\w.+-]+@[\w-]+\.[a-z]{2,}\b': '[EMAIL]',
# Ngày tháng
r'\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b': '[DATE]',
# Tuổi chính xác (giữ age range thay vì exact age)
r'\b(\d+)\s*(tuổi|years?\s*old)\b': lambda m: f"[AGE_{(int(m.group(1))//10)*10}s]",
}
deidentified = text
for pattern, replacement in patterns.items():
if callable(replacement):
deidentified = re.sub(pattern, replacement, deidentified, flags=re.IGNORECASE)
else:
deidentified = re.sub(pattern, replacement, deidentified, flags=re.IGNORECASE)
return deidentified
# Input: "Bệnh nhân Nguyễn Văn A, 65 tuổi, nhập viện ngày 15/01/2024. SĐT: 0912345678"
# Output: "Bệnh nhân [NAME], [AGE_60s], nhập viện ngày [DATE]. SĐT: [PHONE]"
＃＃＃３．３． DICOM画像の匿名化
import pydicom
from pydicom.uid import generate_uid
def deidentify_dicom(input_path: str, output_path: str, patient_pseudoid: str):
"""
De-identify DICOM file theo DICOM Supplement 142 (Clinical Trial De-identification Profiles)
"""
dcm = pydicom.dcmread(input_path)
# Tags cần xóa hoặc thay thế (danh sách đầy đủ hơn nhiều trong thực tế)
tags_to_blank = [
(0x0010, 0x0010), # PatientName
(0x0010, 0x0030), # PatientBirthDate
(0x0010, 0x0040), # PatientSex — có thể giữ cho bias evaluation
(0x0008, 0x0080), # InstitutionName
(0x0008, 0x1070), # OperatorsName
(0x0008, 0x1048), # PhysiciansOfRecord
(0x0010, 0x1000), # OtherPatientIDs
]
for tag in tags_to_blank:
if tag in dcm:
del dcm[tag]
# Thay PatientID bằng pseudonymized ID
dcm.PatientID = patient_pseudoid
# Chỉ giữ năm, xóa tháng ngày
if hasattr(dcm, 'StudyDate'):
dcm.StudyDate = dcm.StudyDate[:4] + "0101" # Keep year only
# Tạo UIDs mới (break linkage giữa các study)
dcm.StudyInstanceUID = generate_uid()
dcm.SeriesInstanceUID = generate_uid()
dcm.SOPInstanceUID = generate_uid()
# QUAN TRỌNG: Xóa private tags (manufacturer-specific metadata)
# Có thể chứa thông tin nhận dạng không chuẩn
dcm.remove_private_tags()
dcm.save_as(output_path, write_like_original=False)
print(f"De-identified DICOM saved to {output_path}")
4. EHR システム — 実際のデータ環境を理解する
＃＃＃４．１．人気の EHR システム
| 彼の/EHR | 市場 | エクスポート形式 |
|---|---|---|
| 壮大な | 米国の病院の 33% | FHIR R4、HL7 v2 |
| Oracle Health (Cerner) | 米国の病院の 25% | FHIR R4、CCD |
| VNPT-HIS | ベトナムの多くの病院 | HL7 v2、CSV |
| バックマイ HIS 病院 | バーモント州バックマイ | 独自 + HL7 |
| メディソフト | 小さなクリニック | CSV、カスタム |
＃＃＃４．２． EHR データを扱う際の実際的な問題
# Những điều bạn sẽ gặp khi nhận EHR data thực tế:
problems = {
"missing_data": """
Lab result thiếu: bệnh nhân không làm xét nghiệm hoặc chưa import
Vitals gap: không điều dưỡng nhập giữa các ca
MCAR vs MAR vs MNAR — missing pattern quan trọng hơn missing rate!
MCAR = Missing Completely At Random (có thể impute an toàn)
MAR = Missing At Random (cần cẩn thận)
MNAR = Missing Not At Random (bias nghiêm trọng — BN nặng thường miss labs)
""",
"inconsistent_units": """
Glucose: một chỗ ghi mmol/L, chỗ khác mg/dL (nhân/chia 18.016!)
Weight: kg vs lbs
Creatinine: mg/dL vs μmol/L
→ Cần unit normalization pipeline nghiêm túc
""",
"free_text_mess": """
"BN khó thở 3 ngày, SOB on exertion, DOE x 3d"
"Pt c/o dyspnea 3 days, worsening with activity"
→ Cùng một ý, viết 100 cách khác nhau
→ NLP cần xử lý: abbreviations, typos, bilingual mixing (VN+EN)
""",
"temporal_issues": """
Timezone: bệnh viện dùng server time khác nhau
Backdating: bác sĩ nhập lại sau 3 ngày
Overlapping encounters: emergency + inpatient cùng lúc
""",
"icd_coding_errors": """
ICD-10 được code bởi billing coder, không phải bác sĩ
Over-coding (để maximize reimbursement)
Under-coding (để tránh audit)
→ Training label quality phụ thuộc vào billing accuracy!
"""
}
5. 基本的な医療データ パイプラインを構築する
from pathlib import Path
import pandas as pd
import pydicom
import torch
from torch.utils.data import Dataset
from torchvision import transforms
class MedicalDataPipeline:
"""
Pipeline đơn giản: DICOM files → ML-ready tensors
"""
def __init__(self, dicom_dir: str, labels_csv: str):
self.dicom_dir = Path(dicom_dir)
self.labels = pd.read_csv(labels_csv)
def load_and_preprocess(self, patient_id: str) -> np.ndarray:
"""Load DICOM, apply windowing, normalize cho model input"""
dcm_path = self.dicom_dir / f"{patient_id}.dcm"
dcm = pydicom.dcmread(dcm_path)
# Convert to HU
pixels = self._to_hu(dcm)
# Apply lung window (ví dụ cho chest X-ray AI)
pixels = apply_windowing(pixels, window_center=-600, window_width=1500)
# Resize to model input (224x224 cho ResNet)
from PIL import Image
img = Image.fromarray((pixels * 255).astype(np.uint8))
img = img.convert("RGB") # Grayscale → 3 channels (cho ImageNet pretrained)
img = img.resize((224, 224), Image.BILINEAR)
return np.array(img)
def _to_hu(self, dcm) -> np.ndarray:
pixels = dcm.pixel_array.astype(np.float32)
slope = float(getattr(dcm, 'RescaleSlope', 1.0))
intercept = float(getattr(dcm, 'RescaleIntercept', 0.0))
return pixels * slope + intercept
class ChestXrayDataset(Dataset):
def __init__(self, df: pd.DataFrame, dicom_dir: str, transform=None, label_cols: list = None):
self.df = df.reset_index(drop=True)
self.pipeline = MedicalDataPipeline(dicom_dir, None)
self.transform = transform
self.label_cols = label_cols or []
def __len__(self):
return len(self.df)
def __getitem__(self, idx):
row = self.df.iloc[idx]
image = self.pipeline.load_and_preprocess(row["patient_id"])
if self.transform:
image = self.transform(image)
labels = torch.FloatTensor([row[col] for col in self.label_cols])
return image, labels
6. まとめと演習
この記事を読み終えると、次のことが理解できるはずです。
- ✅ DICOM 構造: タグ、ピクセルデータ、ハウンズフィールド単位
- ✅ ウィンドウ処理: ウィンドウ処理が必要な理由とその実装方法
- ✅ FHIR R4: メインリソース、REST API
- ✅ 匿名化: HIPAA セーフハーバー、18 個の識別子
- ✅ EHR データの実際的な問題
レッスン 3 では、医療画像の完全な前処理パイプライン、つまり適切な正規化、拡張、および反転垂直胸部 X 線撮影が臨床的に間違っている理由について説明します。
演習
-
DICOM サンプル ファイルを次のサイトからダウンロードします。 TCIA (The Cancer Imaging Archive)。 pydicom で開き、最も重要な 10 個のタグを出力します。
-
関数の書き込み
apply_windowingパラメーター検証あり — window_width ≤ 0 の場合、または画像が HU 範囲 (-1024 ～ 3000) ではない場合、例外をスローします。 -
FHIR公開テストサーバーにアクセスします(
https://r4.smarthealthit.org）。 Python リクエストを使用して患者の条件リストを取得する87a339d0-8cae-418e-89c7-8651e6aab3c6。 ICDコードとは何ですか?
2. アーキテクチャと原則
コアアーキテクチャ
# Example implementation
import torch
import torch.nn as nn
class ExampleModel(nn.Module):
def __init__(self, input_dim, output_dim):
super().__init__()
self.net = nn.Sequential(
nn.Linear(input_dim, 256),
nn.ReLU(),
nn.Dropout(0.2),
nn.Linear(256, 128),
nn.ReLU(),
nn.Linear(128, output_dim),
)
def forward(self, x):
return self.net(x)
3. 練習する
セットアップ
pip install torch transformers datasets
トレーニング パイプライン
# Training loop
model = ExampleModel(input_dim=768, output_dim=10)
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)
criterion = nn.CrossEntropyLoss()
for epoch in range(10):
for batch in train_loader:
optimizer.zero_grad()
outputs = model(batch["input"])
loss = criterion(outputs, batch["label"])
loss.backward()
optimizer.step()
4. ベストプラクティス
| 側面 | 推薦 |
|---|---|
| データ | 量より質 |
| モデル | シンプルに始めてスケールアップ |
| トレーニング | 損失曲線を監視する |
| 評価 | 適切な指標を使用する |
概要
| コンセプト | 重要なポイント |
|---|---|
| 建築 | 問題に適した |
| トレーニング | ハイパーパラメータの慎重な調整 |
| 評価 | 複数のメトリクス |