摘要：在工业自动化（OT）与物联网（IoT）融合的场景下，现场设备（如 PLC、传感器、智能 PDU）通常使用 Modbus TCP 协议，而上层云端/边缘网关则倾向于使用 MQTT 协议。本文将分享如何通过 Python 实现 Modbus TCP 与 MQTT 的双向协议转换，驱动局域网嵌入式告警节点完成 RGB 全彩灯光指示 与 本地 TTS 语音播报，解决工业现场“信息孤岛”与“异常感知延时”问题。

一、 OT 与 IT 告警链路的融合架构

在智能工厂与边缘机房中，告警数据流往往需要兼顾低延时现场响应与云端状态同步：

+------------------------+             +------------------------+
| 工业 PLC / 传感器       |             | 边缘网关 / 云端 SCADA  |
| (Modbus TCP Server)    |             | (MQTT Broker)          |
+-----------+------------+             +-----------+------------+
            |                                      |
            | (Modbus 轮询)                        | (MQTT 订阅)
            v                                      v
+---------------------------------------------------------------+
|                   协议转换与告警路由适配层 (Python)             |
|  - Modbus 寄存器轮询与阈值比对                                |
|  - MQTT Topic 监听与 JSON 报文解析                             |
|  - 状态机控制与滑动窗口去重 (Debounce)                         |
+-------------------------------+-------------------------------+
                                |
                                | (REST API / HMAC 签名)
                                v
+---------------------------------------------------------------+
|                   嵌入式声光告警终端                           |
|  - 全彩 RGB LED 视觉矩阵 (常亮/闪烁/呼吸)                      |
|  - 本地离线 TTS 语音合成芯片 (自然语言播报)                     |
+---------------------------------------------------------------+

关键设计点：

1. 协议双通道：既能轮询底层 PLC 的 Modbus 保持寄存器（Holding Registers），又能订阅上层 MQTT Broker 的 JSON 告警主题。

2. 边缘本地化 TTS：告警文本在边缘节点直接转化为自然语音播报（如：“警告，2号变压器油温超限”），无需连接公有云 API，适应工业内网物理隔离环境。

二、 核心代码实现：Modbus & MQTT 双驱动适配服务

以下为完整可运行的 Python 服务代码，使用 pymodbus 轮询硬件，使用 paho-mqtt 接收云端指令，并通过安全 REST API 驱动声光终端：

Python

import time
import json
import requests
import hashlib
import hmac
from pymodbus.client import ModbusTcpClient
import paho.mqtt.client as mqtt

# 配置参数
ALARM_DEVICE_IP = "192.168.1.200"
API_KEY = "industrial_iot_node"
SECRET_KEY = "HMAC_Secret_Key_2026"

MODBUS_PLC_IP = "192.168.1.50"
MQTT_BROKER_IP = "192.168.1.100"
MQTT_TOPIC = "factory/alarm/event"

# 1. 计算 HMAC-SHA256 签名并推送至声光终端
def send_hardware_alarm(text, level="CRITICAL"):
    url = f"http://{ALARM_DEVICE_IP}/api/v1/send_msg"
    timestamp = str(int(time.time()))
    
    payload = {
        "text": text,
        "color": "#FF0000" if level == "CRITICAL" else "#FFA500",
        "light_mode": "flash" if level == "CRITICAL" else "steady",
        "audio_mode": "cycle" if level == "CRITICAL" else "once",
        "repeat_times": 3 if level == "CRITICAL" else 1
    }
    
    payload_str = json.dumps(payload, separators=(',', ':'))
    
    # 计算签名
    message = f"{timestamp}\n{payload_str}".encode('utf-8')
    signature = hmac.new(SECRET_KEY.encode('utf-8'), message, hashlib.sha256).hexdigest()
    
    headers = {
        "Content-Type": "application/json",
        "X-API-Key": API_KEY,
        "X-Timestamp": timestamp,
        "X-Signature": signature
    }
    
    try:
        resp = requests.post(url, data=payload_str, headers=headers, timeout=3)
        if resp.status_code == 200:
            print(f"[Success] 现场声光已触发: {text}")
    except Exception as e:
        print(f"[Error] 告警终端通信失败: {e}")

# 2. MQTT 消息回调函数 (IT 侧告警)
def on_mqtt_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload.decode('utf-8'))
        device_name = data.get("device", "未知设备")
        alarm_msg = data.get("message", "发生异常")
        severity = data.get("severity", "WARNING")
        
        tts_text = f"MQTT预警，{device_name}{alarm_msg}"
        send_hardware_alarm(tts_text, level=severity)
    except Exception as e:
        print(f"[MQTT Error] 解析报文失败: {e}")

# 3. Modbus TCP 轮询任务 (OT 侧告警)
def poll_modbus_plc():
    client = ModbusTcpClient(MODBUS_PLC_IP, port=502)
    if not client.connect():
        print("[Modbus Error] 无法连接 PLC")
        return

    # 读取保持寄存器 40001 (地址 0)
    result = client.read_holding_registers(address=0, count=1)
    if not result.isError():
        temp_val = result.registers[0]
        # 假设寄存器数值代表温度，超过 80 度触发告警
        if temp_val > 80:
            tts_text = f"PLC告警，1号反应斧温度达到 {temp_val} 度，超过安全阈值"
            send_hardware_alarm(tts_text, level="CRITICAL")
    
    client.close()

# 主运行循环
def main():
    # 初始化 MQTT 客户端
    mqtt_client = mqtt.Client()
    mqtt_client.on_message = on_mqtt_message
    
    try:
        mqtt_client.connect(MQTT_BROKER_IP, 1883, 60)
        mqtt_client.subscribe(MQTT_TOPIC)
        mqtt_client.loop_start()
        print("[System] MQTT 监听已启动...")
    except Exception as e:
        print(f"[MQTT Warning] MQTT Broker 连接失败，仅启动 Modbus 轮询: {e}")

    # 主循环轮询 Modbus
    while True:
        poll_modbus_plc()
        time.sleep(2) # 每 2 秒轮询一次

if __name__ == "__main__":
    main()

三、 工业现场的防扰与高可用设计

1. 状态去重（Debounce）与防风暴： 在工业传感器数据抖动时（如温度在 79~81 度之间反复震荡），可能引发高频重复告警。建议在适配层维护一个 LastAlarmTime 哈希表，同一节点在 180 秒窗口内仅触发一次 TTS 语音。

2. RGB 颜色编码标准： 在光线复杂的车间内，建议统一灯光语义：

   • 红色闪烁 (#FF0000)：P0 级严重故障，需紧急停机。

   • 黄色常亮 (#FFCC00)：P1 级参数超限，提示排查。

   • 绿色常亮 (#00FF00)：故障复位/系统正常运行。

3. 分时段静音策略： 夜间值班时段，可通过 REST API 动态调整终端参数，将 audio_mode 设为 none，仅保留 LED 矩阵闪烁，防止高分贝语音骚扰。

四、 总结

通过构建 Modbus TCP + MQTT -> 离线声光终端 的解耦架构，我们成功将 OT 层的设备状态与 IT 层的消息流进行了有效融合。离线 TTS 与 RGB 灯光的结合，使现场巡检人员可以在第一时间内“听得懂、看得准”，极大地缩短了工业异常处理的平均修复时间（MTTR）。