Python Virtual RTU/IIoT Simulator with IEC-20922 MQTT Communication Protocol

Updated 1,301 0
English 简体中文 繁体中文
Summary

This project enhances a Python virtual RTU/IIoT simulator by integrating IEC-20922 (MQTT) support for industrial cyber twin development and OT security research. The enhanced simulator features an MQTT communication module handling publish-subscribe messaging and an RTU/IIoT framework that models field device operations, manages virtual I/O, and executes control logic. It adheres to a four-level ISA-95 architecture, demonstrating its application through an IoT drone telemetry system and a smart factory air vacuum control system. The simulator is designed for educational, prototyping, and cybersecurity analysis purposes.

Python 虚拟 RTU/IIoT 模拟器，支持 IEC-20922 MQTT 通信协议

项目设计目的 : 在本项目中，我扩展了我之前基于 Python 的虚拟 PLC/RTU 模拟器库（该库通过 Modbus-TCP 和 S7Comm 与 SCADA 系统接口，相关链接：https://www.linkedin.com/pulse/python-virtual-plc-rtu-simulator-yuancheng-liu-elkgc），增加了对 IEC-20922 Message Queuing Telemetry Transport (MQTT) 协议的支持功能。新功能设计包含两个主要组件：

  • MQTT 通信模块 : MQTT 通信模块实现了符合 IEC 20922 标准的 MQTT 协议栈，提供了虚拟设备与 MQTT 代理之间的连接，支持消息发布和订阅、主题管理、遥测数据交换以及命令/控制通信。

  • RTU/IIoT 模拟器框架 : RTU/IIoT 模拟器框架模拟了工业现场设备、远程终端单元 (RTU) 和 IIoT 传感器的运行行为。它管理数字孪生体的虚拟设备输入和输出，处理 MQTT 消息，与物理世界模拟模块接口，并执行用户定义的控制逻辑。

# Author:      Yuancheng Liu
# Created:     2026/06/01
# Version:     v_0.0.1
# Copyright:   Copyright (c) 2026 Liu Yuancheng
# License:     MIT License

1. 项目介绍

Message Queuing Telemetry Transport (MQTT) 协议，标准化为 IEC 20922，是一种轻量级的发布-订阅消息协议，专为资源受限设备和低带宽无线网络设计。由于其简单性、可扩展性和低通信开销，MQTT 已成为工业物联网 (IIoT) 领域以及制造业、能源、交通和智能基础设施等行业的机器对机器 (M2M) 通信中最广泛采用的通信标准之一。

在典型的 IIoT 部署中，现场设备将运行数据发布到中央 MQTT 代理，而监控系统、人机界面 (HMI)、移动应用程序和监控平台则订阅所需的数据流。这种解耦的通信模型简化了系统集成，并为大规模工业监控和控制系统提供了灵活的架构。下面展示了 MQTT 与 IIoT/RTU/PLC 的使用案例示例：

Image Reference : https://macautoinc.com/industrial-communication-protocols/mqtt/

为了支持工业数字孪生体和 OT 网络安全研究平台的开发，本项目将提供一套可重用的 MQTT Broker 和 MQTT Client 模块，这些模块可以集成到不同的数字孪生体组件中。

1.1 系统概述

该模拟器项目 不 是 1:1 模拟真实的 RTU/IIoT/MU 硬件功能（不是数字孪生体），它专注于重现支持 MQTT 的工业设备中常见的核心运行行为，包括：

  • 设备变量和标签存储管理

  • MQTT 发布和订阅通信机制

  • 遥测和控制数据交换工作流程

  • 设备控制逻辑执行周期

  • 现场设备、控制器和监控系统之间的交互

而系统应用的主要目的是提供有效的教育、原型设计和研究环境，例如：

  • 研究工业自动化和 IIoT 架构的学术研究人员

  • 学习 OT 通信协议和 MQTT 设备行为的学生

  • 开发、测试或验证支持 MQTT 的应用程序的开发人员

  • 分析工业通信流和攻击场景的 OT 网络安全专业人员

1.2 系统 ISA-95 架构

该模拟器使用户能够构建反映现代工业环境中常见的分层架构的数字孪生体组件。如下图所示，该框架遵循基于 ISA-95 模型简化的四级 OT 架构，如下图所示：

  • 在 0 级（物理过程现场 I/O 设备），模拟的 IIoT 设备、传感器和计量单元生成运行数据，代表从物理过程中收集的测量值。在 1 级（控制器局域网），虚拟 RTU 处理传入数据并作为 MQTT 客户端运行，将遥测和状态信息发布到 MQTT 代理。

  • 位于 2 级（控制中心处理局域网） 的 MQTT 代理服务器，充当中枢通信中心。它接收来自现场设备的发布消息，管理主题订阅，存储设备数据，并在需要时执行服务器端处理逻辑。同一网络段内的控制 HMI 和操作员控制台也可以通过代理订阅或发布 MQTT 消息。

  • 在 3 级（运营管理区），监控工作站、工程桌面、移动设备和触摸屏操作员面板等监控应用程序运行 MQTT 客户端服务，以订阅设备数据、可视化过程信息并发出控制命令。

