从P波到T波PythonOpenCV心电波形智能分析实战心电信号分析一直是医疗AI领域的热门方向。想象一下当你拿到一份心电图数据时能否用代码自动识别出P波、QRS波群和T波能否通过算法判断ST段是否异常本文将带你用Python构建一套完整的心电波形分析流水线从原始信号处理到特征点检测再到异常初筛算法实现。1. 心电信号预处理从噪声中提取有效信息原始心电信号往往包含各种干扰预处理是后续分析的基础。我们首先需要加载数据并进行初步清洗import numpy as np import pandas as pd from scipy import signal # 加载CSV格式的心电数据 def load_ecg_data(filepath): df pd.read_csv(filepath) ecg_signal df[voltage].values # 假设电压数据存储在voltage列 sampling_rate 1000 # 采样率(Hz)根据实际设备调整 return ecg_signal, sampling_rate常见噪声类型及处理方法噪声类型来源处理方法基线漂移呼吸运动、电极移动高通滤波(0.5Hz)工频干扰50/60Hz电源干扰陷波滤波器肌电噪声肌肉活动低通滤波(40Hz)运动伪影身体移动小波变换去噪# 带通滤波示例 (0.5-40Hz) def bandpass_filter(ecg_signal, fs): nyq 0.5 * fs low 0.5 / nyq high 40.0 / nyq b, a signal.butter(4, [low, high], btypeband) return signal.filtfilt(b, a, ecg_signal)提示滤波参数需要根据实际信号特性调整过度滤波可能导致波形失真2. 特征波检测定位PQRST关键点2.1 R波峰值检测R波振幅最高是最容易识别的特征点。我们使用Pan-Tompkins算法的改进版本def detect_r_peaks(ecg_signal, fs): # 微分 diff np.diff(ecg_signal) # 平方运算增强R波 squared diff ** 2 # 移动平均滤波 window_size int(0.15 * fs) ma_filter np.ones(window_size) / window_size filtered np.convolve(squared, ma_filter, modesame) # 自适应阈值检测 threshold 0.7 * np.max(filtered) peaks, _ signal.find_peaks(filtered, heightthreshold, distance0.6*fs) return peaks2.2 P波和T波检测在R波定位基础上我们可以搜索前后特定时间窗口内的其他特征波def detect_p_t_waves(ecg_signal, r_peaks, fs): p_waves [] t_waves [] for r in r_peaks: # P波搜索窗口(R波前0.2-0.4秒) p_window ecg_signal[max(0, r-int(0.4*fs)) : r-int(0.2*fs)] if len(p_window) 0: p_peak np.argmax(p_window) (r - int(0.4*fs)) p_waves.append(p_peak) # T波搜索窗口(R波后0.2-0.4秒) t_window ecg_signal[rint(0.2*fs) : min(len(ecg_signal), rint(0.4*fs))] if len(t_window) 0: t_peak np.argmax(t_window) (r int(0.2*fs)) t_waves.append(t_peak) return np.array(p_waves), np.array(t_waves)波形特征参数表波形持续时间(s)振幅(mV)形态特征P波0.110-0.3圆钝正向波QRS0.100.5-3.0陡峭多向波T波0.250.1-0.5平缓正向波3. 间期分析与异常检测3.1 PR间期与QT间期计算def calculate_intervals(p_waves, r_peaks, t_waves, fs): # 确保P波和T波数量与R波匹配 min_len min(len(p_waves), len(r_peaks), len(t_waves)) p_waves p_waves[:min_len] r_peaks r_peaks[:min_len] t_waves t_waves[:min_len] # PR间期(P波起点到QRS起点) pr_intervals (r_peaks - p_waves) / fs # QT间期(QRS起点到T波终点) qt_intervals (t_waves - r_peaks) / fs return pr_intervals, qt_intervals3.2 ST段异常检测ST段抬高或压低是心肌缺血的重要指标def detect_st_abnormalities(ecg_signal, r_peaks, t_waves, fs): st_segments [] is_st_elevated [] for r, t in zip(r_peaks, t_waves): # J点(R波结束后约0.04秒) j_point r int(0.04 * fs) # ST段中点(J点后0.06-0.08秒) st_point j_point int(0.07 * fs) # 计算ST段偏移量(相对于TP基线) baseline np.mean(ecg_signal[max(0, t-int(0.2*fs)):t]) st_level ecg_signal[st_point] - baseline st_segments.append(st_level) is_st_elevated.append(abs(st_level) 0.1) # 阈值0.1mV return np.array(st_segments), np.array(is_st_elevated)常见异常模式判断规则心动过速RR间期0.6秒(心率100bpm)心动过缓RR间期1.0秒(心率60bpm)PR间期延长0.20秒(一度房室传导阻滞)ST段抬高0.1mV(可能心肌梗死)ST段压低-0.05mV(可能心肌缺血)4. 可视化与结果输出4.1 波形标注可视化使用Matplotlib标注检测到的特征点import matplotlib.pyplot as plt def plot_annotated_ecg(ecg_signal, p_waves, r_peaks, t_waves, fs): plt.figure(figsize(15, 5)) time np.arange(len(ecg_signal)) / fs plt.plot(time, ecg_signal, labelECG Signal) # 标注特征点 plt.scatter(p_waves/fs, ecg_signal[p_waves], cred, labelP waves) plt.scatter(r_peaks/fs, ecg_signal[r_peaks], cgreen, labelR peaks) plt.scatter(t_waves/fs, ecg_signal[t_waves], cblue, labelT waves) # 标注间期 for p, r, t in zip(p_waves, r_peaks, t_waves): plt.plot([p/fs, r/fs], [ecg_signal[p], ecg_signal[r]], m--, alpha0.5) plt.plot([r/fs, t/fs], [ecg_signal[r], ecg_signal[t]], c--, alpha0.5) plt.xlabel(Time (s)) plt.ylabel(Amplitude (mV)) plt.legend() plt.grid() plt.show()4.2 生成诊断报告def generate_report(pr_intervals, qt_intervals, st_levels, is_st_elevated): report { average_pr: np.mean(pr_intervals), average_qt: np.mean(qt_intervals), st_deviation: np.mean(st_levels), abnormal_beats: np.sum(is_st_elevated), heart_rate: 60/np.mean(np.diff(r_peaks)/fs) if len(r_peaks)1 else 0 } print( ECG分析报告 ) print(f平均心率: {report[heart_rate]:.1f} bpm) print(f平均PR间期: {report[average_pr]:.3f} s) print(f平均QT间期: {report[average_qt]:.3f} s) print(fST段平均偏移: {report[st_deviation]:.3f} mV) print(f异常节律数: {report[abnormal_beats]}) if report[heart_rate] 100: print(警告: 检测到心动过速) elif report[heart_rate] 60: print(警告: 检测到心动过缓) if report[average_pr] 0.2: print(警告: PR间期延长可能一度房室传导阻滞) if report[st_deviation] 0.1: print(警告: ST段抬高建议进一步检查心肌梗死可能) elif report[st_deviation] -0.05: print(警告: ST段压低可能心肌缺血) return report在实际项目中我发现R波检测的准确率直接影响整个分析流程的质量。通过调整移动平均窗口大小和阈值参数可以使算法适应不同质量的ECG信号。对于噪声较大的数据可以尝试先进行小波变换去噪再进行特征检测。