3. pyserial 核心 API 详解:从构造到实战
说实话,pyserial 这个库,你用好了它就是一把瑞士军刀,用不好就是个黑盒子。我最早接触它是在一个工业数据采集项目里,当时设备死活连不上,折腾了两天才发现是超时参数设错了。从那以后,我对这几个核心 API 就格外上心。
今天咱们就把 Serial 类的构造、读写方法、超时机制、事件处理和缓冲区管理这五个核心模块拆开揉碎了讲。你跟着我走一遍,以后遇到串口问题心里就有底了。
3.1 Serial 类构造:你的第一个串口对象
创建一个串口对象,说白了就是告诉 Python:「我要跟哪个设备说话,怎么说话」。构造函数的参数不少,但真正关键的其实就几个。
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
port |
str | None | 端口名,Windows 用 COM3,Linux 用 /dev/ttyUSB0 |
baudrate |
int | 9600 | 波特率,常见值 9600、115200、460800 |
bytesize |
int | 8 | 数据位,可选 5、6、7、8 |
parity |
str | 'N' | 校验位:'N' 无、'E' 偶、'O' 奇、'S' 空格、'M' 标记 |
stopbits |
float | 1 | 停止位,可选 1、1.5、2 |
timeout |
float | None | 读超时(秒),None 表示阻塞等待 |
write_timeout |
float | None | 写超时(秒) |
我的习惯:构造时就把 timeout 设好,哪怕设个 0.1 秒也比 None 强。None 意味着程序可能永远卡在 read() 上,这在生产环境里是灾难。
import serial
# 最简构造
ser = serial.Serial('COM3', 115200)
# 推荐写法:显式指定所有关键参数
ser = serial.Serial(
port='COM3',
baudrate=115200,
bytesize=8,
parity='N',
stopbits=1,
timeout=0.5,
write_timeout=0.5
)
你想想看,如果设备手册上写的是「8 数据位、无校验、1 停止位」,那上面这个配置就是标准答案。我见过有人把校验位设成 'E',结果设备回了一堆乱码,排查了半天才发现是参数对不上。
3.2 read / write 方法:数据的搬运工
读写操作是串口通信的核心。pyserial 提供了几个方法,各有各的适用场景。
3.2.1 写数据:write()
write() 接受字节数据,返回实际写入的字节数。注意,它不保证一次写完所有数据——虽然大多数情况下能写完,但严谨的做法是检查返回值。
data = b'AT+CMD\r\n'
written = ser.write(data)
if written != len(data):
print(f'只写入了 {written} 字节,期望 {len(data)} 字节')
小技巧:发送字符串时记得编码。我习惯用 ser.write('AT\r\n'.encode()),或者直接写 b'AT\r\n'。别传 str 进去,会报错。
3.2.2 读数据:read() 与 read_until()
read(size) 读取指定字节数,readline() 读取直到换行符,read_until(expected) 读取直到遇到指定字节序列。
# 读取固定长度
resp = ser.read(10)
# 读取一行(直到 \n)
line = ser.readline()
# 读取直到遇到 'OK'
resp = ser.read_until(b'OK')
我曾经在一个 GPS 模块项目里,设备返回的数据末尾是 \r\n 而不是 \n。用 readline() 死活读不到完整数据,换成 read_until(b'\r\n') 就对了。嗯,这里要注意:不同设备的行尾符不一样,别想当然。
3.3 超时机制:别让程序死等
超时是串口编程里最容易忽略、也最容易出问题的地方。pyserial 支持三种超时模式。
| 模式 | timeout 值 | 行为 |
|---|---|---|
| 阻塞模式 | None | read() 一直等,直到读够指定字节数 |
| 非阻塞模式 | 0 | read() 立即返回,有数据就返回,没有就返回空字节 |
| 超时模式 | 正数(秒) | read() 最多等 timeout 秒,超时返回已读取的数据 |
避坑指南:我曾经在一个自动化测试脚本里用了 timeout=None,结果设备偶尔不响应,整个脚本就卡死了。后来全部改成 timeout=1,配合循环重试,再也没出过问题。记住:生产环境永远不要用 None 超时。
# 推荐:带超时的循环读取
def read_with_retry(ser, expected_len, max_retries=3):
data = b''
for i in range(max_retries):
data += ser.read(expected_len - len(data))
if len(data) >= expected_len:
break
return data
3.4 事件处理:in_waiting 与事件驱动
很多时候你不想一直轮询,而是希望「有数据来了再处理」。pyserial 没有直接的事件回调,但我们可以用 in_waiting 属性配合线程来实现类似效果。
in_waiting 返回接收缓冲区中等待读取的字节数。这个值实时更新,你可以用它来判断是否有新数据。
import threading
def serial_reader(ser):
while True:
if ser.in_waiting > 0:
data = ser.read(ser.in_waiting)
print(f'收到: {data.hex()}')
time.sleep(0.01)
# 启动读取线程
reader_thread = threading.Thread(target=serial_reader, args=(ser,), daemon=True)
reader_thread.start()
我的经验:用 in_waiting 配合线程,比每次都调 read(1) 效率高得多。尤其是在高波特率(比如 921600)下,一次读一批数据能大幅减少系统调用开销。
3.5 缓冲区管理:输入输出缓冲区的清与读
串口通信中,缓冲区管理是个容易被忽视的细节。pyserial 提供了 reset_input_buffer() 和 reset_output_buffer() 两个方法。
| 方法 | 作用 | 使用场景 |
|---|---|---|
reset_input_buffer() |
清空接收缓冲区 | 设备刚上电时,丢弃残留的乱码数据 |
reset_output_buffer() |
清空发送缓冲区 | 取消尚未发出的数据 |
# 设备上电后,先清空输入缓冲区
ser.reset_input_buffer()
# 发送命令
ser.write(b'AT\r\n')
# 等待响应
time.sleep(0.1)
resp = ser.read(ser.in_waiting)
你想想看,设备刚上电时,串口线上可能有一些随机电平转换产生的垃圾数据。如果不先清空缓冲区,这些垃圾就会被当成有效数据读进来。我习惯在每次建立连接后,先调一次 reset_input_buffer(),这是个好习惯。
3.6 知识体系总览
下面这张图把 pyserial 核心 API 的关系梳理清楚了。你对照着看,心里就有谱了。
这五个模块构成了 pyserial 的核心骨架。你掌握了它们,就等于拿到了串口编程的钥匙。剩下的就是多写、多试、多踩坑——嗯,踩坑也是学习的一部分。