2. MQTT 协议背景知识

Message Queuing Telemetry Transport (MQTT) 是一种轻量级消息协议，标准化为 IEC 20922。它遵循 发布-订阅通信模型，其中设备之间不直接通信。相反，所有消息都通过中央 MQTT 代理 进行交换。

在 MQTT 系统中，充当 发布者 的设备将数据发送到代理托管的特定主题，而 订阅者 则接收它们感兴趣的主题的消息。这种架构降低了通信复杂性，提高了可扩展性，并实现了在低带宽或不可靠网络上的高效运行。

2.1 MQTT 协议数据包结构

MQTT 通信通过客户端和代理之间交换的一系列协议数据包进行。无论数据包类型如何，每个 MQTT 数据包都包含三个逻辑部分：

  1. 固定头 (必需)

  2. 可变头 (可选)

  3. 有效载荷 (可选)

通用的 MQTT 数据包结构如下图所示：

有关详细的数据包分析，请参阅以下文档：

这是一个将每个数据包部分映射到捕获的 MATT 连接请求消息的示例：

2.2 MQTT 协议关键特性

轻量级头： 协议数据包非常小（通常只有几个字节），可以节省带宽、内存和电池寿命。

服务质量 (QoS)： 开发人员可以选择交付保证级别：

  • QoS 0（最多一次）： 快速交付，但消息可能会丢失。

  • QoS 1（至少一次）： 保证交付，但可能出现重复。

  • QoS 2（仅一次）： 消息仅传递一次，无丢失或重复。

遗嘱消息 (LWT)： 允许设备在代理处预先注册一条消息，如果设备意外离线，该消息将被自动广播。

3. MQTT 虚拟 IIoT 和 RTU 设计

本节介绍 MQTT 通信模块的详细设计，并演示如何将其集成到数字孪生体环境中。提供了两个示例应用程序：

  • 模拟的智能工厂真空控制系统

  • 物联网无人机遥测接收系统

这些示例说明了如何将基于 MQTT 的通信纳入工业控制架构的不同层。

3.1 MQTT 通信模块设计

MQTT 通信框架包含两个主要组件：MQTT 代理模块和 MQTT 客户端模块，用于提供模拟现场设备、控制器和监控应用程序之间数据交换所需的通信基础设施。

3.1.1 MQTT 代理设计

对于 MQTT 代理模块，当前实现的 MQTT 数据包类型如下所示。

# MQTT 数据包类型常量（当前所需，未来可能添加更多）
CONNECT     = 0x10	# 与 MQTT 代理建立连接
CONNACK     = 0x20	# 来自代理的连接确认
PUBLISH_Q0  = 0x30  # QoS 级别 0（最多一次），目前我们使用 QoS 级别 0 DUP = 0, Retain = 0
PUBLISH_Q1  = 0x32  # QoS 级别 1（至少一次）
PUBLISH_Q2  = 0x34  # QoS 级别 2（仅一次）
PUBACK      = 0x40	# 发布确认
SUBSCRIBE   = 0x82	# 订阅一个或多个主题
SUBACK      = 0x90	# 订阅确认
PINGREQ     = 0xC0	# 保活请求
PINGRESP    = 0xD0	# 保活响应
DISCONNECT  = 0xE0	# 正常终止连接

对于每个代理模块，当新的 MQTT 客户端与其建立连接时，会创建一个专用的客户端处理程序线程，以独立管理发布和订阅请求。这种多线程架构允许多个 MQTT 客户端同时与代理通信。

为了简化参数访问和标准化数据交换，使用了以下主题命名约定：

主题模式 目的 请求类型
parameters/get/ 请求参数的当前值 发布
parameters/set/ 更新参数的值 发布
parameters/value/ 订阅参数值更新 订阅

除了基本的消息路由外，代理模块还提供了一个名为 executeLogic() 的空接口函数，允许用户实现自定义数据处理和控制算法，如下所示：

def executeLogic(self):
    """ MQTT 代理在主循环中执行控制逻辑的接口函数。"""
	pass

用户可以通过继承基类 MQTTBroker 并用特定于应用程序的逻辑覆盖此函数来创建自定义代理。该函数在通过发布请求更新参数值时自动触发。它也可以在主执行循环中定期调用以执行计划的数据处理任务。

为了简化并最大程度地兼容不同的模拟设备，所有参数值在内部都存储为 string 数据类型。在需要时，可以通过特定于应用程序的逻辑执行类型转换。

