4、数据采集可靠性：传感器数据滤波、异常值检测、数据冗余校验、看门狗定时器编程

各位同学，咱们今天聊一个很实在的话题——数据采集的可靠性。

我在仪器行业摸爬滚打了十几年，见过太多因为数据采集不可靠导致的惨案。有一次在化工厂，一个温度传感器因为现场干扰，读数跳了20多度，结果控制系统误判，差点把反应釜给憋炸了。嗯，从那以后，我对数据采集的可靠性就特别较真。

4.1 传感器数据滤波

传感器数据滤波，说白了就是给原始数据「洗个澡」，把那些不干净的噪声去掉。你想想看，传感器采集到的信号，多多少少都会混入一些干扰——电源纹波、电磁耦合、机械振动，这些都是噪声源。

我个人习惯把滤波分为两大类：硬件滤波和软件滤波。硬件滤波靠RC电路、有源滤波器这些，但今天咱们重点讲软件滤波，因为灵活、成本低，而且调试方便。

4.1.1 限幅滤波法

这个方法最简单，也最实用。它的思路是：如果当前采样值和上一次采样值的差值超过某个阈值，就认为这次采样是无效的，直接丢弃或用上一次的值代替。

# 限幅滤波示例
def limit_filter(new_value, last_value, max_deviation):
    if abs(new_value - last_value) > max_deviation:
        return last_value  # 超出范围，用上次值
    else:
        return new_value

# 使用示例
last_temp = 25.0
current_temp = 26.8
filtered_temp = limit_filter(current_temp, last_temp, 2.0)
print(f"滤波后温度: {filtered_temp}°C")

我在项目中遇到过一种情况：一个压力传感器在启动瞬间会有一个尖峰脉冲，用限幅滤波直接把它干掉了。阈值怎么选？一般取正常信号最大变化速率的1.5到2倍。

4.1.2 中位值滤波法

这个方法对付脉冲干扰特别有效。连续采样N次（一般取奇数，比如5次或7次），然后排序，取中间那个值作为有效值。

# 中位值滤波示例
def median_filter(samples):
    sorted_samples = sorted(samples)
    mid_index = len(sorted_samples) // 2
    return sorted_samples[mid_index]

# 连续采样5次
raw_data = [23.1, 25.7, 22.8, 24.3, 26.2]
filtered_value = median_filter(raw_data)
print(f"中位值滤波结果: {filtered_value}")

为什么取中位数而不是平均值？因为平均值会被极端值拉偏，而中位数不受影响。你想想看，如果5个数据里有一个是100，平均值就废了，但中位数依然稳如泰山。

4.1.3 滑动平均滤波法

这个方法适合周期性噪声。维护一个固定长度的队列，每次新数据进来，把最老的数据踢出去，然后计算队列里所有数据的平均值。

# 滑动平均滤波示例
class MovingAverageFilter:
    def __init__(self, window_size=5):
        self.window_size = window_size
        self.buffer = []
    
    def update(self, new_value):
        self.buffer.append(new_value)
        if len(self.buffer) > self.window_size:
            self.buffer.pop(0)
        return sum(self.buffer) / len(self.buffer)

# 使用示例
filter = MovingAverageFilter(5)
for i in range(10):
    raw = 50 + (i % 3) * 5  # 模拟波动数据
    filtered = filter.update(raw)
    print(f"原始: {raw:.1f}, 滤波后: {filtered:.2f}")
我的小技巧：窗口大小选3-10比较合适。窗口太小滤波效果差，窗口太大响应变慢。我在做温度采集时一般用5点滑动平均，效果不错。

4.2 异常值检测

滤波只能处理小噪声，但遇到真正的异常值——比如传感器断线、AD转换器故障——就得靠异常值检测了。

异常值检测的核心思想是：判断当前值是否在合理的统计范围内。常用的方法有3σ准则和箱线图法。

4.2.1 3σ准则

假设数据服从正态分布，那么99.7%的数据会落在均值±3倍标准差范围内。超出这个范围的，就认为是异常值。

# 3σ异常值检测示例
import numpy as np

def detect_outliers_3sigma(data):
    mean = np.mean(data)
    std = np.std(data)
    threshold = 3 * std
    outliers = []
    for i, value in enumerate(data):
        if abs(value - mean) > threshold:
            outliers.append((i, value))
    return outliers, mean, std

