Allen-Bradley MQTT: Publicando Tags de ControlLogix e CompactLogix

A integração Allen-Bradley MQTT é a forma mais rápida de levar dados de tags de ControlLogix e CompactLogix a dashboards, historiadores e análises na nuvem sem sobrecarregar o seu servidor SCADA. Neste tutorial você vai construir um pequeno gateway Python que lê tags Logix via EtherNet/IP com pycomm3, mapeia-as para topics ISA-95 e publica apenas as mudanças relevantes em um broker MQTT. Primeiro você testará no servidor MQTT público e depois passará para um broker MQTT gerenciado privado com TLS.

Como os CLPs Allen-Bradley se encaixam no MQTT

Os controladores Logix se comunicam nativamente via EtherNet/IP usando CIP (Common Industrial Protocol). Os projetos do Studio 5000 organizam os dados como tags nomeadas, de escopo do controlador ou de escopo do programa, com tipos atômicos (BOOL, DINT, REAL), arrays e tipos definidos pelo usuário (UDTs). Esse modelo de tags nomeadas é um presente para a integração MQTT: ao contrário do Modbus, você não precisa de um mapa de registradores, porque o nome da tag já carrega o significado.

Há três formas comuns de colocar essas tags no MQTT:

Este tutorial segue a primeira abordagem porque mostra cada peça em funcionamento. O mesmo design de topics e payloads vale se você migrar depois para um driver comercial ou para o Ignition com MQTT.

Pré-requisitos

• Um controlador ControlLogix ou CompactLogix acessível a partir do gateway na porta TCP 44818.
• Uma máquina Linux ou Windows com Python 3.9 ou mais recente.
• Algumas tags para ler, por exemplo Line1_Temp (REAL), Line1_Speed (DINT), Line1_Running (BOOL) e uma tag de escopo de programa Program:MainProgram.Pump1_Pressure (REAL).
• Os clientes mosquitto para testes.

Instale as bibliotecas:

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

Passo 1: descubra a lista de tags

Antes de escrever o gateway, confirme a conectividade e veja exatamente como as tags são nomeadas. O pycomm3 faz o upload do banco de tags do controlador ao conectar:

from pycomm3 import LogixDriver

# CompactLogix: apenas o IP. ControlLogix com a CPU no slot 3: "192.168.1.20/3"
PLC_PATH = "192.168.1.20"

with LogixDriver(PLC_PATH) as plc:
    print(plc.info)                       # nome do controlador, revisão, número de série
    for name, definition in plc.tags.items():
        print(name, definition["data_type_name"], definition.get("dim", 0))

Tags de escopo de programa não aparecem por padrão na lista de escopo do controlador; endereçe-as como Program:MainProgram.TagName na leitura. Em chassis ControlLogix, o caminho após a barra é o slot do controlador no backplane. Se errar, a conexão abre com o módulo Ethernet, mas as leituras de tags falham.

A documentação do pycomm3 cobre em detalhes a sintaxe de caminhos, arrays e o tratamento de UDTs.

Passo 2: projete topics ISA-95

Mapeie cada tag para um topic que descreva o equipamento, não o controlador. Uma tabela simples de mapeamento no gateway mantém as convenções de nomes do CLP fora do seu namespace MQTT:

Com um prefixo de site e área, um topic completo fica plant1/assembly/line1/oven/temperature. Caminhos consistentes como este são o que faz funcionar um namespace para toda a planta; veja topics MQTT ISA-95 para um tratamento mais aprofundado de níveis e nomes.

Passo 3: o gateway Allen-Bradley MQTT

O gateway lê todas as tags mapeadas em uma única requisição multi-tag a cada segundo, aplica bandas mortas por tag, publica as mudanças com QoS 1, atualiza tudo a cada cinco minutos e mantém um status online/offline retido com 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"

# Primeiro o broker público de testes. A área pública pode ser lida por qualquer pessoa: apenas dados de teste.
BROKER, PORT, USE_TLS = "broker.connectmq.com", 1883, False
USERNAME, PASSWORD = None, None                  # produção: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"ab-gw-{uuid.uuid4().hex[:8]}"      # produção: "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()

Alguns pontos de design que merecem destaque:

• Uma única leitura multi-tag. O pycomm3 agrupa várias tags no menor número possível de requisições CIP, o que é muito mais gentil com o controlador do que ler cada tag separadamente.
• Erros por tag. Um erro de digitação no nome de uma tag retorna erro apenas para ela. Publicar quality: bad torna o problema visível em vez de congelar um valor silenciosamente.
• UDTs também funcionam. Ler um UDT retorna um dict Python, que o json.dumps transforma em um payload estruturado. Mantenha os payloads de UDT pequenos e evite publicar arrays grandes a cada segundo.

Passo 4: verifique no servidor MQTT público

Inicie o gateway e assine o prefixo dele:

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

Você deve ver imediatamente o status retido do gateway e o primeiro conjunto completo de valores. Alterne Line1_Running pelo Studio 5000 e a mudança chega em menos de um segundo. Usar o servidor MQTT público (gratuito) nesta etapa não tem problema, desde que os valores não sejam sensíveis: ele serve para testes e protótipos, não para dados de produção.

