简介:这份资源面向从事心电信号处理与深度学习分割实践的学习者,提供一套基于U-Net的ECG波形分段方案,用于识别给定心电信号中的P波、QRS波群与T波分量。实现上借助PyTorchWavelets的修改版本,将QTDB数据集中的ECG转换到小波域,把原始信号与标签作为输入,并保存对应尺度及小波实部、虚部信息;标签按P段为1、QRS为2、T段为3的方式标注,训练与测试脚本、网络结构定义及数据加载模块均已拆分就绪。压缩包共27个文件,以14个py源码为主,辅以8个pyc编译文件、3个md说明文档、1个license及1个ipynb交互笔记,整体约197KB,结构紧凑便于阅读与二次开发。目前已有786人学习下载,读者可据此理解小波域特征与U-Net结合的建模思路,参考数据预处理、训练评估流程及标注方式存在的问题,为后续更换数据集或改进分割效果提供可复用的代码基础。
1. 从一张心电图说起:这个项目到底在切什么
拿到一份心电图,第一眼看到的是一条上下起伏的波形,但真正做诊断的人盯的是三个局部结构:P 波、QRS 波群、T 波。P 波对应心房除极,QRS 是心室除极的主峰,T 波是心室复极。人工标注这三段极其费时,一段十秒的导联信号里可能有十几个心搏,每个心搏都要切三刀。这个项目做的事很直接:用 U-Net 把 ECG 波形逐采样点分类,输出每个点属于 P、QRS、T 还是背景,从而把波形自动切成三段。整个工程以 Jupyter Notebook 为载体,适合两类人:一类是想跑通一个生物医学信号分割基线的新手,另一类是打算在 U-Net 骨架上换数据、换损失、换后处理做二次开发的熟手。它不追求临床级精度,追求的是链路完整、能复现、能改。
2. 技术选型与工程结构:为什么是 U-Net 加小波
2.1 一维信号为什么套二维分割网络
U-Net 原本是二维图像分割网络,编码器下采样、解码器上采样、跳跃连接把浅层细节接回深层语义。ECG 是一维时间序列,直接套二维卷积并不合适,常见做法是把卷积核改成Conv1D,池化改成MaxPool1D,转置卷积改成ConvTranspose1D,其余结构照搬。这样做的理由是:ECG 的 P、QRS、T 在时间轴上是局部形态特征,QRS 窄而陡,P/T 宽而缓,编码器的多尺度感受野正好能同时抓住这两类尺度。跳跃连接在这里尤其关键,因为 QRS 的边界只有几个采样点宽,深层特征早就把它糊掉了,必须靠浅层特征把边界找回来。
项目标题里带Wavelet,说明小波变换是这条链路的一部分。小波在 ECG 处理里的典型用法是去噪和特征增强:用离散小波变换把信号分解到不同频带,QRS 主要能量集中在较高频段,基线漂移和呼吸干扰集中在低频,做阈值处理后重构,再送进网络。常见做法是用PyWavelets做 3 到 5 层分解,对近似分量做基线抑制,对细节分量做软阈值收缩。这一步不是必须的,但加了之后网络收敛通常更稳,尤其是原始数据没做过带通滤波的时候。
2.2 目录结构与 Notebook 分工
这类项目通常不会做成大型 Python 包,而是几个 Notebook 加一个数据目录。合理的组织方式大致如下:
unet-ecg-segmentation/ ├── data/ │ ├── raw/ # 原始记录,按记录编号存放 │ └── processed/ # 切窗、归一化后的 .npy 或 .npz ├── notebooks/ │ ├── 01_preprocess.ipynb # 读取、滤波、小波去噪、切窗、打标签 │ ├── 02_train.ipynb # 模型定义、训练循环、保存权重 │ └── 03_inference.ipynb # 加载权重、逐窗预测、拼接回整段 ├── src/ │ ├── model.py # U-Net 1D 定义 │ ├── dataset.py # Dataset / DataLoader │ └── postprocess.py # 峰值检测与边界修正 └── requirements.txt把模型和数据集抽到src/下、Notebook 只做调用,是这类项目最值得坚持的一条工程习惯。原因很实际:Notebook 里改模型结构,改完忘了重启内核,训练用的还是旧类,这种翻车我见过不止一次。抽成.py之后,importlib.reload或者干脆重启内核都能保证拿到最新代码。
2.3 依赖清单与版本约束
依赖不多,但版本要卡一下:
# requirements.txt 参考 numpy==1.26.4 scipy==1.11.4 PyWavelets==1.5.0 torch==2.1.2 matplotlib==3.8.2 scikit-learn==1.3.2 jupyterlab==4.0.9PyWavelets负责小波分解重构,scipy用来做带通滤波和峰值检测,torch是训练框架。版本上最容易出问题的是numpy和torch的兼容,2.x 的 numpy 在部分旧版 torch 上会报 ABI 错误,所以这里锁在 1.26。装环境建议单独建虚拟环境,不要和系统 Python 混用:
python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -r requirements.txt jupyter lab跑起来之后先打开01_preprocess.ipynb,从上到下执行一遍,确认data/processed/下有产出,再进训练。这个顺序别跳,跳过预处理直接训练,最常见的报错是张量形状对不上,排查起来反而更费时间。
3. 核心模块拆解:从原始波形到逐点标签
3.1 预处理:滤波、小波去噪与切窗
原始 ECG 直接送网络效果通常不好,基线漂移会让整段波形上下浮动,网络容易把漂移当成形态。预处理链路一般是:带通滤波去掉 0.5 Hz 以下和 40 Hz 以上的成分,再做小波去噪,最后按固定长度切窗并做 z-score 归一化。
import numpy as np import pywt from scipy.signal import butter, filtfilt def bandpass(sig, fs=360, low=0.5, high=40.0, order=3): # 巴特沃斯带通,去掉基线漂移和高频噪声 b, a = butter(order, [low / (fs / 2), high / (fs / 2)], btype="band") return filtfilt(b, a, sig) def wavelet_denoise(sig, wavelet="db4", level=4): # 分解后对细节系数做软阈值,再重构 coeffs = pywt.wavedec(sig, wavelet, level=level) sigma = np.median(np.abs(coeffs[-1])) / 0.6745 thresh = sigma * np.sqrt(2 * np.log(len(sig))) coeffs[1:] = [pywt.threshold(c, thresh, mode="soft") for c in coeffs[1:]] return pywt.waverec(coeffs, wavelet)[: len(sig)] def make_windows(sig, labels, win=1000, stride=500): # 固定窗长切分,stride 小于 win 保证重叠,缓解边界样本不足 xs, ys = [], [] for start in range(0, len(sig) - win + 1, stride): xs.append(sig[start:start + win]) ys.append(labels[start:start + win]) return np.array(xs), np.array(ys)bandpass里fs是采样率,常见公开数据集是 360 Hz 或 500 Hz,改数据时这个参数必须跟着改,否则截止频率全错。wavelet_denoise的level决定分解层数,层数太深会把 QRS 的高频细节也削掉,4 层是个稳妥起点。make_windows的win和stride是一对权衡:窗太短覆盖不到完整心搏,窗太长显存吃紧,1000 点配 500 步长在 360 Hz 下大约覆盖 2.8 秒,能装下两到三个心搏。
标签的生成方式取决于数据源。如果原始数据带逐采样点标注,直接切;如果只有 R 峰位置,就得按经验规则生成 P、QRS、T 的区间标签,比如 QRS 取 R 峰前后各 50 ms,T 波取 QRS 结束后 100 到 250 ms。这一步的规则质量直接决定上限,规则错了,网络再深也救不回来。
3.2 U-Net 1D 网络定义
网络结构是标准的四层编码加四层解码,每层两个卷积加 BN 加 ReLU,下采样用最大池化,上采样用转置卷积,跳跃连接用通道维拼接。
import torch import torch.nn as nn class DoubleConv(nn.Module): def __init__(self, in_ch, out_ch): super().__init__() self.net = nn.Sequential( nn.Conv1d(in_ch, out_ch, 3, padding=1), nn.BatchNorm1d(out_ch), nn.ReLU(inplace=True), nn.Conv1d(out_ch, out_ch, 3, padding=1), nn.BatchNorm1d(out_ch), nn.ReLU(inplace=True), ) def forward(self, x): return self.net(x) class UNet1D(nn.Module): def __init__(self, in_ch=1, n_classes=4, base=32): super().__init__() self.enc1 = DoubleConv(in_ch, base) self.enc2 = DoubleConv(base, base * 2) self.enc3 = DoubleConv(base * 2, base * 4) self.enc4 = DoubleConv(base * 4, base * 8) self.pool = nn.MaxPool1d(2) self.up3 = nn.ConvTranspose1d(base * 8, base * 4, 2, stride=2) self.dec3 = DoubleConv(base * 8, base * 4) self.up2 = nn.ConvTranspose1d(base * 4, base * 2, 2, stride=2) self.dec2 = DoubleConv(base * 4, base * 2) self.up1 = nn.ConvTranspose1d(base * 2, base, 2, stride=2) self.dec1 = DoubleConv(base * 2, base) self.head = nn.Conv1d(base, n_classes, 1) def forward(self, x): e1 = self.enc1(x) e2 = self.enc2(self.pool(e1)) e3 = self.enc3(self.pool(e2)) e4 = self.enc4(self.pool(e3)) d3 = self.dec3(torch.cat([self.up3(e4), e3], dim=1)) d2 = self.dec2(torch.cat([self.up2(d3), e2], dim=1)) d1 = self.dec1(torch.cat([self.up1(d2), e1], dim=1)) return self.head(d1)n_classes=4对应背景、P、QRS、T 四类,如果只关心 QRS 可以改成 2。base=32是通道基数,显存不够就降到 16,代价是细节分辨能力下降。输入形状是(batch, 1, 1000),输出是(batch, 4, 1000),逐点分类,不需要额外 reshape。这里有个容易忽略的点:win必须是 16 的倍数,因为四次池化每次减半,长度不能被 16 整除时,解码器拼接会因为尺寸差一而报错。1000 不是 16 的倍数,实际跑的时候要么把窗长改成 1024,要么在拼接前做裁剪,我一般直接把窗长设成 1024,省掉这个麻烦。
3.3 训练循环与损失选择
ECG 分割的类别极不均衡,背景点远多于 P 和 T,QRS 占比也不高。纯交叉熵会让网络倾向于全预测背景,所以常见做法是交叉熵加 Dice 的组合损失。
import torch.nn.functional as F def dice_loss(logits, target, n_classes=4, eps=1e-6): probs = F.softmax(logits, dim=1) onehot = F.one_hot(target, n_classes).permute(0, 2, 1).float() dims = (0, 2) inter = (probs * onehot).sum(dims) union = probs.sum(dims) + onehot.sum(dims) return 1 - ((2 * inter + eps) / (union + eps)).mean() def train_one_epoch(model, loader, optimizer, device): model.train() total = 0.0 for x, y in loader: x, y = x.to(device).unsqueeze(1), y.to(device) logits = model(x) loss = F.cross_entropy(logits, y) + dice_loss(logits, y) optimizer.zero_grad() loss.backward() optimizer.step() total += loss.item() * x.size(0) return total / len(loader.dataset)dice_loss里onehot的 permute 是把类别维挪到通道维,和probs对齐。eps防止空类别除零。组合损失里两项权重默认各 1,如果发现 QRS 召回偏低,可以把 Dice 项权重调到 2。优化器用 Adam,学习率 1e-3,配合ReduceLROnPlateau在验证损失不降时减半,这套配置在这个规模的数据上基本不用调。
3.4 推理与后处理:把逐点概率变回区间
网络输出的是每个点的类别概率,直接取 argmax 会得到一堆碎片化的标签,因为相邻窗重叠区域的预测可能不一致。后处理要做两件事:重叠窗的投票融合,以及基于生理约束的边界修正。
def merge_windows(probs_list, stride=512, win=1024, n_classes=4): # 重叠区域按概率累加,再取 argmax total_len = stride * (len(probs_list) - 1) + win acc = np.zeros((n_classes, total_len)) cnt = np.zeros(total_len) for i, p in enumerate(probs_list): s = i * stride acc[:, s:s + win] += p cnt[s:s + win] += 1 acc /= np.maximum(cnt, 1) return acc.argmax(axis=0)stride和win必须和预处理阶段一致,否则拼接位置全错。融合之后还可以加一条规则:QRS 段长度不应短于 40 ms,短于这个值的孤立片段合并到相邻类;P 波必须出现在 QRS 之前一定范围内,不满足的置为背景。这些规则不是拍脑袋,是从生理先验来的,能明显压掉毛刺。
4. 避坑与排查:跑不通时先看这几处
4.1 形状对不上:窗长不是 16 的倍数
现象是前向传播时报Sizes of tensors must match except in dimension 1。原因是四次MaxPool1d(2)之后长度变成原来的 1/16,解码器上采样回来如果原始长度不能被 16 整除,拼接时两边差一个点。解决办法是把窗长改成 1024 或 512 这类 2 的幂,或者在torch.cat前对编码器特征做切片对齐。我倾向改窗长,改结构容易引入新问题。
4.2 损失不降、全预测背景
现象是训练几个 epoch 后验证集上 P、T 的召回接近 0,混淆矩阵几乎全落在背景列。原因通常是类别不均衡加学习率偏大,网络找到了「全背景」这个局部最优。解决分两步:先把组合损失里的 Dice 权重提到 2,再把学习率降到 3e-4 重跑;如果还不行,检查标签生成规则是不是把 P、T 区间标得太窄,窄到样本里几乎没有正例。
4.3 小波重构后长度差一
现象是wavelet_denoise返回的数组比输入少一个点,后续切窗时索引错位。原因是pywt.wavedec加waverec在奇数长度信号上会有边界补齐,重构长度可能多一或少一。代码里已经用[: len(sig)]截断,但如果输入是奇数长度且用了mode="periodization"以外的模式,仍可能差一。稳妥做法是在调用前把信号补零到偶数长度,处理完再截回。
4.4 推理结果碎片化
现象是可视化时标签一段一段跳,同一个 QRS 被切成好几截。原因有两个:重叠窗融合时stride设得太大,重叠区不够;或者没做后处理,argmax 直接输出。先把stride降到窗长的一半,再补上 3.4 节的最小段长合并规则,基本能解决。如果还有零星碎片,检查归一化是不是逐窗做的,逐窗归一化会让相邻窗的幅值基准不一致,边界处预测容易跳变,改成逐记录归一化更稳。
4.5 Notebook 改了模型但训练没变
现象是改了model.py里的通道数,训练日志里的参数量却没变。原因是 Notebook 内核还持有旧的模块对象,import不会自动重载。解决办法是改完.py后重启内核再跑,或者在 Notebook 开头加%load_ext autoreload和%autoreload 2。这条不算技术坑,但踩的人最多。
5. 二次开发:换数据、换损失、换后处理的实操路径
拿到这个项目之后,最值钱的改造方向不是把网络加深,而是把标签链路和后处理做扎实。我一般按三步走。
第一步是换数据时先对齐采样率和标签语义。新数据的采样率如果不是 360 Hz,bandpass的fs、切窗的win、后处理里的最小段长全都要按比例换算。标签语义也要确认:有的数据集把 P 波起点算在 QRS 前 200 ms,有的只标峰,直接混用会让网络学到矛盾的边界。建议在01_preprocess.ipynb里加一段可视化,随机抽十个窗把信号和标签叠在一起画出来,肉眼过一遍再训练,这一步能省掉后面大量排查。
第二步是损失函数按类别做加权。如果下游只关心 QRS 检测,可以在cross_entropy里传weight=torch.tensor([0.1, 1.0, 2.0, 1.0]),把背景权重压低、QRS 权重抬高。改完之后用同一份验证集对比混淆矩阵,别只看总 loss,总 loss 降了但 QRS 召回没动的情况很常见。
第三步是后处理规则参数化。把最小段长、P 与 QRS 的最大间隔、T 波搜索窗口这些常量抽到一个配置字典里,推理时按数据源切换。下面这个结构我用了很多次,改起来不用翻代码:
POST_CFG = { "min_qrs_ms": 40, "p_before_qrs_ms": (200, 20), # P 峰应落在 QRS 起点前 20~200 ms "t_after_qrs_ms": (100, 400), "fs": 360, } def apply_rules(labels, cfg): # 按采样率把毫秒换算成点数,再做段长与位置约束 min_qrs = int(cfg["min_qrs_ms"] * cfg["fs"] / 1000) # 省略具体合并逻辑,核心是把短于 min_qrs 的 QRS 段并入相邻类 return labels验证改造是否有效,别只看整体准确率,那个指标在类别不均衡时几乎没有信息量。我习惯固定看三个数:QRS 的 F1、P 波的召回、T 波的边界误差(预测边界与标注边界的平均采样点距离)。三个数一起看,才知道改动是全面变好还是拆东墙补西墙。
最后说个习惯。这个项目里预处理、训练、推理分在三个 Notebook,参数散落在各处,改一个窗长要同步改三处,漏一处就报形状错误。从那以后我每次动这类项目,都先把窗长、步长、采样率、类别数这四个量抽到一个config.py里,三个 Notebook 统一 import,改一次全生效。看着是多写了一个文件,实际省下的排查时间远不止这些。希望帮到你。
本文还有配套的精品资源,点击获取