Allen-Bradley MQTT: publicar tags de ControlLogix y CompactLogix

La integración Allen-Bradley MQTT es la forma más rápida de llevar datos de tags de ControlLogix y CompactLogix a dashboards, historiadores y analítica en la nube sin añadir carga a tu servidor SCADA. En este tutorial construirás un pequeño gateway en Python que lee tags Logix por EtherNet/IP con pycomm3, los asigna a topics ISA-95 y publica en un broker MQTT solo los cambios significativos. Primero probarás con el servidor MQTT público gratuito de broker.connectmq.com, pensado solo para pruebas y prototipos, y después pasarás a uno privado con TLS.

Cómo encajan los PLC Allen-Bradley con MQTT

Los controladores Logix se comunican de forma nativa por EtherNet/IP mediante CIP (Common Industrial Protocol). Los proyectos de Studio 5000 organizan los datos como tags con nombre, de ámbito de controlador o de programa, con tipos atómicos (BOOL, DINT, REAL), arrays y tipos definidos por el usuario (UDT). Ese modelo de tags con nombre es un regalo para la integración con MQTT: a diferencia de Modbus, no necesitas un mapa de registros, porque el nombre del tag ya transmite su significado.

Hay tres formas habituales de llevar esos tags a MQTT:

Este tutorial sigue el primer enfoque porque muestra todas las piezas. El mismo diseño de topics y payloads sirve si más adelante pasas a un driver comercial o a Ignition con MQTT.

Requisitos previos

• Un controlador ControlLogix o CompactLogix accesible desde el gateway por el puerto TCP 44818.
• Una máquina Linux o Windows con Python 3.9 o posterior.
• Algunos tags que leer, por ejemplo Line1_Temp (REAL), Line1_Speed (DINT), Line1_Running (BOOL) y un tag de ámbito de programa Program:MainProgram.Pump1_Pressure (REAL).
• Los clientes de mosquitto para las pruebas.

Instala las librerías:

python -m venv abmqtt
source abmqtt/bin/activate
pip install pycomm3 "paho-mqtt>=2.0"

Paso 1: descubrir la lista de tags

Antes de escribir el gateway, confirma la conectividad y comprueba exactamente cómo se llaman los tags. pycomm3 descarga la base de datos de tags del controlador al conectarse:

from pycomm3 import LogixDriver

# CompactLogix: just the IP. ControlLogix with the CPU in slot 3: "192.168.1.20/3"
PLC_PATH = "192.168.1.20"

with LogixDriver(PLC_PATH) as plc:
    print(plc.info)                       # controller name, revision, serial
    for name, definition in plc.tags.items():
        print(name, definition["data_type_name"], definition.get("dim", 0))

Los tags de ámbito de programa no aparecen por defecto en la lista de ámbito de controlador; direcciónalos como Program:MainProgram.TagName al leerlos. En chasis ControlLogix, la parte de la ruta tras la barra es el slot del controlador en el backplane. Si te equivocas, la conexión se abre con el módulo Ethernet pero las lecturas de tags fallan.

La documentación de pycomm3 explica en detalle la sintaxis de rutas, los arrays y el manejo de UDT.

Paso 2: diseñar topics ISA-95

Asigna a cada tag un topic que describa el equipo, no el controlador. Una sencilla tabla de mapeo en el gateway mantiene las convenciones de nombres del PLC fuera de tu namespace MQTT:

Con un prefijo de planta y área, un topic completo queda como plant1/assembly/line1/oven/temperature. Rutas coherentes como esta son las que hacen funcionar un namespace a nivel de planta; consulta topics MQTT ISA-95 para profundizar en niveles y nomenclatura.

Paso 3: el gateway Allen-Bradley MQTT

Cada segundo, el gateway lee todos los tags mapeados con una única petición multitag, aplica bandas muertas por tag, publica los cambios con QoS 1, refresca todo cada cinco minutos y mantiene un estado online/offline retained con un Last Will.

import json
import time
import uuid
from datetime import datetime, timezone

from pycomm3 import LogixDriver
import paho.mqtt.client as mqtt

PLC_PATH = "192.168.1.20"

# Public test broker first. The public area is readable by everyone: test data only.
BROKER, PORT, USE_TLS = "broker.connectmq.com", 1883, False
USERNAME, PASSWORD = None, None                  # production: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"ab-gw-{uuid.uuid4().hex[:8]}"      # production: "YOUR_CLIENT_ID"
BASE = f"connectmq/demo/{uuid.uuid4().hex[:6]}/plant1/assembly/line1"

