IoT 场景示例

IoT 组件与 RuleGo 标准组件/端点的典型组合，展示完整数据链路。

场景一：定时采集 → 告警 → 通知

温度超阈值时发邮件告警。

endpoint/schedule(每30秒) → x/iotRead(S7) → jsFilter(温度>80?) → sendEmail(告警)
                                            → x/tsdbWrite(正常也落盘)

{
  "ruleChain": {"name": "temp-alarm", "root": true},
  "metadata": {
    "nodes": [
      {"id": "sch", "type": "endpoint/schedule", "configuration": {"interval": "@every 30s"}},
      {"id": "read", "type": "x/iotRead", "configuration": {
        "driver": "s7", "server": "192.168.1.10:102",
        "points": [
          {"name": "炉温", "addr": "DB1.DBD0", "type": "FLOAT32"},
          {"name": "压力", "addr": "DB1.DBD4", "type": "FLOAT32"}
        ]
      }},
      {"id": "filter", "type": "jsFilter", "configuration": {
        "jsScript": "var data = JSON.parse(msg.data || '[]'); return data.some(function(d){return d.name==='炉温' && d.value > 80;});"
      }},
      {"id": "email", "type": "sendEmail", "configuration": {
        "smtpHost": "smtp.example.com", "smtpPort": 465,
        "from": "alarm@example.com", "to": "ops@example.com",
        "subject": "【告警】炉温超限", "isHtml": false,
        "body": "炉温超过 80°C，当前数据：${msg.data}"
      }},
      {"id": "tsdb", "type": "x/tsdbWrite", "configuration": {
        "driver": "tdengine", "dsn": "root:taosdata@http(localhost:6041)/", "db": "iot",
        "measurement": "device_data",
        "tags": [{"key": "device_id", "value": "s7-01"}],
        "fields": [{"key": "temp", "source": "炉温"}, {"key": "pressure", "source": "压力"}]
      }}
    ],
    "connections": [
      {"fromId": "sch", "toId": "read", "type": "ip"},
      {"fromId": "read", "toId": "filter", "type": "Success"},
      {"fromId": "read", "toId": "tsdb", "type": "Success"},
      {"fromId": "filter", "toId": "email", "type": "True"}
    ]
  }
}

TDengine 落盘需预建超级表（子表写入时自动创建）：CREATE STABLE iot.device_data (ts TIMESTAMP, temp DOUBLE, pressure DOUBLE) TAGS (device_id NCHAR(32));

场景二：HTTP API 按需读取 → 返回 JSON

前端调 REST API 实时读 PLC 数据。

endpoint/http(POST /api/read) → x/iotRead(Modbus) → 响应

{
  "ruleChain": {"name": "api-read", "root": true},
  "metadata": {
    "nodes": [
      {"id": "http", "type": "endpoint/http", "configuration": {
        "server": ":9090", "certFile": "", "certKeyFile": ""
      }},
      {"id": "read", "type": "x/iotRead", "configuration": {
        "driver": "modbus", "server": "tcp://192.168.1.100:502",
        "points": [
          {"name": "temperature", "addr": "40001", "type": "INT16", "scale": 0.1},
          {"name": "humidity", "addr": "40002", "type": "INT16", "scale": 0.1}
        ]
      }}
    ],
    "connections": [
      {"fromId": "http", "toId": "read", "type": "POST /api/read"}
    ]
  }
}

场景三：MQTT 命令 → 写 PLC → 反馈

通过 MQTT 下发控制指令到 PLC，写结果回传 MQTT。

endpoint/mqtt(订阅 cmd/+) → jsTransform(解析命令) → x/iotWrite(S7) → mqttClient(发布结果)

{
  "ruleChain": {"name": "mqtt-control", "root": true},
  "metadata": {
    "nodes": [
      {"id": "mqtt_in", "type": "endpoint/mqtt", "configuration": {
        "server": "tcp://localhost:1883", "topic": "device/cmd/#"
      }},
      {"id": "parse", "type": "jsTransform", "configuration": {
        "jsScript": "var cmd = JSON.parse(msg.data); msg.data = JSON.stringify([{name:'设定温度', addr:'DB1.DBD0', type:'FLOAT32', value: String(cmd.targetTemp)}]); return {msg:msg, metadata:metadata, msgType:msgType};"
      }},
      {"id": "write", "type": "x/iotWrite", "configuration": {
        "driver": "s7", "server": "192.168.1.10:102"
      }},
      {"id": "mqtt_out", "type": "mqttClient", "configuration": {
        "server": "tcp://localhost:1883", "topic": "device/status/${metadata.deviceId}"
      }}
    ],
    "connections": [
      {"fromId": "mqtt_in", "toId": "parse", "type": "mqtt/device/cmd/#"},
      {"fromId": "parse", "toId": "write", "type": "Success"},
      {"fromId": "write", "toId": "mqtt_out", "type": "Success"}
    ]
  }
}

