Gateway OPC UA a MQTT: lleva los datos del PLC al cloud

Un gateway OPC UA a MQTT lee tags de los PLC y de los servidores OPC UA de la red de planta y los publica como mensajes MQTT en un broker al que pueden suscribirse dashboards, historiadores y aplicaciones cloud. En este tutorial construirás uno en Python con librerías de código abierto: se suscribe a nodos OPC UA, los mapea a topics de estilo ISA-95 y publica JSON limpio en un broker en la nube. Puedes ejecutarlo todo en un portátil en unos quince minutos.

¿Por qué hacer de puente entre OPC UA y MQTT?

OPC UA es excelente en la planta. Tiene un modelo de información rico, seguridad integrada y un amplio soporte entre los fabricantes de PLC. Pero es un protocolo cliente/servidor: cada consumidor abre su propia sesión con cada servidor. Eso funciona para un SCADA que habla con diez PLC, y se vuelve incómodo cuando veinte aplicaciones, algunas en el cloud, quieren los mismos datos.

MQTT invierte esa relación. El gateway mantiene una sesión OPC UA por servidor y una única conexión MQTT saliente, y cualquier número de consumidores se suscribe al broker. El firewall de planta solo necesita una conexión saliente al puerto 8883, y los PLC nunca ven tráfico cloud. Si estás comparando ambos protocolos de forma más amplia, nuestra comparativa OPC UA vs MQTT trata las ventajas e inconvenientes; en resumen, la mayoría de las arquitecturas usan los dos.

Arquitectura del gateway

 PLC / OPC UA server            Edge gateway (this tutorial)              MQTT broker
 opc.tcp://plc:4840   ---->  asyncua subscription  ---> paho-mqtt  ---->  broker.connectmq.com:8883
 (tags, monitored items)     map node -> ISA-95 topic   JSON payload       dashboards, historian, apps

El gateway tiene cuatro responsabilidades:

Suscribirse, no sondear. Las suscripciones OPC UA permiten que el servidor envíe los cambios de valor con el intervalo de muestreo que elijas, lo que es mucho más eficiente que leer todos los tags con un temporizador.
Mapear nodos a topics. Los Node IDs como ns=2;s=Line1.CNC1.SpindleTemp no significan nada para un desarrollador cloud. El gateway los traduce a una jerarquía legible.
Normalizar los payloads. Cada mensaje lleva el valor, el estado OPC UA, una marca de tiempo de origen y una unidad de ingeniería.
Sobrevivir a las desconexiones. Ambos lados se caerán en algún momento; el gateway debe reconectarse sin perder su configuración.

Requisitos previos

Python 3.10 o posterior.
asyncua (la librería cliente de código abierto FreeOpcUa) y paho-mqtt 2.x.
Un servidor OPC UA. Si no tienes un PLC a mano, el tutorial incluye un pequeño servidor simulado.
Un broker MQTT. Para pruebas y prototipos, usa el servidor MQTT público gratuito en broker.connectmq.com; para datos reales de planta, usa credenciales privadas de un broker MQTT gestionado.

python -m venv .venv
source .venv/bin/activate        # en Windows: .venv\Scripts\activate
pip install asyncua "paho-mqtt>=2.0"

Paso 1: un servidor OPC UA simulado (opcional)

Si ya tienes un servidor OPC UA, sáltate este paso y anota su URL de endpoint y sus Node IDs. Si no, este script expone tres tags cambiantes en opc.tcp://localhost:4840.

# sim_server.py - un pequeño servidor OPC UA con valores de CNC cambiantes
import asyncio
import math
import random
from asyncua import Server, ua

async def main():
    server = Server()
    await server.init()
    server.set_endpoint("opc.tcp://0.0.0.0:4840/sim/")
    idx = await server.register_namespace("urn:demo:cnc")

    cnc = await server.nodes.objects.add_object(ua.NodeId("Line1.CNC1", idx), "CNC1")
    temp = await cnc.add_variable(ua.NodeId("Line1.CNC1.SpindleTemp", idx), "SpindleTemp", 40.0)
    speed = await cnc.add_variable(ua.NodeId("Line1.CNC1.SpindleSpeed", idx), "SpindleSpeed", 0.0)
    count = await cnc.add_variable(ua.NodeId("Line1.CNC1.PartCount", idx), "PartCount", 0)

    async with server:
        t = 0
        while True:
            t += 1
            await temp.write_value(round(45 + 5 * math.sin(t / 20) + random.random(), 2))
            await speed.write_value(float(random.choice([0, 8000, 12000])))
            if t % 10 == 0:
                await count.write_value(await count.read_value() + 1)
            await asyncio.sleep(1)

if __name__ == "__main__":
    asyncio.run(main())

Como los Node IDs usan identificadores de tipo cadena en el namespace registrado (aquí, índice 2), aparecerán como ns=2;s=Line1.CNC1.SpindleTemp, etc. En un PLC real, explora el espacio de direcciones con una herramienta cliente como UaExpert para encontrar los Node IDs exactos.

