通过单个串行链路，如何每秒读取连续数据（轮询），并同时进行其他查询？

Question

我通过串行链路 RS232 与硬件设备通信（没有其他可用选项）。

• 我需要每秒持续轮询一些数据“foo”（如果可能的话）。
• 有时我还需要向设备询问其他数据，例如“bar”和“baz”

当然，由于它是单个串行链路（并且没有像 HTTP 这样的高级接口），我不能仅通过使用 2 个不同的线程同时执行这两项操作：检查是否没有同时发送并发数据非常重要通过同一条电线的时间。

这是我当前的解决方案：

class Device:
    def __init__(self):
        self.serial_port = serial.Serial(port="COM1", baudrate=9600, parity="N", stopbits=1)
        self.start_foo_polling()

    def start_foo_polling(self):
        self.foo_polling = True
        threading.Thread(target=self.get_foo_thread).start()

    def get_foo_thread(self):
        while self.foo_polling:
            self.serial_port.write(b"QUERY_FOO")
            data = self.serial_port.read(8)    # then, do something with data!
            time.sleep(1)
       
    def stop_foo_polling(self):
        self.foo_polling = False

    def query_bar(self):
        rerun_foo_polling = False                             # *
        if self.foo_polling:                                  # *
            rerun_foo_polling = True                          # *
            self.stop_foo_polling()                           # *
            time.sleep(1)   # wait for the thread to finish   # *   
        self.serial_port.write(b"QUERY_BAR")
        data = self.serial_port.read(8)
        if rerun_foo_polling:                                 # *
            self.start_foo_polling()                          # *

d = Device()
time.sleep(3.9)
d.query_bar()

但它需要对

def query_baz(self): ...

的行 (*) 进行一些代码重复，更一般地说，它似乎不是最佳的。

处理这个并发问题的标准串行链路解决方案是什么？

Answer 1

使用状态机来管理不同的通信任务。使用Python的asyncio模块：

    import asyncio
import serial

class Device:
    def __init__(self):
        self.serial_port = serial.Serial(port="COM1", baudrate=9600, parity="N", stopbits=1)
        self.current_task = None

    async def get_foo(self):
        while True:
            self.serial_port.write(b"QUERY_FOO")
            data = self.serial_port.read(8)  # Process 'foo' data here
            await asyncio.sleep(1)

    async def query_data(self, query_cmd):
        if self.current_task:
            self.current_task.cancel()
            await asyncio.sleep(1)

        self.serial_port.write(query_cmd)
        data = self.serial_port.read(8)  # Process data for 'bar' or 'baz'

        self.current_task = asyncio.create_task(self.get_foo())

# Usage
async def main():
    d = Device()
    await asyncio.gather(d.get_foo(), d.query_data(b"QUERY_BAR"))

asyncio.run(main())

此代码使用 asyncio 来处理异步任务。 get_foo 方法不断轮询“foo”数据，query_data 方法通过取消正在进行的任务（如果有）然后启动新查询来处理其他查询（“bar”或“baz”）。