场景四：多协议并行采集 → 合并 → 落盘

同一规则链采集 Modbus 电表 + S7 PLC，合并后统一落盘。

endpoint/schedule → x/iotRead(Modbus电表) ─┐
                  → x/iotRead(S7 PLC)    ─┤→ join(合并) → x/tsdbWrite（配置 measurement）

{
  "ruleChain": {"name": "multi-protocol", "root": true},
  "metadata": {
    "nodes": [
      {"id": "sch", "type": "endpoint/schedule", "configuration": {"interval": "@every 1m"}},
      {"id": "meter", "type": "x/iotRead", "configuration": {
        "driver": "modbus", "server": "tcp://192.168.1.50:502",
        "points": [{"name": "power", "addr": "40013", "type": "FLOAT32", "scale": 0.1}]
      }},
      {"id": "plc", "type": "x/iotRead", "configuration": {
        "driver": "s7", "server": "192.168.1.10:102",
        "points": [{"name": "speed", "addr": "DB1.DBD0", "type": "FLOAT32"}]
      }},
      {"id": "merge", "type": "join", "configuration": {"joinInterval": 5}},
      {"id": "tsdb", "type": "x/tsdbWrite", "configuration": {
        "driver": "opengemini", "host": "127.0.0.1:8086", "database": "iot",
        "measurement": "factory", "tags": [{"key": "line", "value": "1"}]
      }}
    ],
    "connections": [
      {"fromId": "sch", "toId": "meter", "type": "ip"},
      {"fromId": "sch", "toId": "plc", "type": "ip"},
      {"fromId": "meter", "toId": "merge", "type": "Success"},
      {"fromId": "plc", "toId": "merge", "type": "Success"},
      {"fromId": "merge", "toId": "tsdb", "type": "Success"}
    ]
  }
}

场景五：Modbus 从站 → 数据转换 → REST 推送

SCADA 写 Modbus 寄存器 → 转换格式 → 推送到第三方 HTTP 接口。

endpoint/modbusServer → jsTransform(格式化) → restApiCall(POST 推送)

{
  "ruleChain": {"name": "scada-bridge", "root": true},
  "metadata": {
    "nodes": [
      {"id": "slave", "type": "endpoint/modbusServer", "configuration": {
        "server": "tcp://:5020", "unitId": 1
      }},
      {"id": "format", "type": "jsTransform", "configuration": {
        "jsScript": "var d = JSON.parse(msg.data); msg.data = JSON.stringify({device: 'PLC-01', register: d.addr, value: d.values[0], time: new Date().toISOString()}); return {msg:msg, metadata:metadata, msgType:msgType};"
      }},
      {"id": "push", "type": "restApiCall", "configuration": {
        "restEndpointUrlPattern": "http://third-party:8080/api/data",
        "requestMethod": "POST",
        "headers": {"Content-Type": "application/json"}
      }}
    ],
    "connections": [
      {"fromId": "slave", "toId": "format", "type": "ip"},
      {"fromId": "format", "toId": "push", "type": "Success"}
    ]
  }
}

场景六：SNMP Trap 告警 → 写数据库 → 企业微信通知

网络设备 Trap → 解析 → 告警记录入库 + 企微通知。

endpoint/snmp(Trap) → jsTransform(解析) → dbClient(写告警表)
                                         → restApiCall(企微 webhook)

