别再只会看图了!手把手教你用Python解析DICOM文件里的病人信息(附代码)
·
从DICOM文件中提取结构化数据的Python实战指南
在医疗AI和数据分析领域,DICOM文件常被视为医学图像的载体,但它的价值远不止于此。每个DICOM文件实际上是一个结构化的数据容器,包含了从患者基本信息、检查参数到诊断描述等丰富元数据。本文将带您深入探索如何用Python高效提取这些"隐藏"的宝藏数据。
1. DICOM文件结构与数据组织原理
DICOM标准定义了医学影像及相关信息的存储和传输规范。理解其数据结构是有效提取信息的前提:
- 文件头 :前128字节通常为空(用于兼容旧系统),接着是4字节的"DICM"标识符
- 数据元素(DataElement) :文件主体由多个数据元素顺序组成,每个元素包含:
- Tag :由组号(2字节)和元素号(2字节)组成的唯一标识符
- VR(Value Representation) :定义数据类型和格式(如字符串、数字等)
- 值长度 :数据值的字节长度
- 值域 :实际存储的数据内容
常用Tag组及其典型用途:
| 组号范围 | 主要内容类别 | 典型示例 |
|---|---|---|
| 0002 | 文件元信息 | 传输语法、实现类UID |
| 0008 | 检查特征参数 | 检查日期、序列描述 |
| 0010 | 患者信息 | 姓名、ID、出生日期 |
| 0028 | 图像参数 | 行列数、像素间距 |
2. 搭建Python解析环境
我们使用 pydicom 库进行DICOM文件操作,这是目前最成熟的Python DICOM处理工具:
pip install pydicom
基础读取代码示例:
import pydicom
def load_dicom(file_path):
try:
ds = pydicom.dcmread(file_path)
return ds
except Exception as e:
print(f"读取DICOM文件失败: {str(e)}")
return None
处理常见异常情况:
- 文件损坏或非DICOM格式
- 权限问题导致的读取失败
- 大文件内存限制
3. 核心数据提取技术
3.1 访问基础患者信息
患者信息主要存储在0010组Tag中:
def extract_patient_info(ds):
info = {
'name': getattr(ds, 'PatientName', ''),
'id': getattr(ds, 'PatientID', ''),
'birth_date': getattr(ds, 'PatientBirthDate', ''),
'sex': getattr(ds, 'PatientSex', ''),
'age': getattr(ds, 'PatientAge', '')
}
# 处理可能的编码问题
if isinstance(info['name'], pydicom.valuerep.PersonName):
info['name'] = str(info['name'])
return info
3.2 提取检查与设备信息
检查相关元数据主要分布在0008组和0018组:
def extract_study_info(ds):
return {
'study_date': getattr(ds, 'StudyDate', ''),
'study_time': getattr(ds, 'StudyTime', ''),
'modality': getattr(ds, 'Modality', ''),
'manufacturer': getattr(ds, 'Manufacturer', ''),
'station_name': getattr(ds, 'StationName', ''),
'study_description': getattr(ds, 'StudyDescription', '')
}
3.3 处理特殊数据类型
DICOM中某些数据类型需要特殊处理:
- 序列(SQ) :嵌套的数据元素集合
- 多值数据 :以""分隔的字符串
- 二进制数据 :如像素数据(7FE0,0010)
序列数据处理示例:
def process_sequence(ds, tag):
seq = getattr(ds, tag, [])
results = []
for item in seq:
item_data = {}
for elem in item:
if elem.VR == 'SQ':
item_data[elem.name] = process_sequence(item, elem.name)
else:
item_data[elem.name] = elem.value
results.append(item_data)
return results
4. 高级技巧与实战解决方案
4.1 处理编码问题
DICOM文件可能使用特定字符编码:
def decode_specific_vr(ds, tag):
elem = ds.get(tag)
if not elem:
return None
if elem.VR == 'PN': # 人名特殊处理
return str(elem.value)
elif elem.VR in ['LO', 'LT', 'SH', 'ST', 'UT']: # 文本类型
try:
return elem.value.decode('iso8859_1')
except:
return str(elem.value)
return elem.value
4.2 批量处理优化
高效处理大量DICOM文件的策略:
- 多进程/线程处理
- 内存映射大文件
- 选择性读取只需要的Tag
from multiprocessing import Pool
def batch_process(files, output_dir, workers=4):
with Pool(workers) as p:
results = p.map(process_single_file, files)
save_results(results, output_dir)
4.3 常用Tag快速参考表
下表列出了医疗数据分析中最常用的Tag:
| Tag | 名称 | VR类型 | 描述 |
|---|---|---|---|
| (0010,0010) | PatientName | PN | 患者姓名 |
| (0010,0020) | PatientID | LO | 患者ID |
| (0008,0020) | StudyDate | DA | 检查日期 |
| (0008,1030) | StudyDescription | LO | 检查描述 |
| (0028,0010) | Rows | US | 图像行数 |
| (0028,0011) | Columns | US | 图像列数 |
5. 实际应用案例
5.1 构建患者信息数据库
import sqlite3
def create_patient_db(dicom_files, db_path):
conn = sqlite3.connect(db_path)
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS patients
(id TEXT PRIMARY KEY, name TEXT, birth_date TEXT,
sex TEXT, study_count INTEGER)''')
for file in dicom_files:
ds = pydicom.dcmread(file)
info = extract_patient_info(ds)
c.execute('INSERT OR IGNORE INTO patients VALUES (?,?,?,?,0)',
(info['id'], info['name'], info['birth_date'], info['sex']))
c.execute('UPDATE patients SET study_count=study_count+1 WHERE id=?',
(info['id'],))
conn.commit()
conn.close()
5.2 检查质量分析
统计不同设备的检查参数差异:
def analyze_equipment(dicom_files):
stats = {}
for file in dicom_files:
ds = pydicom.dcmread(file)
manuf = getattr(ds, 'Manufacturer', 'Unknown')
model = getattr(ds, 'ManufacturerModelName', 'Unknown')
key = f"{manuf} {model}"
if key not in stats:
stats[key] = {
'count': 0,
'modalities': set(),
'average_rows': 0,
'average_columns': 0
}
stats[key]['count'] += 1
stats[key]['modalities'].add(getattr(ds, 'Modality', ''))
stats[key]['average_rows'] += getattr(ds, 'Rows', 0)
stats[key]['average_columns'] += getattr(ds, 'Columns', 0)
# 计算平均值
for key in stats:
count = stats[key]['count']
stats[key]['average_rows'] /= count
stats[key]['average_columns'] /= count
stats[key]['modalities'] = list(stats[key]['modalities'])
return stats
6. 性能优化与错误处理
处理大型DICOM数据集时的实用技巧:
-
延迟加载 :只读取文件头信息
ds = pydicom.dcmread('large.dcm', defer_size=1024) -
选择性读取 :指定只加载需要的Tag
tags = [0x00100010, 0x00100020] # 姓名和ID ds = pydicom.dcmread('file.dcm', specific_tags=tags) -
内存映射 :处理超大文件
with open('huge.dcm', 'rb') as f: ds = pydicom.dcmread(f, defer_size=1024)
常见错误处理模式:
def safe_get(ds, tag, default=None):
try:
elem = ds[tag]
if elem.VR == 'SQ':
return [safe_get(item, tag, default) for item in elem]
return elem.value
except:
return default
7. 扩展应用与集成方案
将DICOM元数据提取集成到现代数据处理流水线中:
import pandas as pd
from datetime import datetime
def dicom_to_dataframe(files):
records = []
for file in files:
try:
ds = pydicom.dcmread(file)
record = {
'file_path': file,
'patient_id': getattr(ds, 'PatientID', None),
'study_date': parse_date(getattr(ds, 'StudyDate', None)),
'modality': getattr(ds, 'Modality', None),
'image_size': f"{getattr(ds, 'Rows', 0)}x{getattr(ds, 'Columns', 0)}",
'timestamp': datetime.now()
}
records.append(record)
except Exception as e:
print(f"处理文件{file}时出错: {str(e)}")
return pd.DataFrame(records)
def parse_date(dicom_date):
if not dicom_date:
return None
try:
return datetime.strptime(dicom_date, '%Y%m%d')
except:
return None
8. 安全与合规注意事项
处理医疗数据时的关键考虑因素:
重要:任何涉及患者隐私的数据处理都应遵循相关法规要求,确保数据匿名化和安全存储
- 匿名化敏感信息(姓名、ID等)
- 数据加密存储
- 访问权限控制
- 审计日志记录
匿名化函数示例:
def anonymize_dicom(ds, output_path):
# 删除或替换敏感信息
tags_to_remove = [
(0x0010, 0x0010), # PatientName
(0x0010, 0x0020), # PatientID
(0x0010, 0x0030), # PatientBirthDate
]
for tag in tags_to_remove:
if tag in ds:
del ds[tag]
# 添加匿名化标识
ds.PatientName = "Anonymous"
ds.PatientID = "ANON_" + str(hash(ds.SOPInstanceUID))
ds.save_as(output_path)
更多推荐


所有评论(0)