3.1.2 MQTT 客户端设计

MQTT 客户端模块使用 Eclipse Paho MQTT 库实现 https://pypi.org/project/paho-mqtt/ 。该客户端提供四个函数供数字孪生体组件使用：

  • getParmVal() – 从代理检索参数值。

  • setParmVal() – 在代理上更新参数值。

  • watch() – 订阅指定的参数主题。

  • watchall() – 订阅所有可用的参数主题。

3.2 数字孪生体集成设计

为了演示 MQTT 通信框架的用法，开发了两个数字孪生体组件：一个物联网无人机遥测系统和一个智能工厂真空控制系统。整体集成架构如下所示。

3.2.1 物联网无人机遥测系统

在物联网无人机模拟系统中，无人机模拟器充当 MQTT 客户端，并持续将原始飞行遥测数据发布到 MQTT 代理。传输的数据包括：滚转角、俯仰角、偏航角、高度、速度 和 GPS 位置 等信息。

MQTT 代理执行自定义处理逻辑，将原始传感器测量值转换为人类可读的飞行状态信息。处理后的结果存储在代理的参数数据库中。

无人机状态监控控制台充当另一个 MQTT 客户端，并订阅处理后的遥测主题。通过持续接收来自代理的更新，控制台提供无人机运行状态和飞行条件的实时可视化。

3.2.2 智能工厂真空控制系统

智能工厂真空系统演示了 MQTT 通信与传统 RTU 控制架构的集成。

在物理过程模拟器中，温度传感器、压力传感器和风扇速度传感器生成模拟测量值，这些测量值通过基于 UDP 的电信号模拟通道传输到 RTU。RTU 将这些测量值存储在其内部参数字典中，并执行本地控制逻辑以确定适当的控制操作。

根据当前运行状况，RTU 可能会发出控制命令来：

  • 风扇电机控制器

  • 气管阀控制器

  • 其他模拟现场设备

这些控制命令通过与物理过程模型相同的模拟电信号通道进行传输。

运行在 RTU 内的 MQTT 客户端将运行数据和系统状态信息发布到 MQTT 代理。工厂环境监控 HMI 订阅这些 MQTT 主题，并通过实时仪表板显示信息。

MQTT 基础设施也支持监控控制。当操作员通过控制仪表板修改系统配置时，命令将被发布到 MQTT 代理。RTU 订阅相关的控制主题并接收更新的设置。这些操作员发出的命令可以覆盖 RTU 的自动控制决策，从而在需要时允许手动干预。

接收到新的控制参数后，RTU 会更新其内部控制状态，并向风扇电机控制器发出相应的命令，并打开气管阀，使物理过程模拟器做出相应响应。

为了支持快速维护操作和工程测试，RTU 还托管了一个轻量级的嵌入式 MQTT 代理，直接链接到选定的控制参数。此功能使工程控制台和本地 HMI 能够在不经过中央 MQTT 基础设施的情况下执行低延迟的直接控制和参数修改。

 

4. 用例示例

为了演示 MQTT 虚拟 RTU/IIoT 模拟器的用法，本节介绍了一个简单的风扇控制器 RTU 实现。该示例展示了如何集成 MQTT 代理和 MQTT 客户端模块来创建一个能够执行自动控制逻辑，同时通过 MQTT 暴露遥测和控制接口的 RTU。

以下 Python 模块作为基础示例提供，可以扩展以构建更复杂的支持 MQTT 的工业设备模拟器。

程序文件 执行环境 描述
src/mqttComm.py python 3.7+ 核心库，实现了 IEC-20922 MQTT 客户端/代理 API，用于模拟 IIoT/RTU 与 SCADA 软件之间的数据和命令交互。
src/mqttCommTest.py python 3.7+ 的测试用例模块。它在一个后台线程中启动一个 MQTT 代理服务，创建两个客户端，并测试参数值发布/订阅操作和控制逻辑。
testcase/mqtRtuClientTest.py python 3.7+ 该模块是一个简单的 RTU 连接程序，使用 MQTT 库模块 来模拟一个 SCADA 设备，其中包含一个 MQTT 客户端，用于连接到 ，设置参数的随机值并读取相关响应以验证结果。
testcase/mqtRtuBrokerTest.py python 3.7+ 该模块是一个简单的 RTU 模拟程序，使用库模块 来模拟一个 RTU，其中包含一个 MQTT 代理和一个自动控制逻辑，用于处理来自客户端的可变读取和可更改值设置。

4.1 实现 Broker 控制逻辑

在此示例中，在 MQTT Broker 中实现了一个自动风扇控制器。控制要求很简单：

  • 如果运行模式设置为“自动”且温度超过 50°C，则风扇开启并将风扇速度设置为 50%。

  • 否则，风扇关闭。