Passo 5: produção em um broker privado

Para produção, crie um broker MQTT gerenciado privado no portal do ConnectMQ e altere quatro linhas: porta 8883, USE_TLS = True, o usuário e a senha emitidos e o client ID emitido. Remova o prefixo connectmq/demo/... para que os topics comecem no nível da empresa ou do site. Como os client IDs são vinculados às suas credenciais, dê a cada gateway seu próprio conjunto em vez de compartilhar um entre sites.

Execute o gateway como serviço para que reinicie automaticamente:

[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

Assim que os dados fluírem, a visualização de dispositivos do portal ConnectMQ agrupa os topics ISA-95 por site, área e linha, e regras de alerta por limite podem sinalizar uma temperatura ou pressão fora dos limites. Historiadores, dashboards e um gêmeo digital da linha assinam como qualquer outro cliente do broker MQTT.

Ajustando taxas de leitura e bandas mortas

O intervalo de leitura e as bandas mortas, juntos, determinam quanto tráfego o gateway gera, tanto na rede de controle quanto no broker. Uma forma prática de defini-los:

• Parta do consumidor. Um dashboard que o operador atualiza a cada poucos segundos não precisa de dados a cada 100 ms. Um historiador usado para relatórios de energia pode se contentar com médias de um minuto.
• Dimensione as bandas mortas pelo ruído do sensor. Observe um valor por alguns minutos na trend view do Studio 5000. Uma banda morta um pouco maior que a faixa de ruído elimina oscilações sem esconder mudanças reais.
• Nunca aplique banda morta a estados e alarmes. Booleanos, words de modo e bits de alarme devem ser publicados a cada mudança, e seus topics são bons candidatos a retained messages, para que um novo assinante veja o estado atual imediatamente.
• Mantenha a atualização periódica. Sem ela, um valor que realmente fica constante por horas parece idêntico a um gateway que parou silenciosamente.

Se um único gateway ficar pequeno demais, por exemplo em uma planta de autopeças em Joinville com dezenas de linhas, rode um por área ou linha em vez de um processo gigante para a planta inteira. As falhas ficam contidas e cada instância tem uma lista de tags pequena e fácil de revisar.

Considerações de segurança

O CIP sobre EtherNet/IP não tem autenticação na maioria das instalações, então o gateway é na prática um cliente de engenharia confiável. Coloque-o na rede de controle atrás de um firewall, permita apenas TLS de saída para o broker e nunca deixe comandos MQTT escreverem em tags do CLP sem uma lista explícita de permissões, verificação de faixas e um intertravamento na lógica do CLP. Ler é seguro de automatizar; escrever merece uma revisão de design. O guia de EtherNet/IP para MQTT aborda com mais profundidade a segmentação de rede para tráfego CIP, e o guia de IoT industrial com MQTT explica onde os gateways ficam em uma arquitetura completa.

Solução de problemas

1. A conexão expira. Verifique o roteamento e se a porta TCP 44818 está aberta entre o gateway e o controlador.
2. Conectado, mas todas as tags falham. Slot errado no caminho do ControlLogix, ou a tag é de escopo de programa e precisa do prefixo Program:.
3. Valores REAL com muitas casas decimais. É a representação normal de ponto flutuante; arredonde antes de publicar.
4. O MQTT desconecta a cada poucos segundos. Outro cliente está usando o mesmo client ID.
5. Carga de comunicação alta no controlador. Aumente o intervalo de leitura ou reduza a lista de tags; a maioria dos dashboards não precisa de dados abaixo de um segundo.

Conecte seus controladores Logix ao ConnectMQ

Seu gateway está pronto; dê a ele um broker MQTT na nuvem seguro para publicar. Cadastre-se em app.connectmq.com no plano gratuito Serverless sem cartão de crédito, cole as credenciais geradas no script e veja suas tags Allen-Bradley aparecerem ao vivo.

Perguntas frequentes

Um CLP ControlLogix pode publicar MQTT nativamente?

Os controladores Logix não incluem um cliente MQTT de uso geral no próprio controlador. A abordagem usual é um gateway de borda, servidor de conectividade ou módulo SCADA que lê as tags via EtherNet/IP e as publica.

O pycomm3 funciona com Micro800 ou SLC 500?

O LogixDriver do pycomm3 é voltado para ControlLogix, CompactLogix e Micro800, com algumas limitações; os antigos PLC-5 e SLC 500 usam um driver separado. Consulte a documentação da biblioteca para ver os recursos suportados por família.

Com que frequência devo ler as tags?

Um segundo é suficiente para a maior parte do monitoramento. Use bandas mortas para que apenas mudanças reais sejam publicadas, além de uma atualização completa periódica.

Devo usar Sparkplug B em vez de JSON?

O Sparkplug B adiciona certificados de nascimento e morte e codificação Protobuf compacta, o que ajuda com grandes quantidades de tags e integração com SCADA. JSON simples é mais fácil para começar e para inspecionar.