基于PanTompkins心电波形R峰识别在RT-Thread+RA6M4上的实现

鸣涧

基于PanTompkins心电波形R峰识别在RT-Thread+RA6M4上的实现
作者：David528

原文链接：

https://club.rt-thread.org/ask/article/7d0c454776ea3aa3.html

应用背景目前中国心血管病患病率处于持续上升阶段，心血管病死亡率仍居首位，农村和城市心血管病分别占死因的45.91%和43.56%。心血管疾病是危害人体健康的重要疾病之一，通过穿戴式医疗监护设备监测患者日常生活状态下连续24小时或更长时间心电活动全过程，并借助计算机技术分析心电活动异常特征，是长期预防监测心血管疾病的主要手段。心电信号波形识别是心电信号分析诊断的关键,其准确性与可靠性决定诊断与治疗心血管病患者的效果。一个完整的心拍主要包括P波、QRS波群和T波等。心脏疾病发作时一般都会伴随着心电波形的变化，比如心率不齐往往伴随着QRS波群异常，因此心电信号波形识别对心血管疾病预防和诊断起着重要作用。实现功能目前心电信号波形识别主要方法是传统数学形态学方法，在形态学方法中，R峰的识别是其他波形识别的基础，本文是基于PanTompkins 算法的开源代码进行移植和改造，对开源代码增加了适用于RT-Thread系统的生产者与消费者模型的R峰识别技术实现。根据采集到的心电图数据，实时绘制心电波形和识别到R峰，在下图示例中绘制了心电波形图形（白色）和R峰识别的标注位置（红色竖线）。

实现步骤1、 新建项目

使用RT-Thread studio 新建一个基于开发板的CPK-RA6M4 的一个RT-Thread 项目。

2、 关闭系统控制台和Shell串口的输入输出：

由于本人手头硬件资源少，只有一个串口（USB 转ttl ）转换器。因此需要关闭系统控制台和Shell串口的输入输出，以便独立使用这个串口进行ECG数据的输入和计算结果的输出。

在配置头文件rtconfig.h 中，关闭如下配置：

• 关闭控制台串口输出：
• //#define RT_USING_CONSOLE
• 关闭Shell功能：
• //#define RT_USING_FINSH
  
  

3、 硬件接入

主要就是一个主卡和USB 转ttl 串口转换器，接入方法比较简单。如图：

4、 对开源PanTompkins 算法增加生产者和消费者的支持

（1）首先使用RT-Thread的rt_sem_init方法初始化生产者与消费者和串口接收所需要的信号量：

• // 初始化生产者与消费者使用的信号量
• rt_sem_init(&sem_lock, "lock",     1,      RT_IPC_FLAG_PRIO);
• rt_sem_init(&sem_empty, "empty",   MAXSEM, RT_IPC_FLAG_PRIO);
• rt_sem_init(&sem_full, "full",     0,      RT_IPC_FLAG_PRIO);
• 
  
• // 初始化串口接收使用的信号量
• rt_sem_init(&rx_sem, "rx_sem", 0, RT_IPC_FLAG_FIFO);
  
  

（2）使用RT-Thread的rt_device_find和rt_device_open方法打开设备名称uart7的串口

• serial = rt_device_find(SAMPLE_UART_NAME);
• if (!serial)
• {
•       rt_kprintf("find %s failed!\n", SAMPLE_UART_NAME);
• }
• 
  
• res = rt_device_open(serial, RT_DEVICE_FLAG_INT_RX);
• if (res == RT_EOK)
• {
•     rt_device_set_rx_indicate(serial, uart_input);
• }
  
  

（3）生产者的实现主要源码片段

该部分是基于串口接收，做了无限循环接收串口数据，由于模拟发到板卡的每一条心电数据都包含了\n，所以使用\n 作为每条心电数据结束标志。接收到一条心电数据后就放到消费buffer中。

•   while (1)
•     {
•         dataType data = 0;
•         char ch;
•         char str[10];
•         int i = 0;
• 
  
•         while (1)
•         {
•             if (rt_device_read(serial, -1, &ch, 1) != 1) {
•                 rt_sem_take(&rx_sem, RT_WAITING_FOREVER);
•                 continue;
•             }
•             if (ch != '\n') {
•                 str = ch;
•                 if ( i < (sizeof(str) - 1))
•                     i ++;
•             }
•             else {
•                 data = atoi(str);
•                 break;
•             }
•         }
• 
  
•         rt_sem_take(&sem_empty, RT_WAITING_FOREVER);
• 
  
•         rt_sem_take(&sem_lock, RT_WAITING_FOREVER);
• 
  
•         pc_buffer[pc_in] = data;
•         pc_in = (pc_in + 1) % MAXSEM;
• 
  
•         rt_sem_release(&sem_lock);
• 
  
•         rt_sem_release(&sem_full);
•     }
  
  