为了实现此行为，通过继承基类 MQTTBroker 并覆盖 executeLogic() 接口函数来创建一个自定义的 broker 类。

class TestBroker(mqttComm.MQTTBroker):
    """ 测试 broker 类，它将在子线程中启动一个 broker 和多个客户端
        来测试数据的读取和传输。
    """
    def __init__(self, brokerName='testBroker', brokerPort=1883):
        super().__init__()
        self.mqttClients = []
    def executeLogic(self):
        """ 风扇控制逻辑覆盖了 mqttComm.MQTTBroker 中的 executeLogic。 """
        print("> 执行控制逻辑")
        temp = float(self.getParmVal('temperature'))
        mode = self.getParmVal('mode')
        if mode == 'auto':
            if temp > 50:
                self.setParmVal('fan', 'on')
                self.setParmVal('fanSpeed', '50')
            else:
                self.setParmVal('fan', 'off')
                self.setParmVal('fanSpeed', '0')

4.2 实现 RTU 模块

RTU 模拟器在一个专用的后台线程中托管 MQTT Broker，并初始化控制应用程序所需的参数。

class MQTTbrokerThread(threading.Thread):
    """ MQTT broker 线程类。 """
    def __init__(self):
        threading.Thread.__init__(self)
        self.mqttBroker = TestBroker()
        self.mqttBroker.addParm('temperature', '25.0')
        self.mqttBroker.addParm('mode', 'manual')
        self.mqttBroker.addParm('fan', 'off')
        self.mqttBroker.addParm('fanSpeed', '0')

RTU 定期与物理过程模拟器交换数据。在每个执行周期中，从模拟环境中收集传感器值，存储在 MQTT Broker 参数数据库中，并由自动控制算法处理。

    def run(self):
        while not self.terminate:
            self.fetchDataFromPhysicalWorld()
            self.brokerObj.setParmVal('temperature', str(self.srcValDict['temperature']))
            self.brokerObj.executeLogic()
            time.sleep(0.1) # 休眠 0.1 秒等待 broker 的
            self.srcValDict['autoMode'] = self.brokerObj.getParmVal('mode') == 'auto'
            self.destValDict['fanPwr'] = self.brokerObj.getParmVal('fan') == 'on'
            self.destValDict['fanSpeed'] = int(self.brokerObj.getParmVal('fanSpeed'))
            print("源参数: %s" % str(self.srcValDict))
            print("目标参数: %s" % str(self.destValDict))

4.3 实现 RTU 连接器模块

外部应用程序，如 HMI、SCADA 系统、工程工作站或监控仪表板，可以通过 MQTT Client 模块与 RTU 通信。

要建立连接，将创建一个客户端对象并连接到 Broker，如下所示：

class RtuConnector(object):
    def __init__(self, clientId, host, port=1883):
        self.mHost = host
        self.mPort = port
        self.mClientId = clientId
        self.client = mqttComm.MQTTClient(clientId, host, port=1883)
        self.client.connect()
    #-----------------------------------------------------------------------------
    def startTest(self):
        index = 0
        print("开始客户端测试")
        index += 1
        print("\n测试-%d 从服务器订阅数据。" % index)
        temp = self.client.getParmVal('temperature')
        mode = self.client.getParmVal('mode')
        fan = self.client.getParmVal('fan')
        fanS = self.client.getParmVal('fanSpeed')

5. 项目仓库链接

本项目扩展了一个基于 Python 的虚拟 PLC/RTU 模拟器库，支持 IEC 20922 (MQTT) 协议，为工业数字孪生和 OT 安全研究实现了发布-订阅通信。实现包括两个主要组件：一个符合标准数据包结构的 MQTT 通信模块（支持 CONNECT、PUBLISH、SUBSCRIBE、PING、DISCONNECT），以及一个 RTU/IIoT 模拟器框架，该框架模拟现场设备行为、管理虚拟 I/O 并执行用户定义的控制逻辑。该系统遵循四级 ISA‑95 架构，从模拟的物理过程设备到运营管理应用程序。提供了两个演示用例：一个物联网无人机遥测系统和一个智能工厂真空控制系统。一个简单的风扇控制器示例说明了如何实现自定义 Broker 逻辑并集成 MQTT 客户端进行遥测交换和监控控制。该模拟器可用于教育、原型设计和网络安全分析，服务于研究人员、学生、开发人员和 OT 专业人员。

项目仓库链接：

https://github.com/LiuYuancheng/PLC_and_RTU_Simulator/tree/main/MQTT_RTU_Simulator

感谢您花时间阅读文章详情，如果您有任何问题、建议或发现任何程序错误，请随时给我留言。如果您能提供一些评论和改进建议，我们将不胜感激，以便我们能做得更好~

 

  RELATED

No related programming articles found. Browse all programming tutorials and articles.

  COMMENTS

0

No comment for this article.