绘制实时串行数据最有效的 Python IPC 机制是什么？

Question

从串行端口读取数据到绘制该数据的单独进程的最快 Python 机制是什么？

我正在实时绘制从串行端口读取的脑电图数据。串口读取和数据包解包代码工作正常，如果我读取并存储数据，然后绘制存储的数据，它看起来很棒。像这样：

注意：设备会生成测试正弦波进行调试

我正在使用 pyQtGraph 进行绘图。在读取串行数据的同一过程中更新绘图不是一种选择，因为串行 read() 调用之间的轻微延迟会导致串行缓冲区溢出并随之而来的是错误的校验和。 pyQtGraph 提供了在单独的进程上渲染图形的规定，这很棒，但瓶颈似乎在进程间通信中。我尝试了 Pipe() 和 Queue() 的各种配置，所有这些都会导致图形更新滞后、闪烁。到目前为止，从串口获取新值到图形的最流畅、最一致的方法似乎是通过共享内存，如下所示：

from pyqtgraph.Qt import QtGui
import pyqtgraph as pg
from multiprocessing import Process, Array, Value, Pipe
from serial_interface import EEG64Board
from collections import deque

def serialLoop(arr):
    eeg = EEG64Board(port='/dev/ttyACM0')
    eeg.openSerial() 
    eeg.sendTest('1')        #Tells the eeg device to start sending data
    while True:
        data = eeg.readEEG() #Returns an array of the 8 latest values, one per channel
        if data != False:    #Returns False if bad checksum
            val.value = data[7] 

val = Value('d',0.0)
q = deque([],500)

def graphLoop():
    global val,q
    plt = pg.plot(q)
    while True:
        q.append(val.value)
        plt.plot(q,clear=True)
        QtGui.QApplication.processEvents()

serial_proc = Process(target=serialLoop, args=(val,), name='serial_proc')
serial_proc.start()

try:
    while True:
        graphLoop()
except KeyboardInterrupt:
    print('interrupted')

上面的代码通过简单地拉取serialLoop记录的最新值并将其附加到一个双端队列来执行实时绘图。虽然情节更新顺利，但它只抓取了大约四分之一的值，如结果图中所示：

那么，你会推荐什么样的多进程或线程结构，那么它们之间应该使用什么形式的IPC呢？

更新：

我每秒接收 2,000 个样本。我在想，如果我以 100 fps 更新显示并每帧添加 20 个新样本，那么我应该会很好。实现这一点的最佳 Python 多线程机制是什么？

Answer 1

这可能不是最有效的，但以下代码实现了 100 fps 的一个绘图或 20 fps 的 8 个绘图。这个想法很简单：共享一个数组、索引和锁。串行填充数组并增加索引，而 is 已锁定，绘图过程会定期从数组中获取所有新值并减少索引，再次处于锁定状态。

from pyqtgraph.Qt import QtGui
import pyqtgraph as pg
from multiprocessing import Process, Array, Value, Lock
from serial_interface import EEG64Board
from collections import deque

def serialLoop(arr,idx,lock):
    eeg = EEG64Board(port='/dev/ttyACM0')
    eeg.openSerial() 
    eeg.sendTest('1')        #Tells the eeg device to start sending data
    while True:
        data = eeg.readEEG() #Returns an array of the 8 latest values, one per channel
        if data != False:    #Returns False if bad checksum
            lock.acquire()
            for i in range(8):
                arr[i][idx.value] = data[i] 
            idx.value += 1
            lock.release()
    eeg.sendTest('2') 

arr = [Array('d',range(1024)) for i in range(8)]
idx = Value('i', 0)
q = [deque([],500) for i in range(8)]
iq = deque([],500)
lock = Lock()

lastUpdate = pg.ptime.time()
avgFps = 0.0

def graphLoop():
    global val,q,lock,arr,iq, lastUpdate, avgFps
    win = pg.GraphicsWindow()
    plt = list()
    for i in range(8):
        plt += [win.addPlot(row=(i+1), col=0, colspan=3)]
    #iplt = pg.plot(iq)
    counter = 0
    while True:
        lock.acquire()
        #time.sleep(.01)
        for i in range(idx.value):
            for j in range(8):
                q[j].append(arr[j][i])        
        idx.value = 0
        lock.release()
        for i in range(8):
            plt[i].plot(q[i],clear=True)
        QtGui.QApplication.processEvents()
        counter += 1

        now = pg.ptime.time()
        fps = 1.0 / (now - lastUpdate)
        lastUpdate = now
        avgFps = avgFps * 0.8 + fps * 0.2

serial_proc = Process(target=serialLoop, args=(arr,idx,lock), name='serial_proc')
serial_proc.start()

graphLoop()

serial_proc.terminate()