4、串口通信实战:使用PySerial读取美容仪MCU发送的原始数据包
好,咱们进入第四章节。前面几章我们把美容仪的传感器原理、数据格式都理清了,现在终于到了真刀真枪干的时候——用Python把MCU发过来的原始数据包读上来。
说实话,串口通信这块,我当年刚入行时也踩过不少坑。有一次在产线上调试,数据死活读不对,折腾了一下午,最后发现是波特率配错了。嗯,这种低级错误,谁还没犯过呢?
4.1 准备工作:安装PySerial
PySerial是Python操作串口的标配库,没有之一。我个人习惯用pip安装,简单粗暴:
pip install pyserial
如果你用的是Anaconda环境,也可以:
conda install pyserial
装完之后,在Python里试一下:
import serial
print(serial.__version__)
能打印出版本号,说明装好了。如果报错,多半是网络问题或者Python环境没切对。我建议你直接在终端里跑,别在Jupyter里折腾,省心。
4.2 打开串口:三步走
打开串口其实就三步:
- 找到设备端口号
- 配置参数(波特率、数据位、停止位等)
- 调用
serial.Serial()创建对象
先看代码:
import serial
# 端口号根据实际情况改
# Windows上一般是 COM3、COM4 这种
# Linux/Mac 上是 /dev/ttyUSB0、/dev/ttyAMA0 等
port = 'COM3'
baudrate = 115200
ser = serial.Serial(
port=port,
baudrate=baudrate,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=1
)
if ser.is_open:
print(f"串口 {port} 已打开,波特率 {baudrate}")
这里有个细节——timeout=1。什么意思?就是读数据时最多等1秒,超时了就返回空。我之前有个项目没设timeout,结果程序卡死在读取那里,整个界面都假死了。你想想看,这要是给客户演示,多尴尬。
4.3 读取原始数据包
美容仪MCU发送的数据包,通常是固定长度的。比如我们之前约定的,一包数据是 20 个字节。那怎么读呢?
最直接的方法:
# 读取固定长度的数据包
packet_size = 20
raw_data = ser.read(packet_size)
print(f"收到 {len(raw_data)} 字节: {raw_data.hex()}")
ser.read(n) 会尝试读取 n 个字节。如果数据不够,它会等,直到凑够 n 个或者超时。超时返回的数据可能少于 n 个字节。
所以,更稳妥的做法是:
def read_packet(ser, size, timeout=2):
"""读取完整的数据包,超时返回 None"""
data = b''
start_time = time.time()
while len(data) < size:
remaining = size - len(data)
chunk = ser.read(remaining)
if not chunk:
# 没读到数据,检查是否超时
if time.time() - start_time > timeout:
print("读取超时")
return None
continue
data += chunk
return data
# 使用
packet = read_packet(ser, 20)
if packet:
print(f"完整数据包: {packet.hex()}")
这个函数我用了好多年,基本没出过问题。核心思路就是——一次读不够,就循环读,直到凑齐为止。说白了,就是「攒够了再交货」。
4.4 解析数据包:从字节到有意义的数据
光拿到原始字节还不够,得解析。假设我们的数据包格式是这样的:
| 偏移 | 长度(字节) | 含义 | 示例 |
|---|---|---|---|
| 0 | 2 | 帧头 (0xAA 0x55) | AA 55 |
| 2 | 1 | 数据长度 | 0x10 (16) |
| 3 | 1 | 命令字 | 0x01 (温度数据) |
| 4 | 4 | 温度值 (float, 小端) | ... |
| 8 | 4 | 电导值 (float, 小端) | ... |
| 12 | 4 | 超声波值 (float, 小端) | ... |
| 16 | 2 | 校验和 (CRC16) | ... |
| 18 | 2 | 帧尾 (0x0D 0x0A) | 0D 0A |
解析代码:
import struct
def parse_packet(packet):
"""解析美容仪数据包"""
if len(packet) < 20:
print("数据包太短")
return None
# 检查帧头
if packet[0] != 0xAA or packet[1] != 0x55:
print("帧头错误")
return None
# 检查帧尾
if packet[-2] != 0x0D or packet[-1] != 0x0A:
print("帧尾错误")
return None
# 提取数据
data_len = packet[2]
cmd = packet[3]
# 解析浮点数(小端模式)
temperature = struct.unpack('<f', packet[4:8])[0]
conductance = struct.unpack('<f', packet[8:12])[0]
ultrasonic = struct.unpack('<f', packet[12:16])[0]
# 校验和验证(这里简化,实际用CRC16)
# 略...
return {
'cmd': cmd,
'temperature': temperature,
'conductance': conductance,
'ultrasonic': ultrasonic
}
# 使用
result = parse_packet(packet)
if result:
print(f"温度: {result['temperature']:.2f}°C")
print(f"电导: {result['conductance']:.3f} mS")
print(f"超声波: {result['ultrasonic']:.2f} mm")
4.5 持续读取:让程序跑起来
实际项目中,我们不可能只读一包数据。美容仪是持续发送的,每秒可能几十包。所以需要一个循环:
import time
def continuous_read(ser, packet_size=20):
"""持续读取并解析数据包"""
print("开始读取数据,按 Ctrl+C 停止...")
try:
while True:
packet = read_packet(ser, packet_size)
if packet:
result = parse_packet(packet)
if result:
# 这里可以加时间戳
timestamp = time.strftime("%H:%M:%S.%f")[:-3]
print(f"[{timestamp}] T={result['temperature']:.1f}°C "
f"G={result['conductance']:.2f}mS "
f"U={result['ultrasonic']:.1f}mm")
else:
# 没读到数据,稍等再试
time.sleep(0.01)
except KeyboardInterrupt:
print("\n停止读取")
finally:
ser.close()
# 启动
continuous_read(ser)
跑起来之后,你会看到类似这样的输出:
[14:23:45.123] T=36.5°C G=0.452mS U=12.3mm
[14:23:45.128] T=36.6°C G=0.451mS U=12.4mm
[14:23:45.133] T=36.5°C G=0.453mS U=12.3mm
...
4.6 常见问题与避坑
- 读不到数据? 先检查物理连接——TX接RX,RX接TX,GND接GND。别笑,我见过有人TX接TX的,能通才怪。
- 数据乱码? 多半是波特率不匹配。美容仪MCU的波特率可能和你的设置不一样,用示波器量一下实际波特率。
- 丢包? 缓冲区溢出。可以调大串口的接收缓冲区,或者加快读取频率。
- 程序卡死? 没设timeout,或者死循环里没有sleep。加个time.sleep(0.01)就能解决。
好了,这一章的内容就这些。核心就是——打开串口、循环读取、解析数据。下一章我们会把这些数据存到数据库里,方便后续分析。到时候再聊。