一条数据链路的四段

“读取传感器”听起来只是一行代码，但要让云端看到一条可信的曲线，中间至少要经过四段处理：采样、标定、滤波、上报。任何一段偷懒，最终呈现的数据都会失真——采样太慢会漏掉突变，不标定就没有物理意义，不滤波会让曲线毛刺丛生，不做上报策略则要么流量爆炸，要么关键变化被淹没。

传感器的原始值只是电压或寄存器里的一个整数，它变成“23.5 ℃”这件事，是标定、滤波和单位约定共同完成的。把这四段分开写、分开测，是嵌入式数据采集最基本的工程习惯。
传感器采集配图：方形传感器芯片与数据波形曲线
一条完整的数据链路：采样、标定、滤波、上报，任何一环偷懒，云端看到的曲线都会失真

下面按这四段依次展开，代码以 Python 形式给出便于阅读，逻辑可以直接映射到 C 语言实现。

采样入口：I2C、SPI 与 ADC

数字传感器（温湿度、气压、姿态）通常通过 I2C 或 SPI 读取，模拟传感器则要经过 ADC 采样。三种入口的特点决定了后续处理方式。

入口典型器件特点与注意事项
I2CSHT30、MPU6050、BMP280两线制、可挂多设备；速率较低，长线易受干扰，需要上拉电阻
SPI高精度 ADC、Flash、显示屏速率高、全双工；片选线多，走线要求高
ADC热敏电阻、电位器、电流环需要参考电压与分辨率换算，噪声大，必须滤波
 
# 以 I2C 读取 SHT30 为例（树莓派 / Linux SBC，Python）
import smbus2, time

BUS, ADDR = 1, 0x44                      # SHT30 默认地址
bus = smbus2.SMBus(BUS)

def read_raw():
    bus.write_i2c_block_data(ADDR, 0x2C, [0x06])     # 高重复性、无时钟拉伸
    time.sleep(0.02)
    d = bus.read_i2c_block_data(ADDR, 0x00, 6)
    t_raw = (d[0] << 8) | d[1]
    h_raw = (d[3] << 8) | d[4]
    # SHT30 数据手册给出的换算公式，原始值本身就是定标的
    return -45 + 175 * t_raw / 65535, 100 * h_raw / 65535

print(read_raw())

用 ADC 时则要先明确两个参数：分辨率（多少位）和参考电压。12 位 ADC、参考 3.3V 时，每个计数约 0.8mV，换算关系是 value = raw / 4095 × 3.3。采样前的输入调理电路（分压、运放、RC 滤波）往往比软件更重要。

标定：从计数值到物理量

数字传感器通常在手册里已经给出线性换算公式，而模拟传感器几乎都要求现场标定。最通用的是两点标定：取两个已知参考点，解出直线的斜率与截距。

# 两点标定：已知参考点 (raw1, phys1)、(raw2, phys2)
def fit_two_point(raw1, phys1, raw2, phys2):
    k = (phys2 - phys1) / (raw2 - raw1)
    b = phys1 - k * raw1
    return k, b

def apply(k, b, raw):
    return k * raw + b

# 例：热敏电阻电桥在 0℃ 时输出 512，在 100℃ 时输出 870
k, b = fit_two_point(512, 0.0, 870, 100.0)
print("斜率 %.4f，截距 %.4f" % (k, b))
print("当前温度 %.1f ℃" % apply(k, b, 640))

# 现场标定结果应固化到配置里而不是代码中，便于不同批次单独校准
import json
json.dump({"k": k, "b": b, "unit": "C", "ver": "2026-10-13"},
          open("/etc/sensor_calib.json", "w"))
标定环节的三条纪律
  • 标定参数外置固化在配置文件或 Flash 参数区，不同批次设备可单独校准，不必重新编译固件。
  • 留出标定元数据记录标定时间、参考源、操作人，否则半年后没人能说清这条曲线准不准。
  • 非线性要分段热敏电阻、pH 电极这类非线性器件用单段直线误差会很大，应按区间分段拟合或查表插值。

滤波：四种方法的选择

原始数据里混着三类噪声：随机抖动、尖峰脉冲、系统性漂移。不同噪声要用不同工具处理。

# 1) 滑动平均：抑制随机抖动，代价是响应变慢
from collections import deque
win = deque(maxlen=8)
def avg(v):
    win.append(v)
    return sum(win) / len(win)

# 2) 中值滤波：专治尖峰脉冲（如电机干扰导致的一帧跳变）
def median3(a, b, c):
    return sorted([a, b, c])[1]

# 3) 一阶低通（指数加权）：计算量极小，适合 MCU 逐点执行
#    alpha = dt / (RC + dt)，alpha 越小越平滑、越迟钝
class LowPass:
    def __init__(self, alpha):
        self.a, self.y = alpha, None
    def __call__(self, x):
        self.y = x if self.y is None else self.a * x + (1 - self.a) * self.y
        return self.y

lp = LowPass(0.2)
print([round(lp(v), 2) for v in [23.0, 26.0, 23.2, 23.4, 23.1]])

选择原则很朴素：只要滤掉随机抖动，用滑动平均或低通；如果现场有明显的脉冲干扰（继电器、电机启停），先做中值再低通，顺序反了效果会差很多；只有在需要同时抑制噪声并估计变化趋势（如姿态融合）时，才值得上卡尔曼滤波。低通滤波器的 alpha 参数一定要用实际数据试出来，凭理论公式估算常常过于激进。

上报策略：既要省流量，又要不丢关键变化

把滤波后的每个点都发到云端，是最不经济的做法。常见的三种上报策略通常会组合使用。

import json, time

class Reporter:
    """周期上报 + 死区变化上报 + 极值聚合"""
    def __init__(self, period=60, deadband=0.5, max_len=20):
        self.period, self.deadband = period, deadband
        self.last_report_ts, self.last_value = 0, None
        self.buf = []            # (ts, v)
        self.max_len = max_len

    def push(self, ts, v):
        self.buf.append((ts, v))
        if len(self.buf) > self.max_len:
            self.buf.pop(0)
        changed = self.last_value is None or abs(v - self.last_value) >= self.deadband
        due = ts - self.last_report_ts >= self.period
        if changed or due:
            self.flush(ts, v)

    def flush(self, ts, v):
        payload = {
            "ts": ts,
            "cur": round(v, 2),
            "min": round(min(x for _, x in self.buf), 2),
            "max": round(max(x for _, x in self.buf), 2),
            "avg": round(sum(x for _, x in self.buf) / len(self.buf), 2),
            "n": len(self.buf)
        }
        # 这里换成实际的 MQTT / HTTP 上报
        print("上报:", json.dumps(payload, ensure_ascii=False))
        self.last_report_ts, self.last_value = ts, v
        self.buf.clear()

三种策略各管一段：周期上报保证“无论有没有变化，云端都知道设备还活着”；死区上报保证“有剧烈变化时立刻上报”；极值聚合保证“中间过程不会被完全丢掉”。另外要留一个本地环形缓冲区，断网时先存后补，网络恢复后按顺序重发——这是弱网现场最重要的可靠性设计。