工控报警系统设计分级、去抖、风暴抑制、通知链一次讲透本文是报警主题的深度版。之前在《工控上位机 AI 编程实战》系列第九弹《设备报警系统设计不漏事、不烦人、能追溯》里用 1700 字讲过报警系统三要素和哪些代码适合交给 AI 写。这一篇换一个视角不讲交给谁写讲怎么设计才对——去抖双门槛的参数定法、风暴的全局阈值闸、通知链的升级计时、活动表历史表的数据模型全部给可直接抄的实现。两篇互为表里那篇看思路速览这篇看落地细节。先说结论报警系统做砸的标志不是不报警而是**“报得太多”**——满屏红色弹窗操作工全点确认不看内容真出事那天谁都没反应。好的报警系统有三个特征每条报警都有明确的动作该停机、该打电话、还是只是记录、抖动不重复打扰、风暴时合并成一条而不是刷一百条。这篇按分级 → 去抖 → 抑制 → 通知链 → 持久化五步给一套可直接抄的设计。一、先分级级别决定后面所有行为报警不是布尔值是带级别的。工控现场四级够用级别定义典型例子现场动作L1 提示不影响生产记录即可换班登录、参数被修改只写日志L2 警告性能劣化短时间内不损坏温度偏高、库存低水位界面黄条不弹窗L3 严重影响质量或设备安全通讯断线、气源压力低声光报警 弹窗 班长确认L4 紧急人身安全或重大损失风险安全门打开、超温超压停机联动 电话通知分级的第一原则是级别决定通知链而不是反过来。最常犯的错是这个点挺重要的报 L4 吧——于是半夜三点电话叫醒值班员处理一个完全可以等天亮的问题。定级的检验问题只有一个**“这条报警响的时候我希望谁在多少分钟内做什么动作”**答不上来的降级答得上来的才有资格占用人。二、去抖让报警持续为真才响现场信号天然抖传感器在阈值附近跳变、PLC 扫描周期里闪一个 0、通讯恢复瞬间的毛刺。不去抖的后果是同一件事响二十次操作工的确认动作被打成机械运动。解法是**报警延时on_delay 回差off_delay**双门槛importtimeclassAlarmPoint:单报警点触发需持续 on_delay 秒恢复需持续 off_delay 秒回差防边界抖动def__init__(self,tag,level3,on_delay3.0,off_delay5.0):self.tag,self.leveltag,level self.on_delay,self.off_delayon_delay,off_delay self.stateFalse# 当前是否处于报警态self._raw,self._sinceFalse,Nonedefupdate(self,raw:bool,now:float):每个周期喂一次原始信号返回报警事件或 Noneifself._sinceisNone:self._sincenowifraw!self._raw:# 原始信号翻转重新计时self._raw,self._sinceraw,nowifnotself.stateandrawandnow-self._sinceself.on_delay:self.stateTruereturn(ALARM,self.tag,self.level,now)ifself.stateandnotrawandnow-self._sinceself.off_delay:self.stateFalsereturn(CLEAR,self.tag,self.level,now)returnNone# 用法主循环里 1 秒喂一次pts[AlarmPoint( FurnaceTempHigh,4,on_delay2,off_delay10),AlarmPoint(AirPressureLow,3,on_delay5,off_delay5)]# while True: for ev in filter(None, (p.update(read_tag(p.tag), time.time()) for p in pts)): dispatch(ev)两个参数怎么定on_delay 取最短真实异常持续时间的 1/3——真实超温会持续几分钟那 on_delay 取 2~5 秒足够过滤毛刺off_delay 一般比 on_delay 长让恢复比触发更难防止边界上反复报警-恢复-报警这就是回差思想的时域版。三、风暴抑制一次根因只打扰一次去抖管单点风暴管全局。典型场景气源压力掉了一瞬间几十个气动阀全报阀未到位操作工收到 47 条报警而需要他处理的只有一条——气源问题。这就是报警风暴规则两条同源合并同一 tag 在时间窗内重复触发只通知第一次其余计数classFloodGate:窗口内同源报警只放行第一条其余静默计数窗口结束汇总一条def__init__(self,window60):self.window,self.counterswindow,{}deffilter(self,source:str,now:float)-str:last,cntself.counters.get(source,(None,0))iflastisNoneornow-lastself.window:ifcnt1:# 上个窗口被抑制过出一条汇总print(f[suppressed]{source}上窗口合并{cnt-1}条)self.counters[source](now,1)returnSENDself.counters[source](last,cnt1)returnMUTE全局阈值单位时间如 1 分钟内新报警超过 N 条比如 20系统判定当前处于风暴状态——切换为只发 L4、其余归档。这是承认一个事实风暴期间多发通知没有增量信息根因排查比逐条确认有价值得多。风暴结束后自动出一份汇总哪些报警、各多少次、首末时间一条消息交代清楚。四、通知链升级机制比通道多更重要通道就四种界面弹窗/声音/灯光、企业消息推送、短信、电话。设计重点不是堆通道而是升级计时L3 触发 → 界面声光 消息推送 → 10 分钟无人确认 → 短信 值班员 → 再 10 分钟仍无人确认 → 电话 班长并升级标记 L4 触发 → 界面 电话直呼 联动停机不经确认环节三个要点确认ack和恢复clear是两个独立事件——确认只是有人知道了恢复才是问题没了两者都要记人记时间升级只对未确认生效别让已确认的报警半夜还在打电话值班轮换表外置于配置别把电话号码写死在代码里——换班季节改代码是灾难。五、持久化活动表 历史表两张就够报警数据的价值在事后分析哪台设备报警最多、平均响应多久、哪类报警被确认了没人处理。模型只需要两张表——活动表当前未恢复的报警历史表闭环归档importsqlite3,timedefinit(dbalarms.db):consqlite3.connect(db)con.executescript( CREATE TABLE IF NOT EXISTS active( id INTEGER PRIMARY KEY, tag TEXT, level INT, raised_at REAL, ack_by TEXT, ack_at REAL); CREATE INDEX IF NOT EXISTS idx_active ON active(level, raised_at); CREATE TABLE IF NOT EXISTS history( id INTEGER PRIMARY KEY, tag TEXT, level INT, raised_at REAL, cleared_at REAL, ack_by TEXT, cnt INT); )returncondefraise_alarm(con,tag,level):ifcon.execute(SELECT 1 FROM active WHERE tag?,(tag,)).fetchone():return# 活动报警去重已在响就不再插con.execute(INSERT INTO active(tag,level,raised_at) VALUES(?,?,?),(tag,level,time.time()))con.commit()defclear_alarm(con,tag):rowcon.execute(SELECT id,level,raised_at,ack_by FROM active WHERE tag?,(tag,)).fetchone()ifnotrow:returncon.execute(DELETE FROM active WHERE id?,(row[0],))con.execute(INSERT INTO history(tag,level,raised_at,cleared_at,ack_by,cnt) VALUES(?,?,?,?,?,1),(tag,row[1],row[2],time.time(),row[3]))con.commit()注意活动表故意不存恢复时间——没恢复的报警就一直在表里看板读这张表就是实时报警列表一张 SQL 的事。历史表按月归档分析周报警 TOP10时直接聚合查询。库里报警数据的稳态运维备份、巡检、归档SQLite 运维那篇的三板斧原样适用。六、要点回顾先定级再动手级别希望谁在多少分钟内做什么答不上来就降级去抖双门槛触发要持续、恢复要更久边界抖动天然被过滤风暴两道闸同源窗口合并 全局阈值只放行 L4事后一份汇总确认与恢复分离、升级只盯未确认、联系方式外置配置活动表历史表两张表撑起实时看板和事后分析别做第三张。这套报警模块去抖点、风暴闸、SQLite 持久化、升级链骨架的完整可运行代码放在了作者付费资源里配套数据模型可直接进项目。报警数据再往上一层就是管理和消费——报表怎么自动出、看板怎么选型系列前面的文章里各有整篇打法。status: pending_review