# 测试数据
sensor_data = [25.1, 24.8, 25.3, 25.0, 100.5, 24.9, 25.2]
outliers, mean, std = detect_outliers_3sigma(sensor_data)
print(f"均值: {mean:.2f}, 标准差: {std:.2f}")
print(f"检测到异常值: {outliers}")
注意：3σ准则要求数据量不能太少，至少要有10个以上样本。我曾经在一个只有5个样本的数据集上用3σ，结果把正常值也判成了异常，闹了笑话。

4.2.2 箱线图法

这个方法不依赖正态分布假设，更鲁棒。它用四分位数来定义异常范围：

  • 下四分位数Q1：数据中25%的值小于它
  • 上四分位数Q3：数据中75%的值小于它
  • 四分位距IQR = Q3 - Q1
  • 异常值判定：小于Q1 - 1.5×IQR 或 大于Q3 + 1.5×IQR
# 箱线图异常值检测示例
def detect_outliers_iqr(data):
    sorted_data = sorted(data)
    n = len(sorted_data)
    q1 = sorted_data[n // 4]
    q3 = sorted_data[3 * n // 4]
    iqr = q3 - q1
    lower_bound = q1 - 1.5 * iqr
    upper_bound = q3 + 1.5 * iqr
    outliers = [(i, v) for i, v in enumerate(data) if v < lower_bound or v > upper_bound]
    return outliers, lower_bound, upper_bound

data = [22, 24, 25, 26, 28, 30, 45, 23, 27]
outliers, low, high = detect_outliers_iqr(data)
print(f"正常范围: [{low:.1f}, {high:.1f}]")
print(f"异常值: {outliers}")

4.3 数据冗余校验

数据在传输过程中可能会出错，尤其是工业现场，电磁干扰严重。冗余校验就是给数据加个「防伪标签」，接收方收到后验证标签对不对。

常用的校验方法有：

校验方法 原理 适用场景 检测能力
奇偶校验 统计1的个数是奇数还是偶数 简单串口通信 只能检测奇数位错误
累加和校验 所有字节相加取低8位 Modbus协议 能检测大部分错误
CRC校验 多项式除法求余数 工业总线、存储 检测能力最强

我个人最常用的是CRC16，虽然计算量比累加和大，但可靠性高得多。下面是一个CRC16的Python实现：

# CRC16校验示例（Modbus标准）
def crc16_modbus(data):
    crc = 0xFFFF
    for byte in data:
        crc ^= byte
        for _ in range(8):
            if crc & 0x0001:
                crc = (crc >> 1) ^ 0xA001
            else:
                crc >>= 1
    return crc

# 发送数据时计算校验码
send_data = [0x01, 0x03, 0x00, 0x00, 0x00, 0x01]
crc_value = crc16_modbus(send_data)
print(f"CRC校验码: 0x{crc_value:04X}")

# 接收数据时验证
received_data = [0x01, 0x03, 0x02, 0x00, 0x64, 0x79, 0x87]
received_crc = (received_data[-1] << 8) | received_data[-2]
calculated_crc = crc16_modbus(received_data[:-2])
if calculated_crc == received_crc:
    print("数据校验通过，数据完整")
else:
    print("数据校验失败，数据可能被篡改")
重要提醒：CRC校验虽然强，但不是万能的。我曾经遇到过一个案例，两个不同数据包算出来的CRC竟然一样——这叫CRC碰撞。虽然概率极低（约1/65536），但在高可靠性要求的场合，建议再加一层校验，比如双重CRC或者CRC+累加和。

4.4 看门狗定时器编程

看门狗定时器（Watchdog Timer，WDT）是嵌入式系统的最后一道防线。它的原理很简单：程序正常运行时会定期「喂狗」（重置定时器），如果程序跑飞或死锁，没有及时喂狗，定时器溢出就会触发系统复位。

看门狗有两种常见类型：

  • 硬件看门狗：独立的芯片，可靠性高，但需要额外硬件
  • 软件看门狗：利用MCU内部定时器，成本低，但可能被异常程序干扰

下面是一个软件看门狗的Python模拟实现：

# 看门狗定时器模拟
import time
import threading

class Watchdog:
    def __init__(self, timeout=5.0):
        self.timeout = timeout
        self.last_feed = time.time()
        self.running = True
        self.thread = threading.Thread(target=self._monitor)
        self.thread.daemon = True
        self.thread.start()
    
    def feed(self):
        """喂狗操作"""
        self.last_feed = time.time()
        print(f"看门狗已喂食，当前时间: {time.strftime('%H:%M:%S')}")
    
    def _monitor(self):
        """监控线程"""
        while self.running:
            elapsed = time.time() - self.last_feed
            if elapsed > self.timeout:
                print(f"警告！看门狗超时！已过{elapsed:.1f}秒，执行系统复位！")
                self._system_reset()
            time.sleep(0.5)
    
    def _system_reset(self):
        """模拟系统复位"""
        print("系统正在复位...")
        self.last_feed = time.time()  # 复位后重新开始
        print("系统复位完成")
    
    def stop(self):
        self.running = False

# 使用示例
watchdog = Watchdog(timeout=3.0)

# 模拟正常程序运行
for i in range(5):
    time.sleep(1.0)
    watchdog.feed()  # 每1秒喂一次狗

# 模拟程序卡死
print("程序卡死，不再喂狗...")
time.sleep(5.0)  # 超过3秒超时，看门狗会触发复位

watchdog.stop()
关键注意事项：
  1. 喂狗的位置要选对。不能放在中断里喂，否则主程序死锁了中断还在喂，看门狗就失效了。
  2. 超时时间要合理。太短容易误触发，太长起不到保护作用。一般设为正常喂狗周期的2-3倍。
  3. 多任务系统中，每个关键任务都要有独立的喂狗点，确保所有任务都在正常运行。

我记得有一次调试一个现场设备，程序偶尔会死机，但看门狗从来没触发过。查了两天才发现，喂狗代码放在了定时器中断里，主程序已经跑飞了，中断还在正常执行。嗯，这个教训让我记住了：喂狗一定要放在主循环的关键路径上。

4.5 综合应用示例

最后，我把今天讲的内容整合成一个完整的传感器数据采集程序。这个程序会：

  1. 采集原始数据
  2. 进行限幅滤波和中位值滤波
  3. 用3σ准则检测异常值
  4. 计算CRC校验码
  5. 启动看门狗监控
# 综合数据采集可靠性程序
import time
import threading
import numpy as np

class ReliableDataAcquisition:
    def __init__(self, window_size=5, max_deviation=3.0, watchdog_timeout=5.0):
        self.window_size = window_size
        self.max_deviation = max_deviation
        self.buffer = []
        self.last_value = None
        self.watchdog = Watchdog(timeout=watchdog_timeout)
    
    def acquire_data(self, raw_value):
        """完整的数据采集流程"""
        # 1. 限幅滤波
        if self.last_value is not None:
            filtered = limit_filter(raw_value, self.last_value, self.max_deviation)
        else:
            filtered = raw_value
        self.last_value = filtered
        
        # 2. 中位值滤波
        self.buffer.append(filtered)
        if len(self.buffer) > self.window_size:
            self.buffer.pop(0)
        if len(self.buffer) >= 3:
            filtered = median_filter(self.buffer)
        
        # 3. 异常值检测
        if len(self.buffer) >= 10:
            outliers, _, _ = detect_outliers_3sigma(self.buffer)
            if outliers:
                print(f"检测到异常值: {outliers}")
                return None
        
        # 4. 计算CRC校验
        data_bytes = [int(filtered * 100)]  # 转为整数
        crc = crc16_modbus(data_bytes)
        
        # 5. 喂狗
        self.watchdog.feed()
        
        return {
            'value': filtered,
            'crc': crc,
            'timestamp': time.time()
        }

# 使用示例
daq = ReliableDataAcquisition()
for i in range(20):
    raw = 25.0 + np.random.normal(0, 0.5)  # 模拟带噪声的数据
    if i == 10:
        raw = 100.0  # 模拟异常值
    result = daq.acquire_data(raw)
    if result:
        print(f"采集结果: 值={result['value']:.2f}, CRC=0x{result['crc']:04X}")
    time.sleep(0.5)

好了，这一章的内容就到这里。数据采集可靠性是个系统工程，滤波、异常检测、校验、看门狗，每个环节都不能少。你想想看，如果其中一个环节出了问题，整个系统的数据就可能不可靠，后果可能很严重。

下一章我们会讲数据存储的可靠性，包括日志记录、掉电保护和数据备份。到时候见！

← 上一章 📖 返回目录 下一章 →