使用 asyncio 读取串口的输出答案

【问题标题】：Using asyncio to read the output of a serial port使用 asyncio 读取串口的输出
【发布时间】：2015-12-01 14:26:56
【问题描述】：

我一直在做一些端口读取工作，我需要知道是否应该使用 asyncIO 来读取来自该端口的数据。

这里有一些关于系统构建方式的细节。

同时读取多个传感器并可能产生同时进入的输出。（所有数据都通过连接到 USB 端口的探针 ZU10 进来）所有数据一进来就必须打上时间戳。
数据应该经过处理，然后通过 REST API 发送到 django webapp。

问题是，不丢失任何数据真的很重要，我问这个问题是因为我相信一定有比我想的更简单的方法来做到这一点。

我希望数据进入的方式是通过一个异步过程将数据放入队列并为其加上时间戳，这样就不会出现数据丢失，并且时间戳可能不会超过几分之一秒，这不是问题。

如果有人有任何想法，我会很感激他们

这是我用来打开端口、接收数据以及到目前为止我在实际读取有意义的数据时所获得的代码。

阅读部分如下：

import serial #for port opening
import sys #for exceptions

#

#configure the serial connections (the parameters differs on the device you are connecting to)

class Serializer: 
    def __init__(self, port, baudrate=9600, timeout=.1): 
        self.port = serial.Serial(port = port, baudrate=baudrate, 
        timeout=timeout, writeTimeout=timeout)

    def open(self): 
        ''' Abre Puerto Serial'''
        self.port.open()

    def close(self): 
        ''' Cierra Puerto Serial'''
        self.port.close() 

    def send(self, msg):
        ''' envía mensaje a dispositivo serial'''
        self.port.write(msg)

    def recv(self):
        ''' lee salidas del dispositivo serial '''
        return self.port.readline()

PORT = '/dev/ttyUSB0' #Esto puede necesitar cambiarse


# test main class made for testing
def main():
    test_port = Serializer(port = PORT)

    while True:
        print(test_port.recv())

if __name__ == "__main__":
    main()

我将使用一些东西来过滤掉有意义的读取（忍受它，它可能充满了可怕的错误，也许还有可怕的正则表达式）：

import re
from Lector import ChecaPuertos
from Lector import entrada


patterns = [r'^{5}[0-9],2[0-9a-fA-F] $'] #pattern list

class IterPat:
    def __init__(self, lect, pat = patterns):
        self.pat = pat  # lista de patrones posibles para sensores
        self.lect = lect  # lectura siendo analizada
        self.patLen = len(pat)  #Largo de patrones

    def __iter__(self):
        self.iteracion = 0 #crea la variable a iterar.

    def __next__(self):
        '''
        Primero revisa si ya pasamos por todas las iteraciones posibles
        luego revisa si la iteración es la que pensabamos, de ser así regresa una
        tupla con el patrón correspondiente, y la lectura
        de otra forma para el valor de ser mostrado
        '''
        pattern = re.compile(self.pat[self.iteracion])
        comp = pattern.match(self.lect)
        if comp == True:
            re_value = (self.pattern, self.lect)
            return re_value
        else:
            self.iteración += 1

def main():
    puerto = ChecaPuertos.serial_ports()
    serial = entrada.Serializer(port = puerto[0])

    if serial != open:
        serial.open()
    while True:
        iter = IterPat()

     #This is incomplete right here.

【问题讨论】：

你当然可以使用asyncio从串口读取数据，但我认为你不能使用pyserial。见here。不过，我不认为使用 asyncio 会让您的任务更轻松或更安全（就不丢失数据而言）。
在 9600 波特率下，每毫秒最多 1 个字节，轮询应该不会太糟糕（如果可能不优雅）。串行端口通常也有缓冲区，如果你想在另一个线程/进程中进行阻塞读取，你可以实现自己的（任意大小）。
串行端口缓冲区是自动实现的还是我应该查找它们？
正如 Nick T 所说，你应该设置一个专门用于阅读的线程，并且只使用 PySerial。数据是流式传输的吗？延迟是否重要？如果答案是肯定的和否定的，那么最简单的做法是读取一个固定大小的块并立即将其转储到队列中。否则答案可能是询问有多少个字符可用，然后读入 max(avail, 1) （如果没有可用数据将阻塞），然后将可变大小的字符串放入队列中。
另外，如果您对发送设备有任何控制权，您可以添加简单的错误检测，或者通过前向纠错和/或重传来发疯。

标签： python python-3.x pyserial python-asyncio

【解决方案1】：

我正在使用 asyncio 通过 pyserial 读取/写入串行端口。我让串行连接另一端的设备在准备好接收有效负载时写入一个字节。 Asyncio 监视该字节，然后发送有效负载。它看起来像这样：

serial_f = serial.Serial(port=dev, baudrate=BAUDRATE, timeout=2)

def write_serial():
    status = serial_f.read(1)
    serial_f.write(buffer)

loop = asyncio.get_event_loop()
loop.add_reader(serial_f.fileno(), write_serial)

【讨论】：