Puente OPC UA a MQTT para datos industriales

Un puente OPC UA a MQTT se conecta a un servidor OPC UA como cliente, se suscribe a los tags que te interesan y republica cada cambio de datos como mensaje MQTT en un topic bien nombrado. La planta mantiene OPC UA donde es fuerte —modelos ricos, seguridad, integración de máquinas— mientras dashboards, historiadores y servicios cloud reciben un flujo publish/subscribe ligero que pueden escalar libremente.

Este tutorial construye ese puente en Python con asyncua y paho-mqtt, mapea los node ids de OPC UA a un árbol de topics de Unified Namespace y muestra cómo probarlo con un servidor OPC UA simulado.

Por qué un puente en lugar de conexión directa

OPC UA es cliente/servidor: cada consumidor abre una sesión con cada servidor que necesita. Con cinco consumidores y veinte servidores son cien sesiones, cada una con sus suscripciones, certificados y carga sobre el servidor. Un puente lo convierte en un hub:

Por defecto el puente es de sentido único: OPC UA entra, MQTT sale. Así el lado OT queda protegido mientras IT recibe los datos. Para una comparación más amplia de ambos protocolos, lee MQTT vs OPC UA.

1. Arranca un servidor OPC UA de prueba

No desarrolles contra un servidor de producción. El simulador open source OPC PLC de Microsoft se ejecuta en Docker y expone valores cambiantes:

docker run --rm -p 50000:50000 mcr.microsoft.com/iotedge/opc-plc:latest \
  --pn=50000 --autoaccept --unsecuretransport

El endpoint es opc.tcp://localhost:50000. Explóralo con un cliente gráfico como UaExpert, o con el comando uals que incluye asyncua, y anota los node ids que quieres puentear. El simulador expone nodos como StepUp, AlternatingBoolean y RandomSignedInt32 en su propio namespace; comprueba el índice de namespace en tu navegador, porque puede variar entre versiones.

2. Mapea nodos a topics

El mapeo es el verdadero trabajo de diseño. Un topic debe decir dónde vive el valor en la planta, no qué servidor o node id lo proporciona hoy. Guarda el mapa en configuración:

BASE = "testmqtt/acme/plant1/packaging/line1"

NODES = {
    "ns=3;s=StepUp":             f"{BASE}/filler/counter",
    "ns=3;s=AlternatingBoolean": f"{BASE}/filler/running",
    "ns=3;s=RandomSignedInt32":  f"{BASE}/capper/torque",
}

Cuando más adelante sustituyas el servidor simulado por el real —o muevas un tag a otro PLC— solo cambia el lado izquierdo del mapa. Todos los suscriptores siguen funcionando. Es la idea central de un Unified Namespace: el árbol de topics es el contrato; las fuentes son detalles de implementación.

3. El puente

La API de suscripciones de asyncua llama a un handler cada vez que cambia un monitored item. El handler publica el valor, su calidad OPC UA y el timestamp de origen:

import asyncio, json, os
import paho.mqtt.client as mqtt
from asyncua import Client
from nodes import BASE, NODES

OPC_URL = os.getenv("OPC_URL", "opc.tcp://localhost:50000")
STATUS = f"{BASE}/_bridge/status"

mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="opcua-bridge-line1")
mq.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
mq.tls_set()
mq.will_set(STATUS, "offline", qos=1, retain=True)
mq.connect("broker.connectmq.com", 8883, keepalive=30)
mq.loop_start()

class Handler:
    def datachange_notification(self, node, val, data):
        dv = data.monitored_item.Value
        ts = dv.SourceTimestamp or dv.ServerTimestamp
        payload = {
            "v": val,
            "good": dv.StatusCode.is_good(),
            "ts": ts.isoformat() if ts else None,
        }
        topic = NODES[node.nodeid.to_string()]
        mq.publish(topic, json.dumps(payload, default=str), qos=1, retain=True)

