心电数据分析第一步：手把手教你用Python把XML格式的12导联数据整理成规整的TXT表格

第一次接触临床心电数据的研究者，往往会被原始数据的复杂格式困扰。医院心电图机导出的XML文件虽然包含完整的12导联信息，但直接用于机器学习或统计分析时，数据结构的嵌套和冗余会让后续处理变得异常麻烦。本文将带你用Python实现从原始XML到分析友好型TXT表格的完整转换流程，特别针对I、II、V1-V6这8个临床最常用的导联数据进行结构化处理。

1. 理解心电XML数据结构与转换目标

典型的心电XML文件包含三层关键信息：

• 患者元数据：记录ID、采集时间等基本信息
• 导联配置：12导联的物理连接方式
• 采样数值：每个导联随时间变化的电压值序列

以500Hz采样率、30秒时长的数据为例，每个导联包含15,000个数据点。原始XML通常将这些数值存储为空格分隔的长字符串：

<digits>784 802 812 798 ...</digits>

我们需要提取的目标导联包括：

• 肢体导联：I、II
• 胸导联：V1-V6

转换后的TXT文件应当满足：

1. 每列对应一个导联的数据序列
2. 列间用单个空格分隔
3. 保留原始数值精度
4. 支持np.loadtxt直接读取

2. 搭建Python解析环境

推荐使用以下工具链组合：

# 必需库
import xml.dom.minidom  # XML解析
import re  # 字符串处理
import os  # 文件操作
import numpy as np  # 数值处理
from tqdm import tqdm  # 进度条

# 可选但推荐的库
import pandas as pd  # 数据合并
import multiprocessing  # 并行处理

注意：临床心电数据通常体积较大，建议在8GB以上内存环境中运行批量转换

创建规范的目录结构：

/project
  /raw_xml   # 原始XML存放处
  /output_txt  # 生成TXT保存位置
  /processed_data  # 后续分析用

3. XML解析核心代码实现

3.1 单文件解析函数

def parse_ecg_xml(xml_path):
    """解析单个心电XML文件，返回导联数据字典"""
    dom = minidom.parse(xml_path)
    digits = dom.getElementsByTagName('digits')
    
    leads = {
        'I': digits[0].firstChild.data,
        'II': digits[1].firstChild.data,
        'V1': digits[6].firstChild.data,
        'V2': digits[7].firstChild.data,
        'V3': digits[8].firstChild.data,
        'V4': digits[9].firstChild.data,
        'V5': digits[10].firstChild.data,
        'V6': digits[11].firstChild.data
    }
    
    # 字符串转数值列表
    for lead in leads:
        leads[lead] = [float(x) for x in re.split(r'\s+', leads[lead].strip())]
    
    return leads

3.2 批量转换与保存

def batch_convert(xml_dir, output_dir):
    """批量处理XML文件夹"""
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
    
    xml_files = [f for f in os.listdir(xml_dir) if f.endswith('.xml')]
    
    for xml_file in tqdm(xml_files):
        try:
            leads = parse_ecg_xml(os.path.join(xml_dir, xml_file))
            output_path = os.path.join(output_dir, xml_file.replace('.xml', '.txt'))
            
            # 转置为列式存储
            np.savetxt(output_path, 
                      np.column_stack(list(leads.values())),
                      delimiter=' ',
                      fmt='%.3f')
            
        except Exception as e:
            print(f"Error processing {xml_file}: {str(e)}")

关键参数说明：

• delimiter=' ' 确保空格分隔
• fmt='%.3f' 保留3位小数
• column_stack 实现行转列

4. 高级处理技巧

4.1 多文件合并为DataFrame

def merge_txt_to_df(txt_dir):
    """合并多个TXT文件为单个DataFrame"""
    all_data = []
    txt_files = [f for f in os.listdir(txt_dir) if f.endswith('.txt')]
    
    for txt_file in txt_files:
        file_path = os.path.join(txt_dir, txt_file)
        data = np.loadtxt(file_path)
        df = pd.DataFrame(data, columns=['I','II','V1','V2','V3','V4','V5','V6'])
        df['source_file'] = txt_file
        all_data.append(df)
    
    return pd.concat(all_data, ignore_index=True)

4.2 内存优化方案

对于超大规模数据集（>10GB）：

1. 使用HDF5格式替代CSV：

store = pd.HDFStore('ecg_store.h5')
for chunk in pd.read_csv('merged.csv', chunksize=100000):
    store.append('ecg_data', chunk)

2. 启用Dask进行分布式处理

5. 数据质量验证

转换完成后应检查：

1. 采样点数量一致性：

assert len(set([len(df) for df in all_data])) == 1

2. 数值范围合理性：

for lead in ['I','II','V1']:
    assert df[lead].between(-10, 10).all()

3. 缺失值检测：

missing = df.isnull().sum()
print(f"Missing values per lead:\n{missing}")

典型问题处理方案：

问题类型	检测方法	解决方案
数据截断	检查行数	回查原始XML
数值溢出	描述统计	滤波处理
导联错位	相关性分析	重新索引