EtherNet/IP para MQTT: Lendo Dispositivos CIP e Tags Logix com Python

Uma ponte EtherNet/IP para MQTT permite tirar dados de controladores Rockwell Logix, inversores e outros dispositivos CIP e levá-los a dashboards, historiadores e aplicações na nuvem sem mexer no programa de controle, seja em uma linha de embalagens em Manaus ou em um frigorífico no Centro-Oeste. Neste tutorial você vai entender a diferença entre mensagens implícitas e explícitas e, em seguida, usar a biblioteca open source pycomm3 em um gateway para ler tags Logix e atributos CIP genéricos e publicá-los via MQTT com payloads limpos e autodescritivos.

EtherNet/IP e CIP em cinco minutos

EtherNet/IP é o Common Industrial Protocol (CIP) transportado sobre Ethernet padrão e TCP/UDP. O CIP modela cada dispositivo como um conjunto de objetos com classes, instâncias e atributos. Um inversor, por exemplo, tem um objeto Identity (classe 0x01) que descreve fabricante, código do produto e número de série, além de objetos de aplicação para velocidade, corrente e status.

O tráfego CIP tem duas variantes:

Um gateway de monitoramento deve usar mensagens explícitas. Elas não interferem nas conexões de I/O do controlador, não exigem alterações no projeto do CLP e o gateway pode ser adicionado ou removido sem parada. O custo é que se trata de um protocolo de polling, então você escolhe a frequência de leitura.

Dois tipos de alvo

1. Controladores Logix (ControlLogix, CompactLogix e, até certo ponto, Micro800). Eles suportam acesso simbólico a tags: você lê Line1_Speed ou Conveyor[3].Status pelo nome. É de longe a fonte de dados mais conveniente.
2. Dispositivos CIP genéricos (inversores, adapters de I/O, scanners, dispositivos de terceiros). Eles expõem atributos de objetos que você lê por número de classe, instância e atributo, conforme a documentação do dispositivo ou o arquivo EDS.

O pycomm3 atende aos dois casos: LogixDriver para controladores baseados em tags e CIPDriver para mensagens explícitas genéricas. O pylogix é outra opção open source popular para tags Logix. Siemens, Beckhoff e outras plataformas têm seus próprios caminhos de acesso; para temas específicos da Rockwell, como a configuração do controlador, veja nosso guia Allen-Bradley MQTT.

Pré-requisitos

• Python 3.10 ou mais recente em um gateway que alcance o controlador na porta TCP 44818.
• pycomm3 e paho-mqtt 2.x.
• O endereço IP do controlador e, para controladores em chassi, o número do slot (por exemplo 10.0.0.20/0 para o slot 0).
• Uma lista de tags somente leitura acordada com o engenheiro de automação.

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

Passo 1: explore o controlador

Antes de escrever a ponte, confirme a conectividade e liste as tags disponíveis. Este script exibe as informações do controlador e as primeiras tags de escopo do controlador.

# explore_logix.py - confirma a conectividade e lista as tags
from pycomm3 import LogixDriver

with LogixDriver("10.0.0.20/0") as plc:          # IP/slot do controlador
    print(plc.info)                               # fabricante, produto, revisão, nome
    tags = plc.get_tag_list()                     # tags de escopo do controlador
    for t in tags[:30]:
        print(t["tag_name"], t["data_type_name"])

get_tag_list() faz o upload das definições de tags, incluindo tipos definidos pelo usuário, por isso pode demorar um pouco em projetos grandes. Tags de escopo de programa são lidas com a sintaxe Program:MainProgram.TagName.

Passo 2: leia um atributo CIP genérico

Para um inversor ou qualquer dispositivo CIP sem tags simbólicas, use mensagens explícitas. O exemplo lê o nome do produto no objeto Identity, que todo dispositivo EtherNet/IP implementa.

# identity.py - lê o atributo 7 (Product Name) do objeto Identity de qualquer dispositivo CIP
from pycomm3 import CIPDriver, Services, ClassCode, SHORT_STRING

with CIPDriver("10.0.0.31") as dev:
    resp = dev.generic_message(
        service=Services.get_attribute_single,
        class_code=ClassCode.identity_object,   # 0x01
        instance=1,
        attribute=7,                             # Product Name
        data_type=SHORT_STRING,
        connected=False,
        name="product_name",
    )
    print(resp.value if not resp.error else resp.error)

Substitua classe, instância, atributo e tipo de dado pelos valores do manual do dispositivo para ler dados de aplicação, como frequência de saída ou corrente do motor.