TAGS = {
    "Line1_Temp": ("oven/temperature", "degC", 0.5),
    "Line1_Speed": ("conveyor/speed", "rpm", 5),
    "Line1_Running": ("line/running", None, 0),
    "Program:MainProgram.Pump1_Pressure": ("pump1/pressure", "bar", 0.05),
}


def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT connected:", reason_code)
    client.publish(f"{BASE}/gateway/status", "online", qos=1, retain=True)


def should_publish(tag, value, last, deadband):
    if tag not in last:
        return True
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return value != last[tag]
    return abs(value - last[tag]) >= deadband


def main():
    mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        mq.username_pw_set(USERNAME, PASSWORD)
    if USE_TLS:
        mq.tls_set()
    mq.will_set(f"{BASE}/gateway/status", "offline", qos=1, retain=True)
    mq.on_connect = on_connect
    mq.connect(BROKER, PORT, keepalive=30)
    mq.loop_start()
    print("Publishing under", BASE)

    plc = LogixDriver(PLC_PATH)
    last, last_full = {}, 0.0

    while True:
        try:
            if not plc.connected:
                plc.open()
            results = plc.read(*TAGS.keys())
            now = time.time()
            force = now - last_full > 300
            ts = datetime.now(timezone.utc).isoformat(timespec="seconds")

            for result in results:
                suffix, unit, deadband = TAGS[result.tag]
                if result.error:
                    mq.publish(f"{BASE}/{suffix}", json.dumps(
                        {"value": None, "quality": "bad", "error": result.error, "ts": ts}), qos=1)
                    continue
                value = result.value
                if isinstance(value, float):
                    value = round(value, 3)
                if force or should_publish(result.tag, value, last, deadband):
                    payload = {"value": value, "quality": "good", "ts": ts}
                    if unit:
                        payload["unit"] = unit
                    mq.publish(f"{BASE}/{suffix}", json.dumps(payload), qos=1)
                    last[result.tag] = value
            if force:
                last_full = now
        except Exception as exc:
            print("PLC communication failed:", exc)
            try:
                plc.close()
            except Exception:
                pass
            time.sleep(5)
        time.sleep(1)


if __name__ == "__main__":
    main()

Algunos aspectos de diseño que conviene destacar:

• Una sola lectura multitag. pycomm3 agrupa varios tags en el menor número posible de peticiones CIP, lo que resulta mucho más amable con el controlador que leer cada tag por separado.
• Errores por tag. Una errata en el nombre de un tag devuelve un error solo para ese tag. Publicar quality: bad hace visible el problema en lugar de congelar un valor en silencio.
• Los UDT también funcionan. Leer un UDT devuelve un diccionario de Python, que json.dumps convierte en un payload estructurado. Mantén pequeños los payloads de UDT y evita publicar arrays grandes cada segundo.

Paso 4: verificar en el servidor público

Arranca el gateway y suscríbete a su prefijo:

mosquitto_sub -h broker.connectmq.com -p 1883 -i ab-watch-01 \
  -t 'connectmq/demo/+/plant1/assembly/line1/#' -v

Deberías ver de inmediato el estado retained del gateway y el primer conjunto completo de valores. Cambia Line1_Running desde Studio 5000 y el cambio llegará en menos de un segundo. Usar el servidor MQTT público gratuito en este paso no supone problema siempre que los valores no sean sensibles ni de producción.

Paso 5: producción en un broker privado

Para producción, crea un broker privado en el portal de ConnectMQ, un broker MQTT gestionado, y cambia cuatro líneas: puerto 8883, USE_TLS = True, el usuario y la contraseña emitidos y el client ID emitido. Elimina el prefijo connectmq/demo/... para que los topics empiecen en el nivel de empresa o de planta. Como los client ID están vinculados a sus credenciales, da a cada gateway su propio juego en lugar de compartir uno entre plantas, por ejemplo entre la de Monterrey y la de Querétaro.

Ejecuta el gateway como servicio para que se reinicie automáticamente:

[Unit]
Description=Allen-Bradley MQTT gateway
After=network-online.target

[Service]
ExecStart=/opt/abmqtt/bin/python /opt/abmqtt/gateway.py
Restart=always
RestartSec=5
User=abmqtt

[Install]
WantedBy=multi-user.target

