使用Python类操作串口时出现loop already running异常的原因及疑问

我正在尝试用Python从串口读取数据，计划后续将数据传入asyncua服务器。有C#开发背景，习惯用类封装逻辑，但运行代码时出现“loop is already running”异常。

据我理解任务会被放入事件循环执行，我认为自己遗漏了关键点。从最佳实践角度，Python中这样使用类是否合理？有没有更优实现方式？

运行后第一行数据被拆分为4个包接收，随后抛出循环已运行的异常，请问原因是什么？另外，将值抓取并存储到OPC服务器的最佳位置在哪里？

附上代码：

串口处理类代码 #

import asyncio
import serial_asyncio
from async_serial_output import OutputProtocol,InputChunkProtocol
import logging


class EboxAndino():
    """
    Class which manages the connection to the serial port
    """
    ebox = None
    timeout: int
    counter1Opc = None
    counter2Opc = None
    input1Opc = None
    input2Opc = None
    Heartbeat = None
    logger = None
    debug = False
    setup_dict = {}
    counterValueDict = {}

    hwPort = None
    baud = None
    timeout = None
    debug = False
    logger = logging.getLogger('foo')
    hBeat = False
    ser = None
    rts = False

    def __init__(self, ebox, hwPort, baud, timeout, debug, loggerName) -> None:
        """
              @param ebox: opcObj
              @param hwPort: string
              @param baud: int
              @param timeout: int
              @param debug: bool
              @param loggerName: string
              """
        # self = EboxAndino('Ebox')
        self.ebox = ebox
        self.hwPort = hwPort
        self.baud = baud
        self.timeout = timeout
        self.debug = debug
        self.hBeat = False
        self.ser = None
        self.rts = False
        print('created!')


    async def initCounter(self):
        
        self.hBeat = False
        self.counterValueDict = {"Counter1": None,
                                 "Counter2": None,
                               
                                 "Input1": None,
                                 "Input2": None,
                                }
        setup_list = [*self.counterValueDict]
       
        print ('Counter is up!')

    def get_values(self):
        if self.counterValueDict:
            return self.counterValueDict
        else:
            return None

    async def line_reader(self,loop):
        coro = await serial_asyncio.create_serial_connection(loop, OutputProtocol, '/dev/ttyAMA0', baudrate = 38400)
        await asyncio.sleep(1)
        transport, protocol = loop.run_until_complete(coro)

        await asyncio.sleep(0.5)
        loop.run_forever()
        loop.close()

主函数代码 #

async def main():

    loop = asyncio.get_event_loop()
    andino = EboxAndino('foo', '/dev/ttyAMA0', 38400, 1, True, 'foo')
    await andino.initCounter()
    
    await andino.line_reader(loop)

loop = asyncio.get_event_loop()
asyncio.sleep(2)
loop.run_until_complete(main())

1. "loop is already running" 异常原因 #

你在代码里犯了几个asyncio使用的错误：

• asyncio.sleep(2)是异步函数，不能直接在同步代码中调用，必须用await或loop.run_until_complete(asyncio.sleep(2))，且此处这个延迟完全没必要。
• 在line_reader方法中，你先await serial_asyncio.create_serial_connection(...)（已在运行的事件循环中执行），随后又调用loop.run_until_complete(coro)——但此时事件循环已经启动（main是被loop.run_until_complete触发运行的），不能在一个运行中的循环里再次调用run_until_complete。
• 同理，loop.run_forever()也不能在已运行的循环里调用，这会直接触发异常。serial_asyncio.create_serial_connection本身是异步的，创建连接后事件循环会自动处理串口读写事件，无需手动调用run_forever。

2. 类设计的合理性与优化 #

用类封装串口逻辑是合理的，符合面向对象思路，和C#开发习惯一致，但你的代码有可优化点：

