Gateway OPC UA para MQTT: Dados de CLP na Nuvem

Q: O gateway pode escrever valores de volta no CLP?

Tecnicamente sim, assinando um topic de comando e chamando write_value, mas trate isso como um recurso separado e cuidadosamente protegido. A maioria das implantações mantém o gateway somente leitura.

Um gateway OPC UA para MQTT lê tags de CLPs e servidores OPC UA na rede da planta e as publica como mensagens MQTT em um broker que dashboards, historiadores, um gêmeo digital e aplicações na nuvem podem assinar. Neste tutorial você vai construir um em Python com bibliotecas open source: ele assina nós OPC UA, mapeia-os para topics no estilo ISA-95 e publica JSON limpo em um broker na nuvem. Você pode rodar tudo em um notebook em cerca de quinze minutos.

Por que fazer a ponte entre OPC UA e MQTT?

O OPC UA é excelente no chão de fábrica. Ele tem um modelo de informação rico, segurança embutida e amplo suporte entre os fabricantes de CLP. Mas é um protocolo client/servidor: cada consumidor abre sua própria sessão com cada servidor. Isso funciona para um SCADA conversando com dez CLPs e fica complicado quando vinte aplicações, algumas delas na nuvem, querem os mesmos dados.

O MQTT inverte essa relação. O gateway mantém uma sessão OPC UA por servidor e uma conexão MQTT de saída, e qualquer número de consumidores assina o broker. O firewall da planta só precisa de uma conexão de saída na porta 8883, e os CLPs nunca veem tráfego da nuvem. Se você está comparando os dois protocolos de forma mais ampla, nosso comparativo OPC UA vs MQTT aborda os trade-offs; a versão curta é que a maioria das arquiteturas usa os dois.

Arquitetura do 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

O gateway tem quatro responsabilidades:

Assinar, não fazer polling. As subscriptions OPC UA permitem que o servidor envie mudanças de valor em um intervalo de amostragem que você escolhe, o que é muito mais eficiente do que ler todas as tags em um temporizador.
Mapear nós para topics. Node IDs como ns=2;s=Line1.CNC1.SpindleTemp não significam nada para um desenvolvedor de nuvem. O gateway os traduz para uma hierarquia legível.
Normalizar os payloads. Cada mensagem recebe o valor, o status OPC UA, um timestamp de origem e uma unidade de engenharia.
Sobreviver a desconexões. Os dois lados vão cair em algum momento; o gateway precisa se reconectar sem perder a configuração.

Pré-requisitos

Python 3.10 ou mais recente.
asyncua (a biblioteca client open source do FreeOpcUa) e paho-mqtt 2.x.
Um servidor OPC UA. Se você não tiver um CLP à mão, o tutorial inclui um pequeno servidor simulado.
Um broker MQTT. Para testes, use o servidor MQTT público (gratuito) em broker.connectmq.com; para dados reais da planta, use credenciais privadas de um broker MQTT gerenciado.

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

Passo 1: um servidor OPC UA simulado (opcional)

Se você já tem um servidor OPC UA, pule este passo e anote a URL do endpoint e os node IDs. Caso contrário, este script expõe três tags variáveis em opc.tcp://localhost:4840.

# sim_server.py - um pequeno servidor OPC UA com valores de CNC variando
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 os node IDs usam identificadores do tipo string no namespace registrado (índice 2 aqui), eles aparecerão como ns=2;s=Line1.CNC1.SpindleTemp e assim por diante. Em um CLP real, navegue pelo address space com uma ferramenta client como o UaExpert para encontrar os node IDs exatos.

Passo 2: o mapa de tags

Mantenha o mapeamento na configuração, não no código. Cada entrada associa um nó OPC UA ao caminho de topic e à unidade que você quer que os consumidores vejam. O layout de topics segue uma hierarquia no estilo ISA-95 (site, área, linha, célula), que nosso guia sobre topics MQTT ISA-95 explica em profundidade.

{
  "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" }
  ]
}

Salve-o como tags.json. Substitua a91c por uma string aleatória sua, para que seus topics não colidam com os de outras pessoas no servidor MQTT público. Os topics resultantes têm oito níveis, o máximo que o servidor MQTT público permite; em um broker privado você tem mais espaço.

Passo 3: o código do gateway OPC UA para MQTT

