3、数据解析与清洗：字节流处理、ASCII与二进制协议、异常值过滤
好，咱们进入第三章。说实话，这一章才是真正考验功力的地方。
你想想看，仪器把数据发过来了，一串字节流，可能是乱码一样的二进制，也可能是你能读懂的ASCII字符串。但不管哪种，直接拿来用？那肯定不行。我见过太多新手，拿到原始数据就往数据库里塞，结果画出来的曲线跟心电图似的——全是毛刺。
数据解析与清洗，说白了就是两件事：把字节流翻译成我们能理解的数值，然后把那些明显不靠谱的异常值干掉。今天咱们就聊聊这个。
3.1 字节流处理：从裸数据到结构化信息
仪器传过来的数据，本质上就是一串字节。比如一个16位的整数值，它可能是高字节在前（Big-Endian），也可能是低字节在前（Little-Endian）。搞反了，数值就完全不对了。
核心原则：先搞清楚协议，再动手写代码。别上来就猜。
我习惯把字节流处理分成三步：
- 接收与缓存——把串口或网络来的数据先存到缓冲区
- 帧同步与校验——找到一帧数据的起始和结束，做CRC或校验和
- 字段提取与转换——按协议把每个字节或字节组合解析成实际物理量
举个例子，一个简单的温度采集协议，每帧6个字节：
# 帧格式: [0xAA][0x55][温度高字节][温度低字节][校验和][0x0D]
# 温度值 = (高字节 << 8) | 低字节，单位0.1°C
def parse_temperature_frame(buffer):
if len(buffer) < 6:
return None
# 检查帧头
if buffer[0] != 0xAA or buffer[1] != 0x55:
return None
# 检查帧尾
if buffer[5] != 0x0D:
return None
# 校验和验证
checksum = (buffer[0] + buffer[1] + buffer[2] + buffer[3]) & 0xFF
if checksum != buffer[4]:
return None # 数据损坏，丢弃
# 提取温度值
raw_value = (buffer[2] << 8) | buffer[3]
temperature = raw_value * 0.1 # 转换为实际温度
return temperature我的经验：校验和这一步千万别省。我曾经在一个项目中偷懒没做校验，结果仪器偶尔受干扰发错数据，系统直接记录了一个-273°C的「绝对零度」值。排查了整整两天才发现是校验问题。
3.2 ASCII协议 vs 二进制协议
这两种协议，各有各的适用场景。我简单说说我的看法。
| 特性 | ASCII协议 | 二进制协议 |
|---|---|---|
| 可读性 | 人类可读，调试方便 | 需要工具解析 |
| 数据密度 | 低，传输同样信息需要更多字节 | 高，紧凑高效 |
| 解析速度 | 较慢（字符串转换开销） | 快（直接内存拷贝） |
| 典型应用 | Modbus ASCII、SCPI命令 | Modbus RTU、CAN总线 |
ASCII协议，说白了就是人类能读懂的文本。比如一台光谱仪返回的数据可能是这样的：
# ASCII协议示例
"CH1: 23.45°C, CH2: 24.12°C, CH3: 22.98°C\r\n"解析起来也不难，用字符串分割就行：
def parse_ascii_temperature(data_str):
# 去掉换行符
data_str = data_str.strip()
# 按逗号分割
channels = data_str.split(', ')
result = {}
for ch in channels:
# 每个通道格式: "CH1: 23.45°C"
parts = ch.split(': ')
channel_name = parts[0]
# 提取数值部分，去掉单位
value_str = parts[1].replace('°C', '')
result[channel_name] = float(value_str)
return result二进制协议呢？它更紧凑，但调试起来就麻烦多了。我记得有一次调试一个进口分析仪，它的二进制协议文档有50多页，每个bit都有含义。我不得不写了一个专门的「协议查看器」工具，才能看清每个字节到底代表什么。
注意：二进制协议一定要小心字节序（Endianness）和位对齐。很多仪器用的是大端序，但你的CPU可能是小端序。不做转换的话，数据全错。
3.3 异常值过滤：别让坏数据毁了你的分析
数据解析出来了，但你能直接用吗？不能。仪器本身有噪声，传输过程可能受干扰，偶尔还会出现传感器故障。这些都会产生异常值。
我常用的异常值过滤方法有三种：
- 阈值过滤——超出物理范围的值直接丢弃
- 滑动窗口均值滤波——用附近几个点的平均值代替当前点
- 中值滤波——对突变噪声特别有效
先看阈值过滤，这个最简单：
def threshold_filter(value, min_val, max_val):
"""超出物理范围的值直接丢弃"""
if value < min_val or value > max_val:
return None # 返回None表示无效
return value
# 比如温度传感器，物理范围是-40°C到85°C
filtered_temp = threshold_filter(raw_temp, -40.0, 85.0)滑动窗口均值滤波，适合平滑缓慢变化的信号：
class MovingAverageFilter:
def __init__(self, window_size=5):
self.window = []
self.window_size = window_size
def filter(self, value):
self.window.append(value)
if len(self.window) > self.window_size:
self.window.pop(0)
return sum(self.window) / len(self.window)中值滤波呢？它对那种突然跳变的毛刺特别有效。比如一个压力传感器，正常值在100kPa左右，突然跳到了500kPa然后又回来——这明显是干扰。中值滤波能很好地处理这种情况。
def median_filter(values, window_size=5):
"""中值滤波，对突变噪声效果好"""
if len(values) < window_size:
return values[-1] # 数据不够，返回最后一个值
window = values[-window_size:]
window.sort()
return window[window_size // 2] # 返回中位数我的建议：别只用一种滤波方法。我通常的做法是：先用阈值过滤去掉明显错误的数据，再用中值滤波去掉毛刺，最后用滑动均值平滑曲线。三层过滤下来，数据质量基本就靠谱了。
3.4 实战：一个完整的解析与清洗流程
好了，咱们把上面这些串起来，看看一个完整的流程长什么样。
class DataProcessor:
def __init__(self):
self.buffer = b''
self.moving_avg = MovingAverageFilter(window_size=3)
self.recent_values = []
def process_byte_stream(self, new_data):
"""处理新收到的字节流"""
self.buffer += new_data
results = []
while len(self.buffer) >= 6:
# 尝试解析一帧
temp = parse_temperature_frame(self.buffer[:6])
if temp is not None:
# 第一步：阈值过滤
temp = threshold_filter(temp, -40.0, 85.0)
if temp is not None:
# 第二步：中值滤波
self.recent_values.append(temp)
if len(self.recent_values) > 5:
self.recent_values.pop(0)
temp = median_filter(self.recent_values, 5)
# 第三步：滑动均值
temp = self.moving_avg.filter(temp)
results.append(temp)
# 移除已处理的帧
self.buffer = self.buffer[6:]
else:
# 帧头不对，尝试滑动查找
self.buffer = self.buffer[1:]
return results避坑指南：我曾经在一个项目中，因为缓冲区处理不当，导致数据帧错位，解析出来的温度值忽高忽低。后来我加了一个「帧同步状态机」，每次只处理完整的一帧，不完整的就等下一包数据。问题就解决了。
嗯，这一章的内容就这些。数据解析与清洗，说白了就是「翻译」和「筛选」两个动作。翻译要准确，筛选要合理。你想想看，如果原始数据里混了几个异常值，后面做趋势分析、报警判断，全都会跟着错。所以这一步，值得你花时间好好打磨。
下一章咱们聊聊数据存储——怎么把清洗好的数据高效地存起来，方便后续查询和分析。