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 打开串口:三步走

打开串口其实就三步:

  1. 找到设备端口号
  2. 配置参数(波特率、数据位、停止位等)
  3. 调用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,结果程序卡死在读取那里,整个界面都假死了。你想想看,这要是给客户演示,多尴尬。

小提示: 如果你不确定美容仪MCU用的什么参数,先查数据手册。一般美容仪MCU默认是 115200、8、N、1。实在不行,用示波器抓一下波形,数数位宽也能算出来。

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")
重点: struct.unpack 的格式字符串 'f' 试试。

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
...
注意: 串口读取是阻塞操作。如果你的主程序还要做UI显示、数据存储等,建议把串口读取放在单独的线程里。否则界面会卡死。我曾经在做一个美容仪上位机时,就因为没开线程,结果用户一拔USB线,程序直接无响应。后来加了个 watchdog 线程才解决。

4.6 常见问题与避坑

  • 读不到数据? 先检查物理连接——TX接RX,RX接TX,GND接GND。别笑,我见过有人TX接TX的,能通才怪。
  • 数据乱码? 多半是波特率不匹配。美容仪MCU的波特率可能和你的设置不一样,用示波器量一下实际波特率。
  • 丢包? 缓冲区溢出。可以调大串口的接收缓冲区,或者加快读取频率。
  • 程序卡死? 没设timeout,或者死循环里没有sleep。加个time.sleep(0.01)就能解决。

好了,这一章的内容就这些。核心就是——打开串口、循环读取、解析数据。下一章我们会把这些数据存到数据库里,方便后续分析。到时候再聊。