第4章:串口通信实战——pyserial库详解与数据收发
串口通信,说白了就是让电脑跟硬件设备「对话」。做BMS上位机开发,这步绕不开。我个人习惯把串口比作一条水管——数据就是水,参数配置就是管径和阀门。调不对,水就流不通。
4.1 pyserial库:你的串口瑞士军刀
Python里做串口通信,pyserial是首选。我最早用的时候还踩过坑——装错版本,结果连端口都打不开。记住,用这个命令安装:
pip install pyserial
导入也很简单:
import serial
嗯,这里要注意,别写成 import pyserial,那是错的。我刚开始就犯过这毛病。
4.2 串口参数配置——五个关键参数
配置串口就像调对讲机频道。五个参数必须搞对,否则数据全是乱码。我在项目中遇到过好几次,明明线接对了,数据就是读不出来——最后发现是校验位设错了。
| 参数 | 说明 | 常见值 | 我的建议 |
|---|---|---|---|
| 波特率 | 数据传输速率 | 9600, 115200 | BMS常用115200 |
| 数据位 | 每帧数据位数 | 8 | 几乎都用8 |
| 停止位 | 帧结束标志 | 1, 2 | 默认1就够了 |
| 校验位 | 错误检测 | None, Even, Odd | BMS常用无校验 |
| 超时 | 读取等待时间 | 0.1~1秒 | 别设0,会卡死 |
代码配置长这样:
import serial
ser = serial.Serial(
port='COM3', # Windows用COM口,Linux用/dev/ttyUSB0
baudrate=115200, # 波特率
bytesize=8, # 数据位
parity='N', # 校验位:N无/E偶/O奇
stopbits=1, # 停止位
timeout=0.5 # 超时0.5秒
)
避坑指南:我曾经因为端口号写错,排查了整整两个小时。建议先用设备管理器确认端口号,或者用代码自动扫描:
import serial.tools.list_ports
ports = serial.tools.list_ports.comports()
for p in ports:
print(p.device, p.description)
4.3 数据收发——核心操作
收发数据其实就两个动作:写和读。但细节决定成败。
发送数据
# 发送字符串(要转成bytes)
ser.write(b'AT\r\n')
# 发送十六进制数据
data = bytes([0xAA, 0x55, 0x01, 0x02])
ser.write(data)
# 发送带换行的
ser.write('读取电压\n'.encode('utf-8'))
你想想看,BMS发过来的数据大多是十六进制格式。我习惯用 bytes 类型操作,干净利落。
接收数据
# 读取指定字节数
data = ser.read(10)
# 读取一行(直到换行符)
line = ser.readline()
# 读取所有可用数据
all_data = ser.read(ser.in_waiting)
我的经验:用
ser.in_waiting 配合 read() 是最稳的。它能拿到缓冲区里所有数据,不会漏包。
4.4 缓冲区处理——别让数据「堵车」
缓冲区就像临时停车场。数据来了先停进去,你再去取。但如果不及时处理,就会溢出。
我遇到过最头疼的问题:读取速度跟不上发送速度,结果数据全乱了。后来用了清空缓冲区的办法:
# 清空输入缓冲区
ser.reset_input_buffer()
# 清空输出缓冲区
ser.reset_output_buffer()
# 读取并丢弃所有缓冲数据
ser.read(ser.in_waiting)
说白了,每次开始新通信前,先清一下缓冲区,这是个好习惯。
4.5 多线程接收数据——不卡主线程
串口读取是阻塞的。如果你在主线程里一直 read(),界面就卡死了。解决方案?开个独立线程去收数据。
我最早做BMS上位机时,就犯过这错——界面点不动,还以为程序死了。后来改成多线程,世界清净了。
import threading
import time
class SerialReader(threading.Thread):
def __init__(self, ser):
super().__init__()
self.ser = ser
self.running = True
self.data_queue = [] # 存放接收到的数据
def run(self):
while self.running:
if self.ser.in_waiting > 0:
data = self.ser.read(self.ser.in_waiting)
self.data_queue.append(data)
# 这里可以加回调,通知主界面更新
time.sleep(0.01) # 别让CPU跑满
def stop(self):
self.running = False
# 使用示例
ser = serial.Serial('COM3', 115200, timeout=0.5)
reader = SerialReader(ser)
reader.start()
# 主线程做其他事...
# 需要停止时:
reader.stop()
ser.close()
关键点:
- 线程里加
time.sleep(0.01),防止CPU占用100% - 用
self.running控制线程退出,别用_stop那种野路子 - 数据用队列存起来,主线程定时去取
4.6 实战:完整的收发循环
把上面知识点串起来,写个能用的例子:
import serial
import threading
import time
class BMS_Serial:
def __init__(self, port, baud=115200):
self.ser = serial.Serial(
port=port,
baudrate=baud,
bytesize=8,
parity='N',
stopbits=1,
timeout=0.1
)
self.rx_thread = None
self.running = False
def start(self):
self.running = True
self.rx_thread = threading.Thread(target=self._rx_loop)
self.rx_thread.start()
print(f'串口已打开:{self.ser.port}')
def _rx_loop(self):
while self.running:
try:
if self.ser.in_waiting > 0:
data = self.ser.read(self.ser.in_waiting)
self._process_data(data)
time.sleep(0.01)
except Exception as e:
print(f'接收异常:{e}')
break
def _process_data(self, data):
# 这里写你的数据处理逻辑
hex_str = data.hex().upper()
print(f'收到:{hex_str}')
def send(self, data):
if isinstance(data, str):
data = data.encode()
self.ser.write(data)
print(f'发送:{data.hex().upper()}')
def stop(self):
self.running = False
if self.rx_thread:
self.rx_thread.join()
self.ser.close()
print('串口已关闭')
# 用起来
bms = BMS_Serial('COM3')
bms.start()
# 发送查询指令
bms.send(bytes([0xAA, 0x55, 0x01, 0x00]))
# 等5秒后关闭
time.sleep(5)
bms.stop()
避坑指南:我曾经在
_rx_loop 里忘了加 try-except,结果串口一拔程序就崩了。记住,硬件通信一定要加异常处理。
4.7 常见问题与排查思路
做串口开发,问题无非这几类:
- 打不开端口:权限问题?端口被占用?用
list_ports先扫一遍 - 收到乱码:波特率对不对?数据位/校验位匹配吗?
- 收不到数据:线接好了吗?地线共地了吗?超时设太短?
- 程序卡死:是不是在主线程里
read()了?赶紧改多线程
嗯,串口通信这块,说白了就是耐心活。参数调对了,数据自然就通了。我刚开始做BMS项目时,光调串口就花了两天——后来发现是USB转串口线坏了。所以,硬件问题也别忽略。
下一章,我们会把串口收上来的数据解析成真正的电池参数。准备好了吗?