3、数据采集核心：读取万用表数据帧、解析数据帧、数据校验与异常处理
好，咱们进入最硬核的部分了。说实话，前面那些协议和配置都是开胃菜，真正让万用表数据变成可用信息的，就是这一章的内容。我当年第一次做数据采集时，以为只要把串口数据读出来就完事了，结果发现读出来的全是乱码——嗯，那时候我才意识到，数据帧的解析和校验才是真正的技术活。
3.1 读取万用表数据帧
万用表通过串口发送的数据，不是我们直接能看懂的电压值。它是一串二进制数据帧。不同品牌、不同型号的万用表，帧格式都不一样。我习惯把数据帧的读取分成三步：
- 打开串口：配置好波特率、数据位、停止位、校验位
- 持续监听：用循环或回调方式读取数据
- 帧同步：找到帧头，确定一帧数据的起始位置
举个例子，我常用的某款万用表，数据帧格式是这样的：
| 字节位置 | 含义 | 说明 |
|---|---|---|
| 0 | 帧头 | 0xAA |
| 1 | 功能码 | 0x01=电压，0x02=电流，0x03=电阻 |
| 2-5 | 数据值 | 4字节浮点数，小端模式 |
| 6 | 单位 | 0x00=V，0x01=A，0x02=Ω |
| 7 | 校验和 | 前7字节累加和取低8位 |
读取代码其实不复杂，但要注意一点：串口数据是流式的，你一次读到的可能不是完整的一帧。我建议用缓冲区来拼帧。
import serial
def read_frame(ser):
"""读取一帧完整数据，返回字节数组"""
buffer = bytearray()
while True:
# 等待帧头
byte = ser.read(1)
if byte == b'\xAA':
buffer.append(0xAA)
break
# 读取剩余7个字节
buffer += ser.read(7)
return buffer3.2 解析数据帧
拿到完整的一帧数据后，就要把它翻译成我们能理解的物理量了。说白了，就是把字节拆开，按照协议定义提取各个字段。
解析的核心在于字节序和数据类型转换。我见过太多新手在这个地方翻车。比如万用表传过来的浮点数，到底是小端还是大端？这得看手册。我用的那款是小端模式，也就是低字节在前。
import struct
def parse_frame(frame):
"""解析数据帧，返回(类型, 数值, 单位)"""
if len(frame) != 8:
return None
func_code = frame[1]
# 4字节浮点数，小端模式
value = struct.unpack('<f', frame[2:6])[0]
unit_code = frame[6]
# 映射单位
units = {0: 'V', 1: 'A', 2: 'Ω'}
unit = units.get(unit_code, '未知')
# 映射功能
funcs = {0x01: '电压', 0x02: '电流', 0x03: '电阻'}
func_name = funcs.get(func_code, '未知')
return func_name, value, unit注意：struct.unpack 的格式字符串 '<f' 表示小端4字节浮点数。如果是大端，要用 '>f'。这个搞错了，读出来的数值会完全不对。我曾经在项目里因为这个bug排查了整整一个下午。
3.3 数据校验
数据帧传过来，万一中间有干扰怎么办？你想想看，如果电压显示成1000V，那不得把设备烧了？所以校验是必须的。
万用表常用的校验方式有两种：
- 校验和：把帧头之后的所有字节加起来，取低8位，和帧尾的校验字节比对
- 奇偶校验：串口硬件层面的一种校验，保证每个字节中1的个数是奇数或偶数
我个人更看重校验和，因为它能检测出整帧数据的完整性。奇偶校验只能检测单字节错误，说实话在工业现场不太够用。
def verify_checksum(frame):
"""校验和验证，返回True/False"""
if len(frame) != 8:
return False
# 计算前7个字节的累加和
checksum = sum(frame[:7]) & 0xFF
# 与第8个字节比对
return checksum == frame[7]3.4 异常数据处理
数据采集过程中，异常是常态。不是每一帧数据都能用。我总结了几种常见的异常情况：
- 帧格式错误：长度不对、帧头不对
- 校验失败：数据被干扰
- 数值越界：比如电压值超过了万用表的量程
- 重复数据：万用表在切换量程时，可能会重复发送上一次的值
我的处理策略是：宁可丢数据，也不要用错误数据。具体做法是：
def process_data(frame):
"""处理一帧数据，返回有效数据或None"""
# 1. 基本校验
if not verify_checksum(frame):
return None
# 2. 解析
result = parse_frame(frame)
if result is None:
return None
func_name, value, unit = result
# 3. 合理性检查
# 电压一般不超过1000V，电流不超过20A，电阻不超过100MΩ
if func_name == '电压' and (value < 0 or value > 1000):
return None
if func_name == '电流' and (value < 0 or value > 20):
return None
if func_name == '电阻' and (value < 0 or value > 100_000_000):
return None
return result3.5 完整的数据采集流程
把上面这些串起来，就是一个完整的数据采集循环。我习惯这样组织代码：
def data_acquisition_loop(ser):
"""数据采集主循环"""
while True:
frame = read_frame(ser)
if frame is None:
continue
data = process_data(frame)
if data is None:
# 记录异常日志
log_error(f"无效数据帧: {frame.hex()}")
continue
func_name, value, unit = data
# 这里可以存储到数据库、写入文件、或者推送到前端
save_to_database(func_name, value, unit)
# 每秒采集一次，不要太快
time.sleep(1)嗯，到这里，数据采集的核心逻辑就完整了。从读取原始字节，到解析成物理量，再到校验和异常处理，每一步都不能马虎。我刚开始做这个的时候，总觉得校验是多余的，后来被现场数据坑过几次，就再也不敢偷懒了。
下一章我们会讲如何把这些数据存储起来，以及如何设计一个高效的日志系统。到时候见。