Passo 3: a ponte EtherNet/IP para MQTT

A cada ciclo, a ponte lê uma lista de tags em uma única requisição multi-tag, publica os valores que mudaram (ou que ultrapassaram uma banda morta) e informa erros de leitura como qualidade ruim, em vez de manter silenciosamente valores antigos.

# eip_mqtt_bridge.py - faz polling de tags Logix via EtherNet/IP e publica no MQTT
import json, logging, time, uuid
import paho.mqtt.client as mqtt
from pycomm3 import LogixDriver, CommError

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

PLC_PATH = "10.0.0.20/0"
POLL_SECONDS = 1.0
TOPIC_ROOT = "connectmq/demo/9e6d/plant1/packaging/line2"   # seu próprio id aleatório
MQTT_HOST, MQTT_PORT = "broker.connectmq.com", 1883         # produção: 8883 com TLS
MQTT_USER, MQTT_PASS = None, None                           # produção: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"eip-bridge-{uuid.uuid4().hex[:8]}"            # produção: "YOUR_CLIENT_ID"

# tag do CLP -> (folha do topic, unidade, banda morta)
TAGS = {
    "Line2_Speed":          ("speed", "ppm", 1.0),
    "Line2_Running":        ("running", "", 0),
    "Line2_GoodCount":      ("good_count", "count", 0),
    "Line2_RejectCount":    ("reject_count", "count", 0),
    "Filler.MotorCurrent":  ("filler_current", "A", 0.2),
}


def mqtt_client():
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if MQTT_USER:
        c.username_pw_set(MQTT_USER, MQTT_PASS)
        c.tls_set()
    c.will_set(f"{TOPIC_ROOT}/bridge", "offline", qos=1, retain=True)
    c.on_connect = lambda cl, u, f, rc, p: cl.publish(f"{TOPIC_ROOT}/bridge", "online", qos=1, retain=True)
    c.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
    c.loop_start()
    return c


def changed(old, new, deadband):
    if old is None:
        return True
    if isinstance(new, (int, float)) and not isinstance(new, bool) and deadband:
        return abs(new - old) >= deadband
    return new != old


def main():
    mq = mqtt_client()
    last = {}
    while True:
        try:
            with LogixDriver(PLC_PATH) as plc:
                log.info("connected to %s", plc.info.get("product_name"))
                while True:
                    results = plc.read(*TAGS.keys())
                    if not isinstance(results, list):
                        results = [results]
                    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
                    for r in results:
                        leaf, unit, deadband = TAGS[r.tag]
                        topic = f"{TOPIC_ROOT}/{leaf}"
                        if r.error:
                            if last.get(r.tag, {}).get("quality") != "bad":
                                mq.publish(topic, json.dumps({"value": None, "quality": "bad",
                                                              "error": r.error, "ts": now}), qos=1, retain=True)
                                last[r.tag] = {"value": None, "quality": "bad"}
                            continue
                        prev = last.get(r.tag, {})
                        if prev.get("quality") == "bad" or changed(prev.get("value"), r.value, deadband):
                            payload = {"value": r.value, "unit": unit, "type": r.type,
                                       "quality": "good", "ts": now}
                            mq.publish(topic, json.dumps(payload), qos=1, retain=True)
                            last[r.tag] = {"value": r.value, "quality": "good"}
                    time.sleep(POLL_SECONDS)
        except CommError as exc:
            log.error("EtherNet/IP connection lost: %s - retrying in 5 s", exc)
            time.sleep(5)


if __name__ == "__main__":
    main()

plc.read() com vários nomes de tags agrupa-os no menor número de requisições que o tamanho da conexão permite, o que pesa muito menos no controlador do que uma requisição por tag. Cada resultado é um Tag com os campos tag, value, type e error, de modo que a ponte pode publicar o tipo de dado junto com o valor.

Passo 4: execute e verifique

python eip_mqtt_bridge.py
mosquitto_sub -h broker.connectmq.com -p 1883 -i eip-viewer-9e6d -t "connectmq/demo/9e6d/#" -v

connectmq/demo/9e6d/plant1/packaging/line2/speed {"value": 118.0, "unit": "ppm", "type": "REAL", "quality": "good", "ts": "2026-10-16T12:00:01Z"}
connectmq/demo/9e6d/plant1/packaging/line2/running {"value": true, "unit": "", "type": "BOOL", "quality": "good", "ts": "2026-10-16T12:00:01Z"}