Paso 2: el mapa de tags

Mantén el mapeo en la configuración, no en el código. Cada entrada empareja un nodo OPC UA con la ruta de topic y la unidad que quieres que vean los consumidores. La disposición de los topics sigue una jerarquía de estilo ISA-95 (sitio, área, línea, célula), que nuestra guía sobre topics MQTT ISA-95 explica en profundidad.

{
  "opcua_endpoint": "opc.tcp://localhost:4840/sim/",
  "topic_root": "connectmq/demo/a91c/plant1",
  "tags": [
    { "node": "ns=2;s=Line1.CNC1.SpindleTemp",  "path": "machining/line1/cnc1/spindle_temp",  "unit": "degC" },
    { "node": "ns=2;s=Line1.CNC1.SpindleSpeed", "path": "machining/line1/cnc1/spindle_speed", "unit": "rpm" },
    { "node": "ns=2;s=Line1.CNC1.PartCount",    "path": "machining/line1/cnc1/part_count",    "unit": "count" }
  ]
}

Guárdalo como tags.json. Sustituye a91c por tu propia cadena aleatoria para que tus topics no choquen con los de nadie en el broker público. Los topics resultantes tienen ocho niveles de profundidad, el máximo que permite el broker público; en un broker privado tienes más margen.

Paso 3: el código del gateway OPC UA a MQTT

# gateway.py - suscripción OPC UA -> publicación MQTT
import asyncio
import json
import logging
import ssl
import uuid
from datetime import timezone

import paho.mqtt.client as mqtt
from asyncua import Client

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("gateway")

CONFIG = json.load(open("tags.json"))
MQTT_HOST = "broker.connectmq.com"
MQTT_PORT = 8883
MQTT_USER = None          # producción: "YOUR_USERNAME"
MQTT_PASS = None          # producción: "YOUR_PASSWORD"
CLIENT_ID = f"opcua-gw-{uuid.uuid4().hex[:8]}"   # producción: "YOUR_CLIENT_ID"
STATUS_TOPIC = f"{CONFIG['topic_root']}/gateway/status"


def build_mqtt() -> mqtt.Client:
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                         protocol=mqtt.MQTTv5)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
    client.will_set(STATUS_TOPIC, json.dumps({"online": False}), qos=1, retain=True)

    def on_connect(client, userdata, flags, reason_code, properties):
        log.info("MQTT connected: %s", reason_code)
        client.publish(STATUS_TOPIC, json.dumps({"online": True}), qos=1, retain=True)

    def on_disconnect(client, userdata, flags, reason_code, properties):
        log.warning("MQTT disconnected: %s", reason_code)

    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    client.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
    client.loop_start()
    return client


class DataChangeHandler:
    """asyncua llama a datachange_notification en cada cambio de un monitored item."""

    def __init__(self, mqtt_client, node_map):
        self.mqtt = mqtt_client
        self.node_map = node_map

    def datachange_notification(self, node, val, data):
        tag = self.node_map[node.nodeid.to_string()]
        dv = data.monitored_item.Value
        ts = dv.SourceTimestamp or dv.ServerTimestamp
        payload = {
            "value": val,
            "unit": tag["unit"],
            "quality": dv.StatusCode.name,
            "ts": ts.replace(tzinfo=timezone.utc).isoformat() if ts else None,
        }
        topic = f"{CONFIG['topic_root']}/{tag['path']}"
        self.mqtt.publish(topic, json.dumps(payload), qos=1, retain=True)
        log.info("%s -> %s", topic, payload["value"])


async def run_opcua(mqtt_client):
    node_map = {t["node"]: t for t in CONFIG["tags"]}
    while True:
        try:
            async with Client(url=CONFIG["opcua_endpoint"]) as opc:
                log.info("OPC UA connected to %s", CONFIG["opcua_endpoint"])
                handler = DataChangeHandler(mqtt_client, node_map)
                sub = await opc.create_subscription(500, handler)  # intervalo de publicación en ms
                nodes = [opc.get_node(n) for n in node_map]
                await sub.subscribe_data_change(nodes)
                while True:                       # mantener viva la sesión, detectar fallos
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:                  # reinicio del servidor, pérdida de red, etc.
            log.error("OPC UA error: %s - retrying in 5 s", exc)
            await asyncio.sleep(5)


if __name__ == "__main__":
    asyncio.run(run_opcua(build_mqtt()))

Vale la pena destacar algunas decisiones de diseño:

Los retained messages hacen que un dashboard que se conecta más tarde vea al instante el último valor de cada tag.
El Last Will del topic de estado cambia a {"online": false} si el gateway muere sin una desconexión limpia, para que los consumidores puedan distinguir los datos obsoletos de los datos en vivo.
El código de estado OPC UA viaja con cada valor. Una lectura Good y una lectura BadCommunicationError nunca deberían parecer iguales aguas abajo.
check_connection() lanza una excepción cuando se pierde la sesión, lo que nos lleva al bucle de reintento y reconstruye la suscripción.

