4、Python数据读取：文件I/O操作、二进制数据解析、使用numpy处理心电数据
好，咱们进入第四讲。前面几章我们把心电数据的格式和存储讲清楚了，这一章就来动真格的——用Python把数据读进来。
说实话，文件I/O这块，很多初学者觉得简单，不就是open、read、close嘛。但心电数据不一样，它往往是二进制格式，不是文本文件。你想想看，一个24小时的心电记录，动辄几百兆甚至上G，如果用文本存，那体积得翻好几倍。所以，我们必须学会跟二进制数据打交道。
4.1 文件I/O基础：打开、读取、关闭
Python的文件操作，核心就三个函数：open()、read()、close()。但这里有个坑——很多人忘记关闭文件。我个人习惯用with语句，它会自动帮你关闭，省心。
# 最基本的文件读取
with open('ecg_data.dat', 'rb') as f: # 'rb' 表示以二进制只读模式打开
raw_data = f.read() # 一次性读取所有字节
print(f"读取了 {len(raw_data)} 个字节")
注意这里的'rb'模式。如果你写成'r'，Python会尝试按文本解码，遇到非UTF-8的字节直接报错。我在项目中遇到过，同事把二进制文件当文本读，结果程序跑了一小时，啪一下崩了，就是因为某个字节恰好不是合法UTF-8字符。
4.2 二进制数据解析：struct模块
二进制数据读进来了，但怎么知道每个字节代表什么？这就需要用到struct模块了。它能把字节流按照指定格式解析成Python的数据类型。
心电数据常见的存储格式是这样的：每个采样点用2个字节（16位有符号整数）表示，采样率250Hz，先存12导联的数据，再存时间戳。嗯，这里要注意，不同设备厂商的格式可能不一样，但原理相同。
import struct
# 假设我们读取了前100个字节
with open('ecg_data.dat', 'rb') as f:
chunk = f.read(100)
# 解析前10个采样点（每个采样点2字节，有符号整数，小端序）
# 格式字符串：'<10h' 表示小端序、10个short（2字节整数）
format_str = '<10h'
try:
values = struct.unpack(format_str, chunk[:20])
print(f"前10个采样点: {values}")
except struct.error as e:
print(f"解析失败: {e}")
常用的格式字符：
| 字符 | 类型 | 字节数 |
|---|---|---|
| b / B | 有符号/无符号字节 | 1 |
| h / H | 有符号/无符号短整数 | 2 |
| i / I | 有符号/无符号整数 | 4 |
| f / d | 单精度/双精度浮点数 | 4 / 8 |
4.3 使用numpy处理心电数据
好，数据解析出来了，但接下来才是重头戏。心电数据动辄几十万个采样点，用Python的列表去处理？太慢了。这时候就该numpy登场了。
numpy的核心是ndarray，它底层用C语言实现，处理速度比Python列表快几十倍。而且，它提供了大量现成的数学函数，省得我们自己写循环。
import numpy as np
# 假设我们已经解析出了100000个采样点
# 方式1：从列表创建
ecg_list = [1, 2, 3, ...] # 假设有10万个值
ecg_array = np.array(ecg_list, dtype=np.int16)
# 方式2：直接从二进制数据创建（推荐！）
with open('ecg_data.dat', 'rb') as f:
# 跳过文件头（假设前256字节是头信息）
f.seek(256)
# 直接读取并转换为numpy数组
raw_bytes = f.read(200000) # 读取10万个采样点（每个2字节）
ecg_data = np.frombuffer(raw_bytes, dtype=np.int16)
print(f"数据形状: {ecg_data.shape}")
print(f"数据类型: {ecg_data.dtype}")
print(f"前5个值: {ecg_data[:5]}")
np.frombuffer 是处理二进制心电数据的利器。它直接从内存缓冲区创建数组，不需要中间列表，速度极快。我处理一个1.2G的心电文件，用这个方法只需要0.3秒就加载完毕。
4.4 实战：完整的心电数据读取流程
来，咱们把上面学的串起来，写一个完整的读取函数。这个函数我在实际项目中用过，处理过超过1000份心电数据。
import numpy as np
import struct
def load_ecg_data(filepath, num_leads=12, sample_rate=250):
"""
加载心电数据文件
参数:
filepath: 文件路径
num_leads: 导联数，默认12
sample_rate: 采样率，默认250Hz
返回:
ecg_data: numpy数组，形状为 (num_leads, num_samples)
timestamps: 时间戳数组
"""
# 1. 读取文件头（假设前64字节是头信息）
with open(filepath, 'rb') as f:
header = f.read(64)
# 解析头信息：前4字节是采样点数，后4字节是采样率
num_samples = struct.unpack('<I', header[0:4])[0]
actual_rate = struct.unpack('<I', header[4:8])[0]
print(f"文件头信息：采样点数={num_samples}, 采样率={actual_rate}")
# 2. 读取心电数据
# 每个导联每个采样点2字节，总字节数 = num_leads * num_samples * 2
expected_bytes = num_leads * num_samples * 2
raw_data = f.read(expected_bytes)
# 3. 转换为numpy数组
# 数据存储顺序：先存完所有导联的第一个采样点，再存第二个...
# 所以形状是 (num_samples, num_leads)
ecg_2d = np.frombuffer(raw_data, dtype=np.int16)
ecg_2d = ecg_2d.reshape(num_samples, num_leads)
# 4. 转置为 (num_leads, num_samples) 方便后续处理
ecg_data = ecg_2d.T
# 5. 生成时间戳
duration = num_samples / actual_rate
timestamps = np.linspace(0, duration, num_samples, endpoint=False)
return ecg_data, timestamps
# 使用示例
ecg, time = load_ecg_data('patient_001.dat')
print(f"心电数据形状: {ecg.shape}") # 输出: (12, 100000)
print(f"时间范围: {time[0]:.3f} ~ {time[-1]:.3f} 秒")
# 提取第II导联的数据
lead_ii = ecg[1] # 假设第II导联是索引1
print(f"II导联前10个值: {lead_ii[:10]}")
4.5 避坑指南
最后，分享几个我踩过的坑：
- 字节序问题： 我曾经解析一个德国设备的数据，怎么都对不上。折腾了两天，最后发现是网络字节序（大端）。记住：
struct.unpack('<h', data)是小端，struct.unpack('>h', data)是大端。 - 数据类型溢出： 心电信号有时会有基线漂移，导致数值超出int16范围。我建议用
np.int32或np.float32来存储中间结果，最后再转回int16。 - 内存管理： 一个24小时、12导联、500Hz的心电数据，大约需要 12 × 24×3600×500 × 2字节 ≈ 1.04GB。如果你的电脑内存不够，可以用
np.memmap做内存映射，按需读取。
# 使用内存映射处理超大文件
# 这样不会一次性加载到内存，而是按需读取
filename = 'huge_ecg.dat'
# 假设文件包含 12导联 × 5000000采样点
ecg_mmap = np.memmap(filename, dtype=np.int16, mode='r',
shape=(12, 5000000))
# 只读取前1000个采样点
small_chunk = ecg_mmap[:, :1000]
print(f"读取了 {small_chunk.shape[1]} 个采样点")
好了，这一章的内容就到这里。说白了，心电数据读取就三步：打开文件、解析二进制、转成numpy数组。但每一步都有细节，尤其是字节序和数据类型，搞错了数据就全废了。下一章我们来讲讲怎么把这些数据可视化，看看心电波形到底长什么样。