You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

拔插串口设备时自定义类未更新pyserial连接状态问题求助

问题根源

  • 你的handleconnect重连逻辑仅在self.ser is None时才会触发重连,但串口拔出后self.ser对象仍然存在,只是底层连接已经失效,不会自动变成None,所以永远不会触发重连逻辑
  • 读写操作触发异常后,你只是打印了错误提示,没有将失效的self.ser重置为None,也没有关闭无效连接,导致后续操作一直使用失效的串口对象
  • handleconnect里新建串口实例后先调用close()再open()是冗余操作,serial.Serial初始化时默认就会打开端口
  • 初始化方法里直接创建串口实例没有捕获异常,程序启动时如果串口没插就会直接崩溃

修正后可实现自动重连的代码

import serial
import uuid
import time


class SerialWrapper:
    # 把端口和波特率做成参数,后续改配置更方便
    def __init__(self, port="COM3", baudrate=9600):
        self.port = port
        self.baudrate = baudrate
        self.ser = None
        # 初始化的时候先尝试连接一次
        self.handleconnect()

    def handleconnect(self):
        # 已经正常连接的话就跳过,不用重复操作
        if self.ser is not None and self.ser.is_open:
            return
        try:
            print("尝试连接串口...")
            self.ser = serial.Serial(self.port, self.baudrate, timeout=2)
            print(f"串口{self.port}连接成功")
        except Exception as e:
            # 连接失败就重置ser为None
            if self.ser is not None:
                try:
                    self.ser.close()
                except:
                    pass
            self.ser = None
            print(f"串口连接失败: {str(e)}")

    def serialwrite(self, data):
        self.handleconnect()
        if self.ser is None:
            print("串口未连接,写入失败")
            return 0
        try:
            datalength = self.ser.write(data.encode())
            print(f"成功写入{str(datalength)}字节数据")
            return datalength
        except Exception as e:
            print(f"写入数据出错: {str(e)}")
            # 出错就重置串口对象,下次触发重连
            if self.ser is not None:
                try:
                    self.ser.close()
                except:
                    pass
            self.ser = None
            return 0

    def serialread(self, datalength):
        self.handleconnect()
        if self.ser is None:
            print("串口未连接,读取失败")
            return ""
        try:
            data = self.ser.read(datalength).decode()
            print(f"成功读取{str(len(data))}字节数据")
            return data
        except Exception as e:
            print(f"读取数据出错: {str(e)}")
            # 出错就重置串口对象,下次触发重连
            if self.ser is not None:
                try:
                    self.ser.close()
                except:
                    pass
            self.ser = None
            return ""


def main():
    ser = SerialWrapper()
    while 1:
        value = uuid.uuid4().hex
        length = ser.serialwrite(value)
        if length > 0:
            print(ser.serialread(length))
        # 加个延迟避免占用过高CPU
        time.sleep(1)


if __name__ == "__main__":
    main()

内容的提问来源于stack exchange,提问作者Tomáš Kretek

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.27 04:06:04