边缘计算：物联网架构演进

划分职责是边缘项目最容易出错的地方：把该在本地做的判断放到了云上，或者把本该集中管理的数据散落在现场的网关里。可以参考下面这张职责表。

一个判断标准很好用：如果这个逻辑在断网时必须继续工作，它就应该放在端或边；如果这个逻辑需要跨设备、跨站点才能算出来，它就应该放在云上。按这个标准切分，边界通常会自然浮现。

“边缘”落地时是一台具体的机器，形态差异很大，选型时按算力需求分三档。

• 协议网关型：算力最低，主要工作是协议转换（Modbus/OPC UA 转 MQTT）与数据转发，常见于设备替换成本高的老旧产线。
• 通用计算型：x86 或 ARM 多核，能跑容器与数据库，适合做规则引擎、边缘缓存、轻量视觉处理。
• 带 NPU 的推理型：如集成 NPU 的 ARM SoC，可以在本地跑量化后的视觉或振动模型，适合质检、预测性维护这类需要模型推理的场景。

边缘最直接的价值是“把原始流变成结论”。下面这段例子把 200ms 的高频采样在本地压成每秒一条统计结果，并只在越限时立即上报。

import json, time
from collections import defaultdict

class EdgeAggregator:
    """本地聚合 + 越限即时上报"""
    def __init__(self, window=1.0, hi=85.0, lo=5.0):
        self.window, self.hi, self.lo = window, hi, lo
        self.buf = defaultdict(list)
        self.start = time.time()

    def feed(self, tag, ts, value):
        self.buf[tag].append((ts, value))
        # 越限：不走聚合，立刻上报，保证告警时效
        if value > self.hi or value < self.lo:
            self.emit_alarm(tag, ts, value)
        if ts - self.start >= self.window:
            self.flush(ts)

    def emit_alarm(self, tag, ts, v):
        print("ALARM", json.dumps({"tag": tag, "ts": ts, "v": v, "level": "high" if v > self.hi else "low"}))

    def flush(self, ts):
        out = []
        for tag, pts in self.buf.items():
            vs = [v for _, v in pts]
            out.append({"tag": tag, "n": len(vs), "avg": round(sum(vs) / len(vs), 2),
                        "max": max(vs), "min": min(vs)})
        if out:
            print("UPLOAD", json.dumps({"ts": ts, "points": out}, ensure_ascii=False))
        self.buf.clear()
        self.start = ts

这段逻辑把数据量压缩了一到两个数量级：现场每秒 5 次采样，上云每秒 1 条统计；而真正重要的越限事件反而比原来更快送达。这就是“能压缩成结论的数据就不要传原始流”的具体含义。

带 NPU 的网关越来越多，但把训练好的模型搬上去并不只是复制文件。核心问题是资源约束：内存小、算力有限、往往没有 GPU。常见的做法是先量化再部署，INT8 量化通常能把模型体积压到原来的四分之一，推理速度提升数倍，而精度损失通常可以接受。

# 边缘推理示例：ONNX Runtime（CPU/NPU 后端）
import numpy as np, onnxruntime as ort

sess = ort.InferenceSession(
    "models/vibration_int8.onnx",
    providers=["CPUExecutionProvider"]     # 有 NPU 时换成厂商 EP
)
input_name = sess.get_inputs()[0].name

def predict(feature_vec):
    x = np.asarray([feature_vec], dtype=np.float32)   # 形如 (1, 1024)
    logits = sess.run(None, {input_name: x})[0]
    prob = float(1 / (1 + np.exp(-logits[0][0])))     # 二分类：异常概率
    return prob

# 现场侧只上报结论，不上传原始波形
prob = predict(load_vibration_frame())
if prob > 0.85:
    publish("site/jjy/edge/gw-01/alert",
            {"level": "warn", "type": "bearing", "prob": round(prob, 3)})