（4）消费者的实现主要源码片段
这部分就是经典教科书消费者实现代码，不做解释了。该部分代码是PanTompkins 算法的数据输入实现。

•     rt_sem_take(&sem_full, RT_WAITING_FOREVER);
• 
  
•     rt_sem_take(&sem_lock, RT_WAITING_FOREVER);
• 
  
•     num = pc_buffer[pc_out];
•     pc_out = (pc_out + 1) % MAXSEM;
• 
  
•     rt_sem_release(&sem_lock);
•     rt_sem_release(&sem_empty);
  
  

（5）最后使用RT-Thread线程创建方法rt_thread_create、rt_thread_startup创建和启动两个线程，一个是生产者一个是消费者。

•     tid = rt_thread_create("thread1",
•                             producer_thread_entry, (void*)0,
•                             THREAD_STACK_SIZE,
•                             THREAD_PRIORITY, THREAD_TIMESLICE);
•     if (tid != RT_NULL)
•         rt_thread_startup(tid);
• 
  
•     tid = rt_thread_create("thread2",
•                             consumer_thread_entry, (void*)0,
•                             THREAD_STACK_SIZE,
•                             THREAD_PRIORITY, THREAD_TIMESLICE);
•     if (tid != RT_NULL)
•         rt_thread_startup(tid);
  
  

5、 Python实现ECG数据模拟输入和ECG绘制。

在心电图数据模拟输入和绘制实现部分，在主入口函数部分，首先开启一个串口COM3，然后创建并启动两个线程，一个是发送模拟的ECG数据，一个是接收ECG数据和识别结果。在绘制ECG波形和R峰（红竖线）源码中，使用了QTimer及pyqtgraph绘制的ECG图形。全部实现的Python源码：

• import serial
• import threading
• import time
• import pyqtgraph as pg
• 
  
• # ECG 频率
• FS = 360
• 
  
• #通过串口向板卡模拟发送ECG数据
• def sendECGData(ser):
•     with open('test_input.txt', 'r') as f:
•         for s in f.readlines():
•             ser.write(s.encode())
•             time.sleep(1.0/FS)
• 
  
• #存储5秒内的ECG数据及识别结果
• ay = []
• def recvECGData(ser):
•     global ay
•     len = 5 * FS - 1
•     while True:
•         if ser.in_waiting:
•             str = ser.readline(ser.in_waiting).decode()
•             str = str.replace("\n","")
•             print(str)
•             ay = ay[-len:]
•             ay.append(str)
• 
  
• #绘制ECG波形和R峰（红竖线）
• p1 = None
• def plotData():
•     data = []
•     i = -22
•     pos = []
•     for str in ay:
•         array = str.split(',')
•         signal = int(array[0])
•         R = int(array[1])
•         i = i + 1
•         if R == 1:
•             pos.append(i)
•         data.append(signal)
•     p1.clear()
•     p1.plot(data)
•     for p in pos:
•         p1.addLine(x=p, pen = 'r')
• 
  
• #使用QTimer及pyqtgraph绘制ECG图形
• def drawECGData():
•     app = pg.QtGui.QApplication([])
•     view = pg.GraphicsView()
•     l = pg.GraphicsLayout()
•     view.setCentralItem(l)
•     view.show()
•     global p1
•     p1 = l.addPlot(title='绘制ECG图形')
• 
  
•     timer = pg.QtCore.QTimer()
•     timer.timeout.connect(plotData)
•     timer.start(1000)
•     app.exec_()
• 
  
• #主入口函数
• if __name__ == '__main__':
•     #开启串口
•     ser = serial.Serial('COM3', 115200, timeout=0.01)
• 
  
•     #开启两个线程，一个是发送模拟的ECG数据，一个是接收ECG数据和识别结果
•     t1 = threading.Thread(target=sendECGData, args=(ser,))
•     t2 = threading.Thread(target=recvECGData, args=(ser,))
•     t1.start()
•     t2.start()
• 
  
•     #主线程实时绘制ECG图形
•     drawECGData()
  
  

项目源码

https://gitee.com/david528/rt-thread-pan-tompkins.git

参考资料

1、 PanTompkins 开源实现：

https://github.com/rafaelmmoreira/PanTompkinsQRS
2、 基于 RT-Thread Studio的CPK-RA6M4 开发环境搭建指南：
https://mp.weixin.qq.com/s/phEV5jGjTOoe7Y0ihI6ftg