{
  "ruleChain": {"name": "snmp-alarm", "root": true},
  "metadata": {
    "nodes": [
      {"id": "trap", "type": "endpoint/snmp", "configuration": {
        "server": "0.0.0.0:162", "version": "v2c", "community": "public"
      }},
      {"id": "parse", "type": "jsTransform", "configuration": {
        "jsScript": "var d = JSON.parse(msg.data); msg.data = JSON.stringify({source: d.from, oid: metadata.trapOID, time: Date.now()}); return {msg:msg, metadata:metadata, msgType:msgType};"
      }},
      {"id": "db", "type": "dbClient", "configuration": {
        "driverName": "postgres", "dsn": "postgres://user:pass@localhost:5432/alarm?sslmode=disable",
        "sql": "INSERT INTO alerts(source, oid, created_at) VALUES('${source}', '${oid}', NOW())"
      }},
      {"id": "wecom", "type": "restApiCall", "configuration": {
        "restEndpointUrlPattern": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY",
        "requestMethod": "POST",
        "headers": {"Content-Type": "application/json"}
      }}
    ],
    "connections": [
      {"fromId": "trap", "toId": "parse", "type": "ip"},
      {"fromId": "parse", "toId": "db", "type": "Success"},
      {"fromId": "parse", "toId": "wecom", "type": "Success"}
    ]
  }
}

软PLC式逻辑控制

x/control/timer（定时器）与 x/control/watchdog（看门狗）是协议无关的逻辑组件，在规则链里与 9 协议的读写节点串接，即可实现软PLC式的延时动作与失联保护。组件文档：控制定时器、看门狗。

读侧注意：x/iotRead 输出是点位数组 [{name,value,timestamp,error}]，进下游流聚合/定时器前先摊平成 {name:value}（一行 jsTransform）：

var d = JSON.parse(msg.data || '[]'); var out = {}; d.forEach(function(p){ if(!p.error) out[p.name] = p.value; }); msg.data = JSON.stringify(out); return {msg:msg, metadata:metadata, msgType:msgType};

场景七：罐体超温延时关阀（持续条件）

罐温连续 5 秒全程高于 80°C 才关闭进料阀。窗口聚合 HAVING MIN>80 = 全程过高，天然防抖（单次瞬时尖峰不触发）。

endpoint/schedule(每1秒) → x/iotRead(S7) → jsTransform(摊平) → x/streamAggregator(5s窗口) → x/iotWrite(关阀)

{
  "ruleChain": {"name": "tank-overtemp-shutoff", "root": true},
  "metadata": {
    "nodes": [
      {"id": "sch", "type": "endpoint/schedule", "configuration": {"interval": "@every 1s"}},
      {"id": "read", "type": "x/iotRead", "configuration": {
        "driver": "s7", "server": "192.168.1.10:102",
        "points": [{"name": "temperature", "addr": "DB1.DBD0", "type": "REAL"}]
      }},
      {"id": "flat", "type": "jsTransform", "configuration": {
        "jsScript": "var d = JSON.parse(msg.data || '[]'); var out = {}; d.forEach(function(p){ if(!p.error) out[p.name] = p.value; }); msg.data = JSON.stringify(out); return {msg:msg, metadata:metadata, msgType:msgType};"
      }},
      {"id": "agg", "type": "x/streamAggregator", "configuration": {
        "sql": "SELECT MIN(temperature) AS min_temp FROM stream GROUP BY TumblingWindow('5s') HAVING min_temp > 80"
      }},
      {"id": "write", "type": "x/iotWrite", "configuration": {
        "driver": "s7", "server": "192.168.1.10:102",
        "points": [{"name": "valve", "addr": "DB1.DBX4.0", "type": "BOOL", "value": "false"}]
      }}
    ],
    "connections": [
      {"fromId": "sch", "toId": "read", "type": "ip"},
      {"fromId": "read", "toId": "flat", "type": "Success"},
      {"fromId": "flat", "toId": "agg", "type": "Success"},
      {"fromId": "agg", "toId": "write", "type": "stream_event"}
    ]
  }
}

• HAVING min_temp > 80：窗口内只要有一笔采样跌回 80 以下，整个窗口被过滤；只有全程超温的窗口才产出，无窗口满足时聚合节点不输出、写节点不动作。
• HAVING 须引用 SELECT 别名（min_temp），不能复述聚合函数；聚合结果走 stream_event 关系链（Success 透传原始消息）。
• 写节点用配置点位（固定关阀 false），聚合输出只决定"是否触发"。

场景八：电机延时启动、中途可取消

启动信号上升沿 → 延时 3 秒 → 电机得电；延时期间信号消失则取消（TON 接通延时语义）。

endpoint/schedule(每1秒) → x/iotRead(读启动信号) → jsTransform(摊平并写 metadata.start) → x/control/timer(TON,3s) → x/iotWrite(电机)