# gateway.py - subscription OPC UA -> publish 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          # produção: "YOUR_USERNAME"
MQTT_PASS = None          # produção: "YOUR_PASSWORD"
CLIENT_ID = f"opcua-gw-{uuid.uuid4().hex[:8]}"   # produção: "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:
    """O asyncua chama datachange_notification a cada mudança de um 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 publicação em ms
                nodes = [opc.get_node(n) for n in node_map]
                await sub.subscribe_data_change(nodes)
                while True:                       # mantém a sessão viva e detecta falhas
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:                  # reinício do servidor, perda de rede 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()))

Algumas decisões de projeto merecem destaque:

Retained messages fazem com que um dashboard que se conecte mais tarde veja imediatamente o último valor de cada tag.
O Last Will no topic de status muda para {"online": false} se o gateway morrer sem uma desconexão limpa, para que os consumidores distingam dados desatualizados de dados ao vivo.
O status code OPC UA viaja junto com cada valor. Uma leitura Good e uma leitura BadCommunicationError nunca devem parecer iguais mais adiante na cadeia.
check_connection() lança uma exceção quando a sessão é perdida, o que nos leva ao loop de nova tentativa e reconstrói a subscription.

Passo 4: execute e verifique

Inicie o servidor simulado e depois o gateway, em dois terminais:

python sim_server.py
python gateway.py

Em um terceiro terminal, assine tudo abaixo da sua raiz:

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

Você deverá ver linhas 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}

Indo para produção

O servidor MQTT público pode ser lido por qualquer pessoa, então serve apenas para testes e protótipos com dados de amostra. Para dados reais de CLP, seja de uma planta em Manaus ou de uma estação de saneamento, crie um broker privado e defina MQTT_USER, MQTT_PASS e CLIENT_ID com as credenciais do portal. O ConnectMQ vincula os client IDs às credenciais emitidas, então use o client ID gerado em vez de um aleatório.

Checklist antes de colocar o gateway em uma rede de planta:

Item	Por que importa
Modo de segurança OPC UA `SignAndEncrypt` com certificado de client	Sessões anônimas e sem criptografia servem para laboratório, não para um CLP
Usuário OPC UA dedicado e somente leitura	O gateway nunca deve conseguir escrever setpoints
Intervalo de amostragem por classe de tag	Rápido para vibração, lento para nível de tanques; evite inundar o broker
Banda morta em tags analógicas	Reduz mensagens causadas por ruído, veja report by exception
Rodar como serviço com reinício automático	systemd ou um serviço do Windows, com rotação de logs
Buffer durante quedas do broker	O paho enfileira enquanto está desconectado, mas para quedas longas considere store and forward
Regra de firewall somente de saída para a porta 8883	Nenhuma porta de entrada aberta para a rede de TO

No ConnectMQ, o explorador de topics ao vivo do portal mostra sua árvore ISA-95 à medida que ela se preenche, e a visão de dispositivos agrupa as máquinas por níveis ISA-95, uma forma rápida de confirmar que o seu mapeamento de topics é consistente entre as linhas. Regras de alerta por limite podem então sinalizar, por exemplo, uma temperatura de spindle acima de um limite sem escrever nenhum código extra. Para ver onde este gateway se encaixa no panorama geral, consulte nosso guia de IoT industrial com MQTT.

Solução de problemas

BadIdentityTokenRejected ou BadUserAccessDenied: o servidor não permite acesso anônimo. Passe um usuário com Client.set_user() e set_password() ou configure certificados.
BadNodeIdUnknown: a string do node ID está errada, muitas vezes o índice do namespace. Os índices de namespace podem mudar entre versões do servidor; resolva-os pela URI do namespace com get_namespace_index() na inicialização.
Os valores chegam, mas os timestamps são nulos: alguns servidores não definem um timestamp de origem. O código recorre ao timestamp do servidor.
O MQTT conecta e desconecta imediatamente: no servidor MQTT público, verifique se o seu client ID é único e não vazio. Em um broker privado, verifique se o client ID corresponde às credenciais emitidas.
Topics rejeitados no servidor MQTT público: você excedeu oito níveis de topic. Encurte a raiz ou o caminho.

Rode seu gateway OPC UA no ConnectMQ

Crie um broker gratuito em app.connectmq.com no plano Serverless, sem cartão de crédito. Copie o usuário, a senha e o client ID gerados para o gateway.py e, em poucos minutos, suas tags de CLP aparecerão em um broker MQTT privado na nuvem, protegido por TLS e pronto para dashboards e historiadores.

Perguntas frequentes

Preciso de um produto comercial para um gateway OPC UA para MQTT?

Não. Bibliotecas open source como asyncua e paho-mqtt bastam para muitos projetos. Gateways comerciais acrescentam interfaces gráficas, store and forward e suporte do fabricante, o que pode fazer diferença em escala.

O gateway deve fazer polling das tags OPC UA ou usar subscriptions?

Use subscriptions. O servidor amostra os valores e envia apenas as mudanças, o que reduz a carga no CLP e na rede em comparação com leituras cíclicas.

O gateway pode escrever valores de volta no CLP?

Tecnicamente sim, assinando um topic de comando e chamando write_value, mas trate isso como um recurso separado e cuidadosamente protegido. A maioria das implantações mantém o gateway somente leitura.

Qual QoS MQTT devo usar para dados de tags de CLP?

QoS 1 é a escolha usual: a entrega é garantida pelo menos uma vez e o overhead é baixo. Combine-o com retained messages para que novos consumidores recebam o estado atual imediatamente.

Gateway OPC UA para MQTT: Levando Dados de CLPs para a Nuvem