4、串口通信脚本：pySerial库详解、自定义串口协议解析器、多串口并发测试

串口通信，在嵌入式测试里可以说是最基础也最常用的手段了。我做了这么多年测试，跟串口打交道的次数，比跟家人吃饭的次数还多（笑）。今天咱们就聊聊怎么用 Python 的 pySerial 库，把串口测试玩出花来。

4.1 pySerial 库：你的串口瑞士军刀

pySerial 这个库，说白了就是 Python 操作串口的标准答案。它封装了底层操作系统的串口 API，让你不用关心 Windows 和 Linux 的差异。我个人习惯在项目一开始就把 pySerial 装好，因为后面调试基本离不开它。

pip install pyserial

嗯，这里要注意，别装错了。有个叫 serial 的库是旧的，pySerial 才是正主。我刚开始用的时候也踩过这个坑。

4.1.1 基本操作：打开、读写、关闭

先看一个最基础的例子：

import serial

# 打开串口
ser = serial.Serial(
    port='COM3',        # Windows 下用 COMx，Linux 下用 /dev/ttyUSB0
    baudrate=115200,    # 波特率，常见的有 9600、115200
    bytesize=8,         # 数据位，一般是 8
    parity='N',         # 校验位：N 无校验，E 偶校验，O 奇校验
    stopbits=1,         # 停止位，一般是 1
    timeout=1           # 超时时间，单位秒
)

# 发送数据
ser.write(b'AT\r\n')

# 读取数据
response = ser.read(100)  # 读取 100 字节
# 或者按行读取
line = ser.readline()     # 读到换行符为止

# 关闭串口
ser.close()

这里有个小细节：timeout 参数我建议一定要设。不设的话，read() 会一直阻塞，程序就卡死了。我在项目中遇到过测试脚本跑了一整夜，结果因为串口没数据，第二天早上发现脚本还卡在第一行——那叫一个尴尬。

4.1.2 常用方法与属性

if ser.in_waiting > 0:
    data = ser.read(ser.in_waiting)

4.2 自定义串口协议解析器

实际项目中，很少有设备直接用纯文本通信。大多数都是自定义的二进制协议。你想想看，如果每次都要手动解析字节流，那不得累死？所以我们需要一个协议解析器。

4.2.1 协议帧结构设计

我见过最常用的协议帧结构是这样的：

| 帧头(2B) | 长度(1B) | 命令(1B) | 数据(NB) | 校验(1B) | 帧尾(1B) |
| 0xAA 0x55 |   N+3    |  0x01    |  ...     |  XOR     |  0x0D    |

为什么用 0xAA 0x55 做帧头？因为这两个字节的二进制是 10101010 和 01010101，交替的 0 和 1，不容易跟数据混淆。这是我个人的经验之谈。

4.2.2 解析器实现

来看一个完整的解析器代码：

class ProtocolParser:
    def __init__(self):
        self.buffer = bytearray()
        self.FRAME_HEAD = bytes([0xAA, 0x55])
        self.FRAME_TAIL = 0x0D
    
    def feed(self, data):
        """喂数据给解析器"""
        self.buffer.extend(data)
        frames = []
        
        while len(self.buffer) >= 5:  # 最小帧长度
            # 查找帧头
            head_idx = self.buffer.find(self.FRAME_HEAD)
            if head_idx == -1:
                # 没找到帧头，清空缓冲区
                self.buffer.clear()
                break
            
            # 丢弃帧头前的无效数据
            if head_idx > 0:
                self.buffer = self.buffer[head_idx:]
            
            # 检查长度是否足够
            if len(self.buffer) < 5:
                break
            
            frame_len = self.buffer[2]  # 长度字段
            
            # 检查完整帧是否到达
            if len(self.buffer) < frame_len + 4:
                break
            
            # 提取一帧
            frame = self.buffer[:frame_len + 4]
            
            # 校验
            if self._check_frame(frame):
                frames.append(self._parse_frame(frame))
            
            # 移除已处理的帧
            self.buffer = self.buffer[frame_len + 4:]
        
        return frames
    
    def _check_frame(self, frame):
        """校验帧：XOR 校验"""
        calc_xor = 0
        for b in frame[1:-2]:  # 从长度字段到数据末尾
            calc_xor ^= b
        return calc_xor == frame[-2]
    
    def _parse_frame(self, frame):
        """解析帧内容"""
        return {
            'cmd': frame[3],
            'data': frame[4:-2],
            'raw': frame
        }

4.3 多串口并发测试

做嵌入式测试，经常要同时跟多个设备通信。比如测试一个网关设备，它可能有 4 个串口，每个串口接一个从设备。这时候单线程就不够用了。

4.3.1 使用 threading 实现并发

我推荐用 Python 的 threading 模块，每个串口开一个线程：

import serial
import threading
import time

class SerialWorker(threading.Thread):
    def __init__(self, port, baudrate, name):
        super().__init__()
        self.port = port
        self.baudrate = baudrate
        self.name = name
        self.running = True
        self.ser = None
    
    def run(self):
        try:
            self.ser = serial.Serial(
                port=self.port,
                baudrate=self.baudrate,
                timeout=0.1
            )
            print(f"[{self.name}] 串口 {self.port} 已打开")
            
            while self.running:
                if self.ser.in_waiting > 0:
                    data = self.ser.read(self.ser.in_waiting)
                    print(f"[{self.name}] 收到: {data.hex()}")
                time.sleep(0.01)
        
        except Exception as e:
            print(f"[{self.name}] 错误: {e}")
        finally:
            if self.ser:
                self.ser.close()
    
    def stop(self):
        self.running = False

# 启动多个串口
workers = []
ports = ['COM3', 'COM4', 'COM5', 'COM6']

for i, port in enumerate(ports):
    worker = SerialWorker(port, 115200, f"Worker-{i+1}")
    worker.start()
    workers.append(worker)

# 运行一段时间后停止
time.sleep(60)
for w in workers:
    w.stop()
for w in workers:
    w.join()

这里有个关键点：每个线程的 timeout 要设得短一点，比如 0.1 秒。这样线程能及时响应停止信号，不会卡死。

4.3.2 使用队列管理数据

多线程环境下，数据共享是个麻烦事。我习惯用 queue.Queue 来做线程间通信：

import queue

# 全局数据队列
data_queue = queue.Queue()

class SerialWorker(threading.Thread):
    def run(self):
        # ... 省略串口打开代码 ...
        while self.running:
            if self.ser.in_waiting > 0:
                data = self.ser.read(self.ser.in_waiting)
                # 把数据放到队列里
                data_queue.put({
                    'port': self.port,
                    'data': data,
                    'time': time.time()
                })

# 主线程处理数据
def process_data():
    while True:
        try:
            item = data_queue.get(timeout=1)
            print(f"处理来自 {item['port']} 的数据: {item['data'].hex()}")
        except queue.Empty:
            pass

4.3.3 并发测试的注意事项

• 资源限制：Windows 下串口数量有限，一般不超过 256 个。Linux 下好一些，但也要注意文件描述符限制。
• 线程安全：pySerial 的读写操作是线程安全的，但 open() 和 close() 不是。建议在同一个线程里打开和关闭串口。
• 异常处理：串口可能随时断开，每个线程都要有完善的异常处理机制。

好了，关于串口通信脚本，今天就聊这么多。pySerial 的用法其实不难，难的是怎么把它用好、用稳。多串口并发测试更是考验代码的健壮性。希望这些经验能帮你少走些弯路。