从P波到T波：如何用Python+OpenCV给心电波形图做“自动体检”？

从P波到T波：Python+OpenCV心电波形智能分析实战

心电信号分析一直是医疗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.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], btype='band') 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, mode='same') # 自适应阈值检测 threshold = 0.7 * np.max(filtered) peaks, _ = signal.find_peaks(filtered, height=threshold, distance=0.6*fs) return peaks

2.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[r+int(0.2*fs) : min(len(ecg_signal), r+int(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)

波形特征参数表：

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_intervals

3.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, label='ECG Signal') # 标注特征点 plt.scatter(p_waves/fs, ecg_signal[p_waves], c='red', label='P waves') plt.scatter(r_peaks/fs, ecg_signal[r_peaks], c='green', label='R peaks') plt.scatter(t_waves/fs, ecg_signal[t_waves], c='blue', label='T 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--', alpha=0.5) plt.plot([r/fs, t/fs], [ecg_signal[r], ecg_signal[t]], 'c--', alpha=0.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(f"ST段平均偏移: {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信号。对于噪声较大的数据，可以尝试先进行小波变换去噪再进行特征检测。