Python matplotlib 中实时心电图信号的 R 峰值检测

Question

我已经对 Raspberry Pi 3 进行了编程来绘制心电图信号。我在Python2.7中使用了matplotlib库。 Arduino 板已用于获取模拟输入。

创建的 GUI 包含 2 个子图，用于从 arduino 的 2 个模拟引脚读取信号。

现在我需要从实时心电图信号中检测 R 峰值。

我尝试了几种代码来检测，但无法在同一个图中检测到它。 请建议我如何在实时心电图信号绘图的下面的代码中添加 r 峰值检测代码。

这是我的代码：-

import serial
import numpy as np
from matplotlib import pyplot as plt
from matplotlib import animation

ser = serial.Serial('/dev/ttyACM0', 115200)

n = 200
fig = plt.figure(figsize=(12,9))

ax1 = fig.add_subplot(2,1,1)
ax2 = fig.add_subplot(2,1,2)


ch1, = ax1.plot([], [], 'b', label = 'Channel 1')
ch2, = ax2.plot([], [], 'r', label = 'Channel 2')

axes = [ax1, ax2]

for ax in axes:
    ax.set_xlim(0, n+1)
    ax.set_ylim(0, n-1)
    ax.set_ylabel('values')
    ax.legend(loc='upper right')
    ax.grid(True)

ax1.set_title('Real-time ECG plot')
ax2.set_xlabel('Values')

t = list(range(0,n))
channel1 = [0] * n
channel2 = [0] * n


def init():
    ch1.set_data([], [])
    ch2.set_data([], [])

    return ch1, ch2,

def animate(i):

    while (ser.inWaiting() == 0):
        pass
    arduinoString = ser.readline()
    dataArray = arduinoString.split(',')

    channel1.append(float(dataArray[0]))
    channel2.append(float(dataArray[1]))

    channel1.pop(0)
    channel2.pop(0)

    ch1.set_data(t, channel1)
    ch2.set_data(t, channel2)



    return ch1, ch2, ch3,

delay = 0
anim = animation.FuncAnimation(fig, animate, init_func=init,interval=delay, blit=True)

plt.show()

ser.close()

Answer 1

要检测实时心电图上的 R 峰值，我认为更快的版本应该是简单地使用 scipy 中的 find_peaks，它可以顺便找到所有峰值。 这可以通过 biospy 的 api 进一步修改。

import matplotlib.pyplot as plt
import numpy as np
from scipy.signal import find_peaks


# Function for real-time peak detection
def detect_peaks(ecg_signal):
    peaks, _ = find_peaks(ecg_signal, height=0.3, distance=100)
    return peaks

# Main loop for real-time ECG processing
def main():
    # Set up time vector (replace this with your actual time vector)
    fs = 1000  # Sample rate (Hz)
    duration = 10  # Duration of the ECG signal (seconds)
    t = np.arange(0, duration, 1/fs)

    plt.ion()  # Turn on interactive mode for real-time plotting

    # Create a plot for real-time visualization
    fig, ax = plt.subplots()

    try:
        for i in range(1000):  # Replace with an appropriate number of iterations or use a while loop
            ecg_data = get_ecg_data()
            peaks = detect_peaks(ecg_data)

            # Plot ECG signal
            ax.clear()
            ax.plot(t, ecg_data, label='ECG Signal')
            
            # Mark detected peaks
            ax.plot(t[peaks], ecg_data[peaks], 'rx', label='Detected Peaks')

            ax.legend()
            ax.set_title('Real-time ECG with Peak Detection')
            ax.set_xlabel('Time (s)')
            ax.set_ylabel('Amplitude')

            plt.pause(0.1)  # Pause to update the plot

    except KeyboardInterrupt:
        print("Real-time processing interrupted.")

    plt.ioff()  # Turn off interactive mode
    plt.show()

if __name__ == "__main__":
    main()