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

Python串口通信日志实现:如何捕获设备主动发送的意外数据?

Python串口通信捕获设备主动数据的实现方案

核心思路:常驻后台线程监听+线程安全机制解决资源冲突

不用在通信函数前后频繁启停线程,而是启动一个常驻后台线程持续监听串口,配合线程安全的锁和队列,解决主动收发与被动监听的串口资源竞争问题,同时完整捕获设备在任意时机发送的数据。

具体实现步骤

  • 初始化核心组件:用serial库打开串口,创建queue.Queue存储监听到的被动数据,用threading.Lock保证串口读写互斥,避免同时操作串口导致的异常。
  • 启动常驻监听线程:线程内循环检测串口缓冲区,一旦有数据就读取,直接记录被动日志并存入队列,全程低延迟运行。
  • 封装主动通信函数:主动收发数据时先获取锁,独占串口完成发送-接收流程,释放锁后记录主动通信日志,确保主动操作和被动监听互不干扰。
  • 日志区分标记:在日志中明确标注主动通信、被动接收的数据类型,方便后续排查问题。

示例代码片段

import serial
import threading
import queue
import time

class SerialDevice:
    def __init__(self, port, baudrate=9600):
        self.ser = serial.Serial(port, baudrate, timeout=0.1)
        self.passive_data_queue = queue.Queue()
        self.ser_lock = threading.Lock()
        # 启动后台监听线程,设为守护线程随主进程退出
        self.listen_thread = threading.Thread(target=self._listen_loop, daemon=True)
        self.listen_thread.start()

    def _listen_loop(self):
        """后台持续监听串口被动数据"""
        while True:
            if self.ser.in_waiting > 0:
                with self.ser_lock:
                    raw_data = self.ser.read_all().decode('utf-8', errors='replace')
                # 记录被动接收日志
                print(f"[被动接收] {raw_data.strip()}")
                self.passive_data_queue.put(raw_data)
            time.sleep(0.01)

    def active_comm(self, send_msg):
        """主动发送并接收响应的函数"""
        with self.ser_lock:
            # 发送数据
            self.ser.write(send_msg.encode('utf-8'))
            # 等待设备响应(可根据设备实际调整等待时间)
            time.sleep(0.2)
            resp_data = self.ser.read_all().decode('utf-8', errors='replace')
        # 记录主动通信日志
        print(f"[主动发送] {send_msg.strip()}")
        print(f"[主动响应] {resp_data.strip()}")
        return resp_data

# 测试示例
if __name__ == "__main__":
    dev = SerialDevice('COM3')
    # 主动发起一次通信
    dev.active_comm("AT+STATUS\r\n")
    # 模拟设备主动发送数据(重启告警等),监听线程会自动捕获
    time.sleep(3)
    # 从队列取出缓存的被动数据
    while not dev.passive_data_queue.empty():
        print(f"缓存被动数据: {dev.passive_data_queue.get().strip()}")

方案优势

  • 避免频繁启停线程的开销和潜在资源泄漏问题
  • 线程锁彻底解决了串口读写的资源竞争,确保主动通信和被动监听的稳定性
  • 常驻线程能完整捕获设备在重启、异常告警等非预期时机发送的所有数据
  • 队列缓存被动数据,可根据业务需求随时取出处理

注意事项

  • 串口超时时间建议设为0.1~0.5秒,平衡监听实时性和CPU占用率
  • 数据解码时使用errors='replace'或errors='ignore',避免设备发送非标准编码导致监听线程崩溃
  • 守护线程设置确保主程序退出时监听线程自动终止,无需手动清理

内容的提问来源于stack exchange,提问作者Gábor Erdős

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.07.05 19:06:23