async def run():
    while True:
        try:
            async with Client(url=OPC_URL, timeout=10) as opc:
                nodes = [opc.get_node(nid) for nid in NODES]
                sub = await opc.create_subscription(500, Handler())   # intervalo de publicación de 500 ms
                await sub.subscribe_data_change(nodes)
                mq.publish(STATUS, "online", qos=1, retain=True)
                print("puenteando", len(nodes), "nodos")
                while True:
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:          # conexión perdida, reinicio del servidor, node id erróneo
            print("Error OPC UA:", exc)
            mq.publish(STATUS, "opcua-unreachable", qos=1, retain=True)
            await asyncio.sleep(5)

asyncio.run(run())

pip install "asyncua>=1.1" "paho-mqtt>=2.0"
python opcua_mqtt_bridge.py

Lo que te da este diseño:

• Tráfico guiado por cambios. OPC UA envía notificaciones solo cuando cambian los valores, así que el lado MQTT no lleva muestras redundantes. Ajusta el intervalo de publicación (aquí 500 ms) y el intervalo de muestreo de cada item a tu proceso.
• La calidad viaja con el valor. Un status code OPC UA malo o incierto es información; mantenlo en el payload en lugar de descartar el mensaje en silencio.
• Timestamps de origen. Los historiadores deben guardar cuándo cambió el valor en origen, no cuándo lo reenvió el puente.
• Valores retenidos y topic de estado. Los nuevos suscriptores ven el estado actual al instante, y el Last Will pone el estado del puente en offline si el proceso muere. Más en mensajes retenidos.
• Recuperación automática. Si el servidor OPC UA se reinicia, el bucle exterior se reconecta y recrea la suscripción. paho se reconecta solo al broker tras loop_start().

4. Prueba el puente

1. Arranca el simulador y el puente. El puente debería imprimir el número de nodos puenteados.
2. Abre el cliente MQTT online, conéctate a broker.connectmq.com y suscríbete a testmqtt/acme/plant1/#. filler/counter debería ir subiendo, filler/running alternar y _bridge/status mostrar online.
3. Detén el contenedor del simulador. En pocos segundos el estado debería pasar a opcua-unreachable; arráncalo de nuevo y los valores deberían volver sin reiniciar el puente.
4. Mata el puente con kill -9. Tras el timeout del keep-alive, el broker publica el Last Will: offline.

Observa los tags OPC UA puenteados en tu navegador

Comprueba con el topic matcher que los filtros que usan los consumidores coinciden con lo que publica el puente, y ten en cuenta los valores estructurados grandes (arrays, extension objects): la calculadora de tamaño de paquete ayuda a estimar lo que cuesta una notificación en la red.

Otras formas de hacer el puente

Un pequeño puente en Python es ideal para aprender, para un puñado de servidores y para tener control total del mapeo. En instalaciones grandes, considera:

Elijas la herramienta que elijas, las reglas son las mismas: los topics describen la planta, los payloads llevan valor + calidad + timestamp, el puente anuncia su propia salud y las escrituras de vuelta a OPC UA son un camino separado y protegido deliberadamente.

Checklist de producción

• Usa la seguridad de OPC UA (Basic256Sha256 con firma y cifrado) y un certificado de aplicación dedicado para el puente en servidores reales; la opción sin seguridad del simulador es solo para pruebas locales.
• Consulta los límites del servidor de monitored items por suscripción y reparte listas grandes de tags en varias suscripciones.
• Ejecuta un puente por área o línea para que un fallo quede localizado.
• Versiona el mapa nodo→topic y revísalo como código.
• Monitoriza el topic de estado del puente y alerta cuando no esté en online.

Un puente OPC UA a MQTT te permite mantener OPC UA para lo que mejor hace y dar a todos los demás sistemas un flujo sencillo y escalable. Mapea nodos a topics con forma de planta, publica las notificaciones de cambio con calidad y timestamps, prueba contra un simulador y el cliente online, y tus datos OT se convertirán en algo a lo que toda la empresa puede suscribirse.

Preguntas frecuentes