工业边缘节点Python编程：快速实现生产数据的预处理与异常告警

边缘节点跑Python，第一件事是划边界：PLC负责联锁和执行，Python只做采集、清洗、特征提取、本地告警。 别在Python里直接写输出继电器，网络一断控制逻辑就悬空。正确模型是“边采边洗、本地判异、断网缓存、恢复补传”。

一、数据记录先定型

每条数据必须带采集时间、源点名、质量位、序列号，别用“到达服务器时间”当时间戳——缓冲重连后全错。

class Sample:

   def __init__(self, tag, value, ts, seq):

       self.tag = tag

       self.value = value

       self.ts = ts          # 采集时刻，非上报时刻

       self.seq = seq

       self.quality = 1      # 0坏值 1好

同值不同时间不是重复，去重按(tag, seq)做，别按数值比。

二、预处理流水线

边缘侧先做四步：去坏值→重采样→滑窗统计→特征压缩。

from collections import deque

import numpy as np

class Preproc:

   def __init__(self, win=30, k=3.0):

       self.buf = deque(maxlen=win)

       self.k = k

   def push(self, s):

       # 1. 量程硬过滤

       if not (-40 <= s.value <= 200):

           s.quality = 0

           return None

       self.buf.append(s.value)

       if len(self.buf) < self.buf.maxlen:

           return None

       # 2. 滑动Z-score

       arr = np.array(self.buf)

       mu, std = arr.mean(), arr.std()

       if std < 1e-6:

           return s

       z = (s.value - mu) / std

       if abs(z) > self.k:

           s.quality = 0

           return None

       return s

振动类点位别传原始数组，先提RMS和主频：

rms = np.sqrt(np.mean(raw**2))

features = {"rms": rms, "peak": raw.max()}

原始1kHz流压成百字节JSON，上行带宽省90%。

三、异常告警：轻量优先

工业现场别一上来堆LSTM，3σ/Z-score最稳、最好解释。 真要消抖再加滞后区间，避免传感器抖一下狂报：

class Alarm:

   def __init__(self, hi, lo, dead=2.0):

       self.hi, self.lo = hi, lo

       self.dead = dead

       self.active = False

   def check(self, v):

       if self.active:

           if v < self.hi - self.dead:

               self.active = False

       else:

           if v > self.hi + self.dead:

               self.active = True

       return self.active

多级告警分开：

提示级：本地日志+MQTT

工艺级：本地声光/关阀联动

设备级：上报云平台+工单

断网时本地告警照常触发，不能等云端回执。

四、断网缓存与补传

Python侧维护本地队列（SQLite或文件），网络OK才批量发：

def report_loop(q, mqtt):

   while True:

       if net_ok():

           batch = q.pop(32)

           for s in batch:

               ok = mqtt.pub("tele/"+s.tag, to_json(s))

               if not ok:

                   q.push_front(s)      # 失败回队首

                   break

       else:

           time.sleep(2)

规矩：

队列定容，写满按FIFO丢最老，别撑爆磁盘

补传带seq，云端按(tag,seq)幂等去重

补传限速，别恢复瞬间打满上行

告警数据优先级高于遥测，先补告警

本地规则由云端下推，边缘留默认兜底

五、落坑清单

Python直接控PLC输出→断网失控

用到达时间当采集时间→重连后时序错

按数值去重→稳态同值被吞

原始振动流直传→带宽烧穿

单点Z-score不设死区→抖动误报

队列无限增长→磁盘写满宕机

补传不分优先级→告警被遥测堵住

云端规则漂移→边云判断不一致

异常检测不分段工况→升温过程狂报

六、落地顺序

定Sample结构：tag/ts/seq/quality

采集线程只入队，不碰业务判断

滑窗去坏值+Z-score过滤

振动/高频点本地提特征再发

告警加滞后死区，分级处理

本地SQLite队列缓存，net_ok才发

补传按seq幂等+限速+告警优先

本地规则云端下推，断网自治

边缘Python的价值不是“替代PLC”，是把云端扛不住的脏活在产线侧消化掉：坏值不往上送、原始流不占带宽、异常先在本地判、断网数据不丢。告警逻辑必须留在边缘——云端断链时现场还得自己知道超温超压。复杂模型可以后接，但基线一定是3σ+滑窗+队列缓存这套；跑稳了再上孤立森林或轻量自编码器，别反过来。