Paso 4: ejecutar y verificar

Arranca el servidor simulado y después el gateway, en dos terminales:

python sim_server.py
python gateway.py

En un tercer terminal, suscríbete a todo lo que cuelga de tu raíz:

mosquitto_sub -h broker.connectmq.com -p 1883 -i opcua-viewer-a91c \
  -t "connectmq/demo/a91c/plant1/#" -v

Deberías ver líneas como estas:

connectmq/demo/a91c/plant1/machining/line1/cnc1/spindle_temp {"value": 47.31, "unit": "degC", "quality": "Good", "ts": "2026-10-16T09:12:04.118000+00:00"}
connectmq/demo/a91c/plant1/gateway/status {"online": true}

Paso a producción

El servidor público gratuito lo puede leer cualquiera, así que solo sirve para pruebas y datos de muestra, nunca para datos de producción. Para datos reales de PLC, crea un broker privado y configura MQTT_USER, MQTT_PASS y CLIENT_ID con las credenciales del portal. ConnectMQ vincula los client IDs a las credenciales emitidas, así que usa el client ID generado y no uno aleatorio.

Lista de comprobación antes de poner el gateway en una red de planta:

Elemento	Por qué importa
Modo de seguridad OPC UA `SignAndEncrypt` con certificado de cliente	Las sesiones anónimas y sin cifrar valen para un laboratorio, no para un PLC
Usuario OPC UA dedicado de solo lectura	El gateway nunca debería poder escribir consignas
Intervalo de muestreo por clase de tag	Rápido para vibración, lento para niveles de depósito; evita saturar
Banda muerta en tags analógicos	Reduce los mensajes provocados por ruido, consulta el informe por excepción
Ejecutar como servicio con reinicio	systemd o un servicio de Windows, con rotación de logs
Búfer durante caídas del broker	paho encola mientras está desconectado, pero para cortes largos considera store and forward
Regla de firewall solo saliente al puerto 8883	No se abre ningún puerto entrante hacia la red OT

En ConnectMQ, el explorador de topics en vivo del portal muestra tu árbol ISA-95 a medida que se llena, y la vista de dispositivos agrupa las máquinas por niveles ISA-95, una forma rápida de confirmar que tu mapeo de topics es coherente entre líneas. Las reglas de alerta por umbral pueden avisar, por ejemplo, de una temperatura del husillo por encima de un límite sin escribir código adicional. Para ver dónde encaja este gateway en el panorama general, consulta nuestra guía de IoT industrial con MQTT.

Resolución de problemas

BadIdentityTokenRejected o BadUserAccessDenied: el servidor no permite el acceso anónimo. Pasa un usuario con Client.set_user() y set_password(), o configura certificados.
BadNodeIdUnknown: la cadena del Node ID es incorrecta, a menudo el índice del namespace. Los índices de namespace pueden cambiar entre versiones del servidor; resuélvelos por URI de namespace con get_namespace_index() al arrancar.
Llegan valores pero las marcas de tiempo son nulas: algunos servidores no establecen una marca de tiempo de origen. El código recurre a la marca de tiempo del servidor.
MQTT se conecta y se desconecta inmediatamente: en el broker público, comprueba que tu client ID sea único y no esté vacío. En un broker privado, comprueba que el client ID coincida con las credenciales emitidas.
Topics rechazados en el broker público: has superado los ocho niveles de topic. Acorta la raíz o la ruta.

Ejecuta tu gateway OPC UA en ConnectMQ

Crea un broker gratuito en app.connectmq.com con el plan Serverless, sin tarjeta de crédito. Copia en gateway.py el nombre de usuario, la contraseña y el client ID generados y, en pocos minutos, los tags de tu PLC aparecerán en un broker MQTT cloud privado y protegido con TLS, listo para dashboards e historiadores.

Preguntas frecuentes

¿Necesito un producto comercial para un gateway OPC UA a MQTT?

No. Las librerías de código abierto como asyncua y paho-mqtt bastan para muchos proyectos. Los gateways comerciales añaden interfaces gráficas, store and forward y soporte del fabricante, lo que puede importar a escala.

¿El gateway debe sondear los tags OPC UA o usar suscripciones?

Usa suscripciones. El servidor muestrea los valores y envía solo los cambios, lo que reduce la carga sobre el PLC y la red frente a las lecturas cíclicas.

¿Puede el gateway escribir valores de vuelta en el PLC?

Técnicamente sí, suscribiéndose a un topic de comandos y llamando a write_value, pero trátalo como una función separada y cuidadosamente protegida. La mayoría de los despliegues mantienen el gateway en solo lectura.

¿Qué QoS de MQTT debo usar para los datos de tags de PLC?

QoS 1 es la opción habitual: la entrega está garantizada al menos una vez y la sobrecarga es baja. Combínalo con retained messages para que los nuevos consumidores reciban el estado actual al instante.

Gateway OPC UA a MQTT: cómo llevar los datos del PLC al cloud