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
相关产品推荐
相关产品推荐