{
  "ruleChain": {"name": "motor-delay-start", "root": true},
  "metadata": {
    "nodes": [
      {"id": "sch", "type": "endpoint/schedule", "configuration": {"interval": "@every 1s"}},
      {"id": "read", "type": "x/iotRead", "configuration": {
        "driver": "s7", "server": "192.168.1.10:102",
        "points": [{"name": "启动", "addr": "DB1.DBX0.0", "type": "BOOL"}]
      }},
      {"id": "flat", "type": "jsTransform", "configuration": {
        "jsScript": "var d = JSON.parse(msg.data || '[]'); var out = {}; d.forEach(function(p){ if(!p.error) out[p.name] = p.value; }); metadata.start = out['启动']; msg.data = JSON.stringify(out); return {msg:msg, metadata:metadata, msgType:msgType};"
      }},
      {"id": "timer", "type": "x/control/timer", "configuration": {
        "mode": "TON", "pt": "3s", "in": "${metadata.start}", "out": "q"
      }},
      {"id": "write", "type": "x/iotWrite", "configuration": {
        "driver": "s7", "server": "192.168.1.10:102",
        "points": [{"name": "电机", "addr": "Q0.0", "type": "BOOL", "value": "${metadata.q}"}]
      }}
    ],
    "connections": [
      {"fromId": "sch", "toId": "read", "type": "ip"},
      {"fromId": "read", "toId": "flat", "type": "Success"},
      {"fromId": "flat", "toId": "timer", "type": "Success"},
      {"fromId": "timer", "toId": "write", "type": "Success"}
    ]
  }
}

• 上升沿（false→true）开始计时，持续 pt=3s 后 metadata.q 置 true；输入提前变回 false 则计时取消、q 复位，下次上升沿重新触发。
• 定时器结果写入 metadata.q（out 字段），写节点以 ${metadata.q} 引用。

场景九：上位机失联 → 安全停机

上位机每隔几秒发心跳，连续 10 秒未收到则看门狗下发故障安全 JSON，关阀停电机。

endpoint/mqtt(订阅 scada/heartbeat/#) → x/control/watchdog(10s) → x/iotWrite(阀/电机)

{
  "ruleChain": {"name": "heartbeat-failsafe", "root": true},
  "metadata": {
    "nodes": [
      {"id": "hb", "type": "endpoint/mqtt", "configuration": {
        "server": "tcp://localhost:1883", "topic": "scada/heartbeat/#"
      }},
      {"id": "wd", "type": "x/control/watchdog", "configuration": {
        "timeout": "10s",
        "failsafe": {"valve": 0, "motor": 0}
      }},
      {"id": "write", "type": "x/iotWrite", "configuration": {
        "driver": "modbus", "server": "tcp://192.168.1.100:502",
        "points": [
          {"name": "阀门", "addr": "00001", "type": "BOOL", "value": "${msg.valve}"},
          {"name": "电机", "addr": "00002", "type": "BOOL", "value": "${msg.motor}"}
        ]
      }}
    ],
    "connections": [
      {"fromId": "hb", "toId": "wd", "type": "mqtt/scada/heartbeat/#"},
      {"fromId": "wd", "toId": "write", "type": "Success"}
    ]
  }
}

• 心跳正常到达：看门狗透传 msg.Data（心跳携带的阀/电机状态）到下游，并重新武装计时；timeout 内无消息则经 Success 链下发 failsafe {"valve":0,"motor":0}。
• 心跳载荷与 failsafe 使用相同 valve/motor 字段名，写节点统一以 ${msg.xx} 引用。

组件组合速查

场景	触发	采集/接收	处理	输出
定时采集落盘	endpoint/schedule	x/iotRead	—	x/tsdbWrite（配置 measurement）
超限告警	同上	同上	jsFilter	sendEmail / restApiCall
API 按需读	endpoint/http	x/iotRead	—	响应
MQTT 控制	endpoint/mqtt	—	jsTransform	x/iotWrite + mqttClient
多协议合并	endpoint/schedule	多个 x/iotRead	join	x/tsdbWrite
SCADA 桥接	endpoint/modbusServer	—	jsTransform	restApiCall / x/tsdbWrite
Trap 告警	endpoint/snmp	—	jsTransform	dbClient + restApiCall
环保数采	endpoint/hj212	—	—	x/tsdbWrite（配置 measurement）