如何在异步任务间传递serial_asyncio的Reader和Writer对象

问题背景 #

我需要在两个异步任务之间传递serial_asyncio的reader和writer对象：

• 第一个任务监听用户键盘输入，按下x时向Arduino发送信息
• 第二个任务读取Arduino的响应，并针对特定响应执行操作

原代码尝试在loop.create_task前直接调用await open_serial_connection时，出现了unexpected indent错误；而把连接逻辑放在serial函数里的话，reader和writer又无法传递给另外两个任务。

原代码如下：

import warnings
import serial
import serial.tools.list_ports
import requests
import json
import asyncio
import keyboard
import time
from serial_asyncio import open_serial_connection

with open("config.json") as config:
    cfg = json.load(config)
    config.close()

headers = {'Accept': 'application/json'}

url = f"""https://{cfg["web"]}/login?username={cfg["user"]}&password={cfg["password"]}""" 

get_token = requests.post(url, headers=headers)
token = get_token.json()
my_token = token["token"]

arduino_ports = [
    p.device
    for p in serial.tools.list_ports.comports()
    if p.manufacturer and 'Arduino' in p.manufacturer
]
if not arduino_ports:
    raise IOError("No Arduino found")
if len(arduino_ports) > 1:
    warnings.warn("Multiple arduinos, select first")


async def arduino_listen(cfg, headers, reader):
    print('arduino_listen task created')
    while True:
        await asyncio.sleep(0.1)
        info = await reader.readline()
        print('arduino respond:', info)
        if "good" in str(info):
            print('good from arduino')
        if "bad" in str(info):
            print('bad from arduino')


async def arduino_writer(cfg, headers, writer):
    print('arduino_writer task created')
    keypress = False
    key = 'x'
    while True:
        if keypress and not keyboard.is_pressed(key):
            print('x pressed on keyboard, send 5')
            writer.write(b'5')
            await asyncio.sleep(0.3)
            keypress = False
        elif keyboard.is_pressed(key) and not keypress:
            keypress = True


async def serial():
    print(f'serial task created, arduino port {arduino_ports[0]} selected')
    reader, writer = await open_serial_connection(url=arduino_ports[0], baudrate=115200)

#reader, writer = await open_serial_connection(url=arduino_ports[0], baudrate=115200)

loop = asyncio.get_event_loop()
loop.create_task(serial())
loop.create_task(arduino_listen(cfg, headers, reader))
loop.create_task(arduino_writer(cfg, headers, writer))
loop.run_forever()

解决方案 #

核心问题是不能在事件循环之外直接使用await关键字，await只能在异步函数内部执行。正确的做法是编写一个入口异步函数，先完成串口连接的建立，再启动监听和写入任务，具体实现如下：

修改后的完整代码 #

import warnings
import serial
import serial.tools.list_ports
import requests
import json
import asyncio
import keyboard
from serial_asyncio import open_serial_connection

with open("config.json") as config:
    cfg = json.load(config)

headers = {'Accept': 'application/json'}

url = f"https://{cfg['web']}/login?username={cfg['user']}&password={cfg['password']}" 
get_token = requests.post(url, headers=headers)
token = get_token.json()
my_token = token["token"]

arduino_ports = [
    p.device
    for p in serial.tools.list_ports.comports()
    if p.manufacturer and 'Arduino' in p.manufacturer
]
if not arduino_ports:
    raise IOError("No Arduino found")
if len(arduino_ports) > 1:
    warnings.warn("Multiple arduinos, select first")


async def arduino_listen(reader):
    print('arduino_listen task created')
    while True:
        # 异步等待串口数据，无需主动sleep
        info = await reader.readline()
        if not info:
            continue  # 连接断开时跳过空数据
        print('arduino respond:', info)
        info_str = info.decode('utf-8').strip()
        if "good" in info_str:
            print('good from arduino')
        elif "bad" in info_str:
            print('bad from arduino')


async def arduino_writer(writer):
    print('arduino_writer task created')
    keypress = False
    key = 'x'
    while True:
        await asyncio.sleep(0.05)  # 降低轮询频率，减少资源占用
        current_press = keyboard.is_pressed(key)
        if keypress and not current_press:
            print('x pressed on keyboard, send 5')
            writer.write(b'5')
            await writer.drain()  # 确保数据完全发送到串口
            keypress = False
        elif current_press and not keypress:
            keypress = True


async def main():
    print(f'serial connection established, arduino port {arduino_ports[0]} selected')
    # 先建立串口连接，获取reader和writer
    reader, writer = await open_serial_connection(url=arduino_ports[0], baudrate=115200)
    
    # 创建并启动两个异步任务
    listen_task = asyncio.create_task(arduino_listen(reader))
    writer_task = asyncio.create_task(arduino_writer(writer))
    
    # 等待任务持续运行（两个任务都是无限循环）
    await asyncio.gather(listen_task, writer_task)


if __name__ == "__main__":
    # Python 3.7+推荐的异步入口方式，自动管理事件循环
    asyncio.run(main())

关键说明 #

1. 使用asyncio.run(main())作为异步程序入口，替代手动创建事件循环，这是Python官方推荐的写法
2. 在main函数中先完成串口连接，再将reader和writer直接传递给两个任务，解决对象传递问题
3. 给writer.write添加await writer.drain()，确保数据真正发送到串口，避免缓冲区溢出
4. 优化arduino_listen逻辑，去掉不必要的await asyncio.sleep(0.1)，直接异步等待串口数据，更符合异步IO特性

内容的提问来源于stack exchange，提问作者Martin