一条报警消息的旅程：用华为云 IoTDA + FunctionGraph 做注塑机停机告警闭环

车间里最贵的不是电费，是停机。转盘机报警之后如果没人听见，机器就那么干等着，等班组长发现、等师傅从另一个车间过来。这次做的事情很简单：让报警自己找人，而不是等人找报警。

一、先把信号采出来，别急着上平台

老机器没有网口，PLC 也没有对外接口。能用的是两样东西：故障指示灯的输出触点，和继电器动作时的那一路 24V 信号。方案是用一块带干接点输入的采集网关，把三色灯和继电器各接一路进去，网关自己跑 MQTT 客户端。

这一步最关键的不是协议，是接哪一路。三色灯的黄灯和红灯代表不同级别：黄灯是缺料、模温不到这类可以自恢复的；红灯是机械卡死、液压超压这类必须停机处理的。分级必须在采集端就分好，否则后面所有逻辑都在猜。

{
  "device_id": "htf160-cavity4",
  "alarm_code": "E207",
  "level": "red",
  "ts": 1759452122000,
  "note": "转盘定位超时"
}

二、IoTDA 接入：用自定义 Topic，别用物模型

设备侧接华为云 IoTDA，最省事的做法是自定义 Topic 上报，格式定成 $oc/devices/{device_id}/sys/messages/up，payload 就是上面那段 JSON。物模型（属性上报）当然更规范，但老机器每次报警的字段结构不一样，硬套物模型要频繁改模型定义，反而不好维护。

接入时有两个参数必须确认：一是设备密钥的鉴权方式（推荐密钥认证 + 一机一密），二是 QoS 用 1，保证报警至少到一次。MQTT over 1883 在厂区内网直连没问题；跨公网记得换 8883 + TLS。

三、规则引擎转发到 FunctionGraph：这里做三件事

IoTDA 的规则动作配成「转发到 FunctionGraph」，函数收到原始报警后做三件事，都不复杂但都必要：

• 去重：同一台机同一报警码 60 秒内只推一次。现场的抖动非常频繁，继电器会连跳三四下，不去重就是报警风暴。
• 分级路由：红灯直接推到值班群 + 语音电话；黄灯只进群，不打电话。
• 落地：写一条记录到数据库，字段是机台、报警码、开始时间、恢复时间。没有这张表，事后想算停机率只能靠回忆。

import json, os, time
import requests

def handler(event, context):
    body = json.loads(event.get("body", "{}") or "{}")
    dev, code, level = body.get("device_id"), body.get("alarm_code"), body.get("level")
    key = "%s:%s" % (dev, code)
    # 去重：用函数实例的临时缓存不够可靠，正式环境建议换成 Redis
    now = int(time.time())
    if now - CACHE.get(key, 0) < 60:
        return {"statusCode": 200, "body": "dup"}
    CACHE[key] = now

    text = "【%s】%s 报警 %s" % ("停机" if level == "red" else "提示", dev, code)
    if level == "red":
        requests.post(os.environ["WEBHOOK"], json={"text": text}, timeout=5)
        requests.post(os.environ["CALL_API"], json={"text": text}, timeout=5)
    else:
        requests.post(os.environ["WEBHOOK"], json={"text": text}, timeout=5)
    return {"statusCode": 200, "body": "ok"}

CACHE = {}

函数的超时时间设 10 秒就够，但要注意冷启动：第一次调用可能要 1 到 3 秒。报警推送对这几秒不敏感，把它当实时系统设计反而会过度工程。

四、通知渠道：群里发一条，比什么大屏都管用

试过做大屏，结论是没人看。最后落在企业微信机器人的 webhook 上：一条消息里带机台号、报警码、发生时间，红灯加一个「已确认」的按钮链接。谁来处理就点一下，点完写回数据库，这条记录才闭合。

这里有个容易忽略的细节：恢复也要推。只推报警不推恢复，群里最后全是未闭合的告警，三天之后所有人都开始无视它。

五、四个坑，按踩到的顺序

• 时钟漂移：网关没有 NTP，跑一个月快了四分钟。报警时间和实际差四分钟，事后对齐停机记录全乱。解决办法是网关开机必须校时，服务端再按接收时间兜底。
• 报警风暴：一个传感器接触不良，一小时推了 200 多条。去重窗口从此变成硬需求，并且加了一条「同机台 5 分钟内超过 10 条报警 → 合并成一条汇总」。
• 离线判定：设备断网和机器正常运行在平台上看起来一样。加了心跳，5 分钟收不到心跳单独推一条「设备离线」，这条比报警更值得看。
• 规则引擎的匹配串写错一个字符就静默失效：不报错、不告警，只是消息不进来。上线前一定要用真实设备打一条报文，确认函数日志里有记录。

六、效果

上线两个月，平均停机响应从 28 分钟降到 3 分钟出头，主要是夜班受益——夜班只有一个人看十几台机，以前纯粹靠运气。真正省下来的不是修机时间，是「等发现」的那段时间。

这套链路没有用到任何高级能力：一个采集网关、一个 IoTDA 实例、一个函数、一个 webhook。复杂度全在细节里，而不在架构里。