• 不要定义类属性（如ebox = None、hwPort = None），这些都应该是实例属性，仅在__init__中初始化，避免多实例共享属性导致的问题。
• 串口参数（如/dev/ttyAMA0）应使用实例的self.hwPort，而非在line_reader中硬编码。
• 自定义Protocol应与类关联，方便在协议内直接更新类的counterValueDict，无需通过get_values间接获取。

优化后的类示例：

import asyncio
import serial_asyncio
import logging


class AndinoSerialProtocol(asyncio.Protocol):
    def __init__(self, andino_instance):
        self.andino = andino_instance
        self.buffer = b''

    def data_received(self, data):
        # 处理分包：维护缓冲区，直到收到完整一行数据再解析
        self.buffer += data
        while b'\n' in self.buffer:
            line, self.buffer = self.buffer.split(b'\n', 1)
            self._parse_line(line.decode().strip())

    def _parse_line(self, line):
        # 根据实际串口数据格式解析，示例假设格式为"Counter1:123"
        if ':' in line:
            key, value = line.split(':', 1)
            if key in self.andino.counterValueDict:
                self.andino.counterValueDict[key] = int(value)
                # 直接在此处更新OPC节点
                self.andino.update_opc_node(key, value)


class EboxAndino():
    def __init__(self, ebox, hwPort, baud, timeout, debug, loggerName) -> None:
        self.ebox = ebox
        self.hwPort = hwPort
        self.baud = baud
        self.timeout = timeout
        self.debug = debug
        self.logger = logging.getLogger(loggerName)
        self.hBeat = False
        self.transport = None
        self.protocol = None
        self.counterValueDict = {}
        # 提前绑定OPC节点映射（需根据实际初始化逻辑调整）
        self.counter1Opc = None
        self.counter2Opc = None
        self.input1Opc = None
        self.input2Opc = None

    async def initCounter(self):
        self.hBeat = False
        self.counterValueDict = {
            "Counter1": None,
            "Counter2": None,
            "Input1": None,
            "Input2": None,
        }
        print('Counter is up!')

    async def start_serial(self):
        # 创建串口连接，使用自定义协议关联当前实例
        self.transport, self.protocol = await serial_asyncio.create_serial_connection(
            asyncio.get_running_loop(),
            lambda: AndinoSerialProtocol(self),
            self.hwPort,
            baudrate=self.baud
        )
        self.logger.info(f"Serial connected to {self.hwPort}")

    def update_opc_node(self, key, value):
        # 调用asyncua API更新对应节点，异步操作丢到事件循环执行
        node_map = {
            "Counter1": self.counter1Opc,
            "Counter2": self.counter2Opc,
            "Input1": self.input1Opc,
            "Input2": self.input2Opc,
        }
        if key in node_map and node_map[key]:
            asyncio.create_task(node_map[key].write_value(value))

3. 数据分包问题的解决 #

串口数据是流式传输，分包是正常现象。解决办法是在自定义Protocol中维护缓冲区，每次收到数据后追加到缓冲区，直到检测到协议约定的结束符（如换行符），再解析完整的一条数据，就像上面AndinoSerialProtocol的data_received方法实现的那样。

4. OPC服务器值更新的最佳位置 #

最佳位置是串口数据解析完成后立即更新，也就是自定义Protocol的_parse_line方法内（如示例中的update_opc_node调用）。这样数据一解析完成就同步到OPC节点，延迟最低，逻辑也最清晰。

如果OPC写操作是异步的，用asyncio.create_task将任务丢到事件循环执行即可，不会阻塞串口数据的接收。

优化后的主函数 #

async def main():
    # 初始化OPC客户端/服务器（替换为实际asyncua实例初始化逻辑）
    opc_instance = None
    andino = EboxAndino(opc_instance, '/dev/ttyAMA0', 38400, 1, True, 'foo')
    await andino.initCounter()
    # 启动串口监听，事件循环会持续运行直到被终止
    await andino.start_serial()
    # 保持程序运行，可通过监听终止信号优雅退出
    await asyncio.Future()

if __name__ == "__main__":
    # Python3.7+推荐用asyncio.run()自动管理事件循环
    asyncio.run(main())

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