A árvore de topics segue um layout no estilo ISA-95 (site, área, linha, ponto), então os mesmos consumidores funcionam quer os dados venham de EtherNet/IP, Modbus ou OPC UA. Nosso guia de topics MQTT ISA-95 explica as convenções.

Escolhendo tags e taxas de polling

Nem toda tag de um projeto Logix merece um topic. Parta das perguntas que os consumidores vão fazer: a linha está rodando, a que velocidade, quantas unidades boas e rejeitadas, quais alarmes estão ativos e quanta energia os acionamentos principais consomem. Isso geralmente significa algumas dezenas de valores por linha, não milhares. Agrupe-os pela velocidade com que mudam. Estados e contadores podem ser lidos a cada segundo e publicados quando mudam; valores analógicos lentos, como níveis de tanque ou temperaturas, podem ser lidos a cada cinco ou dez segundos com banda morta. Se um valor só é útil como total do turno, considere deixar o CLP acumulá-lo e publicá-lo em um ciclo mais lento.

Boas práticas para produção

1. Somente leitura por definição. Não adicione suporte a escrita em uma ponte de monitoramento. Se algum dia precisar de comandos, crie um serviço separado com lógica de aprovação explícita.
2. Use um conjunto de tags de interface. Peça ao engenheiro de automação que copie os valores necessários para um UDT ou array dedicado. Assim você lê uma única estrutura em vez de dezenas de tags espalhadas, e alterações no programa não quebram a ponte.
3. Respeite os limites de conexão. Os controladores suportam um número limitado de conexões CIP, compartilhadas com IHMs, SCADA e ferramentas de engenharia. Uma conexão de longa duração por ponte é o ideal; evite reconectar a cada ciclo.
4. Faça polling na velocidade necessária. Um segundo é mais que suficiente para contagens de produção e estados. Polling mais rápido adiciona carga sem adicionar informação.
5. Segmentação de rede. Coloque o gateway na rede OT ou em uma DMZ com acesso apenas à porta 44818 dos controladores e saída na porta 8883 para o broker. Nosso artigo sobre DMZ MQTT aborda o layout.
6. Broker privado para dados reais. O servidor MQTT público (gratuito) usado nos exemplos serve para testes e protótipos e pode ser lido por qualquer pessoa. Migre para um broker MQTT gerenciado com TLS e credenciais por broker antes de conectar uma linha real.

Solução de problemas

Onde o ConnectMQ se encaixa

O ConnectMQ é o lado nuvem deste pipeline: um broker MQTT gerenciado com TLS na porta 8883 e WebSockets para dashboards no navegador. Ele detecta estruturas de topics ISA-95, então a visualização de dispositivos agrupa linhas e máquinas automaticamente, e regras de alerta por limite podem sinalizar uma contagem de rejeitos ou uma corrente de motor fora dos limites. Os mesmos topics também podem alimentar um gêmeo digital da linha. Se você é novo em brokers gerenciados, veja como um broker MQTT na nuvem se compara à hospedagem própria e, para a arquitetura completa, leia o guia de MQTT para IoT industrial.

Comece a usar o ConnectMQ

Conecte hoje sua primeira linha EtherNet/IP. Cadastre-se em app.connectmq.com, escolha o plano gratuito Serverless sem cartão de crédito, crie um broker e cole o usuário, a senha e o client ID gerados na ponte. Suas tags Logix aparecerão no explorador de topics do ConnectMQ em segundos.

Perguntas frequentes

Ler tags via EtherNet/IP afeta o programa do CLP?

As mensagens explícitas usam os recursos de comunicação do controlador, não suas conexões de I/O, e não alteram o programa. Mantenha o polling moderado e poucas conexões para não afetar IHMs e SCADA.

Posso usar mensagens implícitas de I/O em um gateway MQTT?

É possível se o gateway estiver configurado como adapter na árvore de I/O do CLP, mas isso exige alterar o projeto. Leituras explícitas de tags são mais simples e não intrusivas para monitoramento.

O pycomm3 funciona com Micro800 e com controladores mais antigos PLC-5 ou SLC?

O pycomm3 é voltado para controladores Logix e tem suporte parcial ao Micro800; as famílias mais antigas PLC-5 e SLC usam endereçamento diferente. Consulte a documentação da biblioteca para o seu controlador específico.

Qual porta as mensagens explícitas EtherNet/IP usam?

A porta TCP 44818. O tráfego implícito de I/O usa a porta UDP 2222.