Code Examples

常见集成场景的最小可运行示例：4-20 mA 换算、Modbus RTU 读取、MQTT 上行、SDI-12 命令、Hampel 滤波。

示例遵循一个共同约定：单位与质量标记随数据一起流动，不允许出现无单位的裸数字。

所有示例都包含错误处理：超量程、缺失、校验失败。

# 1) 4-20 mA → 工程量（含故障诊断）
def eu_from_current(i_ma, eu_min, eu_max):
    if i_ma < 3.8:          # 活零点以下：断线或故障
        return None, "outside_nominal_range: consult_device_config"
    if i_ma > 20.5:
        return None, "outside_nominal_range: consult_device_config"
    return eu_min + (i_ma - 4.0) / 16.0 * (eu_max - eu_min), "ok"

print(eu_from_current(12.0, 0, 10))     # (5.0, 'ok')

# 2) Modbus RTU 读取 float32（注意字节序）
import struct
def read_float(regs, byte_order="ABCD"):
    b = b"".join(r.to_bytes(2, "big") for r in regs)
    if byte_order == "CDAB": b = b[2:4] + b[0:2]
    if byte_order == "BADC": b = b"".join(x[::-1] for x in [b[0:2], b[2:4]])
    if byte_order == "DCBA": b = b[::-1]
    return struct.unpack(">f", b)[0]     # 大端 float32

# 3) Hampel 滤波：保留阶跃、剔除尖峰
import numpy as np
def hampel(x, window=5, n_sigma=3.0):
    x = np.asarray(x, dtype=float); out = x.copy()
    k = window // 2
    for i in range(k, len(x) - k):
        w = x[i-k:i+k+1]
        m = np.median(w)
        mad = 1.4826 * np.median(np.abs(w - m))
        if mad > 0 and abs(x[i] - m) > n_sigma * mad:
            out[i] = m
    return out