📚 课程目录第3章读取 PLC 数据并映射到 OPC UA 节点

3. 读取 PLC 数据并映射到 OPC UA 节点

任务目标

本案例实现一个最小可运行的 OPC UA 网关：从 PLC 的保持寄存器读取数据，并将这些数据映射为 OPC UA 服务器上的节点，使 OPC UA 客户端能够通过标准地址空间读取到 PLC 中的实时值。案例使用 Modbus TCP 模拟 PLC 数据源，OPC UA 服务器采用开源库构建，整体逻辑聚焦于“读取-映射-发布”这条核心链路。

环境与依赖

Python 3.9+
安装依赖：pip install opcua pymodbus
使用 pymodbus 的同步客户端读取 PLC 寄存器
使用 opcua 库（FreeOpcUa）创建服务器并暴露节点

实现方案

网关进程启动后同时完成两件事：一是启动 OPC UA 服务器并创建自定义节点；二是在后台线程中周期性通过 Modbus TCP 读取 PLC 寄存器值，并更新 OPC UA 节点的数值。OPC UA 客户端连接后，直接读取或订阅该节点即可获得 PLC 数据。为便于演示，案例内置一个模拟 PLC 寄存器值的回调函数，实际部署时替换为真实 PLC 地址即可。

关键完整代码

import time
import threading
from opcua import Server, ua
from pymodbus.client import ModbusTcpClient

# ---------- 配置 ----------
PLC_HOST = "127.0.0.1"
PLC_PORT = 502
PLC_REGISTER_ADDR = 0          # 读取的保持寄存器起始地址
PLC_REGISTER_COUNT = 2         # 读取 2 个寄存器
OPC_UA_PORT = 4840
OPC_UA_NODE_ID = "ns=2;s=PLC_Data"

# ---------- 模拟 PLC 数据源（实际部署时替换为真实读取） ----------
def read_plc_registers():
    # 真实场景：client = ModbusTcpClient(PLC_HOST, port=PLC_PORT)
    # result = client.read_holding_registers(PLC_REGISTER_ADDR, PLC_REGISTER_COUNT)
    # return result.registers
    # 此处模拟两个寄存器的值，便于独立运行验证
    return [100 + int(time.time()) % 50, 200 + int(time.time()) % 80]

# ---------- OPC UA 服务器与映射逻辑 ----------
def update_loop(server, node):
    while True:
        values = read_plc_registers()
        # 将 PLC 寄存器值映射为 OPC UA 节点值（打包为 Variant）
        dv = ua.DataValue(ua.Variant(values, ua.VariantType.Int64))
        node.set_value(dv)
        time.sleep(1)  # 1 秒刷新一次

def main():
    # 1. 初始化 OPC UA 服务器
    server = Server()
    server.set_endpoint(f"opc.tcp://0.0.0.0:{OPC_UA_PORT}")
    server.set_server_name("PLC-to-OPCUA Gateway")

    # 2. 创建自定义节点（映射 PLC 数据）
    idx = server.register_namespace("http://plc.gateway")
    objects = server.get_objects_node()
    plc_node = objects.add_variable(idx, "PLC_Registers", [0, 0])
    plc_node.set_writable(False)  # 只读节点

    # 3. 启动服务器
    server.start()
    print(f"OPC UA 服务器已启动: opc.tcp://localhost:{OPC_UA_PORT}")

    # 4. 后台线程持续更新节点值
    t = threading.Thread(target=update_loop, args=(server, plc_node), daemon=True)
    t.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        server.stop()
        print("网关已停止")

if __name__ == "__main__":
    main()

运行与验证

启动网关：运行 python gateway.py，控制台输出“OPC UA 服务器已启动”即表示服务正常。
客户端读取验证：另开终端，使用 opcua-client 工具或以下 Python 脚本连接：

from opcua import Client

client = Client("opc.tcp://localhost:4840")
client.connect()
node = client.get_node("ns=2;s=PLC_Data")
value = node.get_value()
print("读取到 PLC 映射值:", value)
client.disconnect()

通过条件：客户端能成功连接并打印出包含两个整数的列表（如 [123, 245]），且间隔数秒后重复读取数值发生变化（因模拟数据基于时间变化），说明 PLC 数据到 OPC UA 节点的映射链路已打通。

常见问题

端口占用：若 4840 端口被占用，修改 OPC_UA_PORT 后重启。
连接真实 PLC：将 read_plc_registers() 函数体替换为注释中的真实 Modbus 读取代码，并确保 PLC 的 IP 与寄存器地址正确。
节点类型：本案例使用 Int64 数组承载寄存器值，若 PLC 数据为浮点或布尔，需相应调整 VariantType 与读取函数。