Cuando los datos empiezan a fluir, la vista de dispositivos del portal de ConnectMQ agrupa los topics ISA-95 por planta, área y línea, y las reglas de alerta por umbral pueden avisar de una temperatura o presión fuera de sus límites. Historiadores, dashboards y un gemelo digital 3D de la línea se suscriben como cualquier otro cliente del broker MQTT.

Ajustar frecuencias de lectura y bandas muertas

El intervalo de lectura y las bandas muertas deciden juntos cuánto tráfico genera el gateway, tanto en la red de control como en el broker. Una forma práctica de fijarlos:

• Parte del consumidor. Un dashboard que un operador consulta cada pocos segundos no necesita datos cada 100 ms. Un historiador usado para informes energéticos puede conformarse con medias de un minuto.
• Dimensiona las bandas muertas según el ruido del sensor. Observa un valor durante unos minutos en la vista de tendencias de Studio 5000. Una banda muerta algo mayor que la banda de ruido elimina las oscilaciones sin ocultar cambios reales.
• Nunca apliques banda muerta a estados y alarmas. Booleanos, palabras de modo y bits de alarma deben publicarse en cada cambio, y sus topics son buenos candidatos para mensajes retained, de modo que un nuevo suscriptor vea el estado actual al instante.
• Mantén el refresco periódico. Sin él, un valor que realmente permanece constante durante horas no se distingue de un gateway que se ha detenido en silencio.

Si más adelante un solo gateway se te queda pequeño, ejecuta uno por área o por línea en lugar de un único proceso gigante para toda la planta. Los fallos quedan acotados y cada instancia tiene una lista de tags pequeña y fácil de revisar.

Consideraciones de seguridad

CIP sobre EtherNet/IP no tiene autenticación en la mayoría de instalaciones, así que el gateway es, en la práctica, un cliente de ingeniería de confianza. Colócalo en la red de control detrás de un firewall, permite desde él solo TLS saliente hacia el broker y nunca dejes que comandos MQTT escriban en tags del PLC sin una lista de permitidos explícita, comprobaciones de rango y un enclavamiento en la lógica del PLC. Automatizar la lectura es seguro; la escritura merece una revisión de diseño. La guía de EtherNet/IP a MQTT trata con más detalle la segmentación de red para el tráfico CIP, y la guía de IoT industrial con MQTT explica dónde se sitúan los gateways en una arquitectura completa.

Solución de problemas

1. La conexión agota el tiempo de espera. Revisa el enrutamiento y que el puerto TCP 44818 esté abierto entre el gateway y el controlador.
2. Conecta pero fallan todos los tags. Slot incorrecto en una ruta ControlLogix, o el tag es de ámbito de programa y necesita el prefijo Program:.
3. Valores REAL con muchos decimales. Es la representación normal de los float; redondea antes de publicar.
4. MQTT se desconecta cada pocos segundos. Otro cliente está usando el mismo client ID.
5. Carga de comunicación alta en el controlador. Aumenta el intervalo de lectura o reduce la lista de tags; la mayoría de dashboards no necesitan datos por debajo del segundo.

Conecta tus controladores Logix a ConnectMQ

Tu gateway está listo; dale un broker MQTT en la nube seguro en el que publicar. Regístrate en app.connectmq.com para obtener un plan Serverless gratuito sin tarjeta de crédito, pega en el script las credenciales generadas y observa cómo aparecen en directo tus tags Allen-Bradley.

Preguntas frecuentes

¿Puede un PLC ControlLogix publicar MQTT de forma nativa?

Los controladores Logix no incluyen en el propio controlador un cliente MQTT de uso general. Lo habitual es un gateway de borde, un servidor de conectividad o un módulo SCADA que lee los tags por EtherNet/IP y los publica.

¿Funciona pycomm3 con Micro800 o SLC 500?

El LogixDriver de pycomm3 está orientado a ControlLogix, CompactLogix y Micro800 con algunas limitaciones; los antiguos PLC-5 y SLC 500 usan un driver aparte. Consulta la documentación de la librería para ver las funciones admitidas por familia.

¿Con qué frecuencia debo leer los tags?

Un segundo es más que suficiente para la mayoría de la monitorización. Usa bandas muertas para que solo se publiquen cambios reales, junto con un refresco completo periódico.

¿Debería usar Sparkplug B en lugar de JSON?

Sparkplug B añade certificados de nacimiento y defunción (birth y death) y una codificación Protobuf compacta, lo que ayuda con grandes cantidades de tags y con la integración SCADA. JSON sin más es más sencillo para empezar y fácil de inspeccionar.