Ponte OPC UA para MQTT para dados industriais

Uma ponte OPC UA para MQTT se conecta a um servidor OPC UA como cliente, assina as tags que interessam e republica cada mudança de dados como mensagem MQTT em um tópico bem nomeado. A planta mantém o OPC UA onde ele é forte — modelos ricos, segurança, integração de máquinas — enquanto dashboards, historiadores e serviços em nuvem recebem um fluxo publish/subscribe leve que podem escalar livremente.

Este tutorial monta essa ponte em Python com asyncua e paho-mqtt, mapeia node ids OPC UA para uma árvore de tópicos de Unified Namespace e mostra como testá-la com um servidor OPC UA simulado.

Por que uma ponte em vez de conexão direta

OPC UA é cliente/servidor: cada consumidor abre uma sessão com cada servidor de que precisa. Com cinco consumidores e vinte servidores, são cem sessões, cada uma com suas subscriptions, certificados e carga no servidor. Uma ponte transforma isso em um hub:

Por padrão a ponte é mão única: OPC UA entra, MQTT sai. Assim o lado de TA fica protegido enquanto a TI recebe os dados. Para uma comparação mais ampla dos dois protocolos, leia MQTT vs OPC UA.

1. Suba um servidor OPC UA de teste

Não desenvolva contra um servidor de produção. O simulador open source OPC PLC da Microsoft roda em Docker e expõe valores que mudam:

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

O endpoint é opc.tcp://localhost:50000. Navegue por ele com um cliente gráfico como o UaExpert, ou com o comando uals que vem com o asyncua, e anote os node ids que quer levar para a ponte. O simulador expõe nós como StepUp, AlternatingBoolean e RandomSignedInt32 no seu próprio namespace; confira o índice do namespace no navegador, porque ele pode variar entre versões.

2. Mapeie nós para tópicos

O mapeamento é o verdadeiro trabalho de design. Um tópico deve dizer onde o valor vive na planta, não qual servidor ou node id o fornece hoje. Mantenha o mapa em configuração:

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",
}

Quando depois você trocar o servidor simulado pelo real — ou mover uma tag para outro CLP — só o lado esquerdo do mapa muda. Todos os assinantes continuam funcionando. Essa é a ideia central de um Unified Namespace: a árvore de tópicos é o contrato, as fontes são detalhes de implementação.

3. A ponte

A API de subscriptions do asyncua chama um handler sempre que um monitored item muda. O handler publica o valor, sua qualidade OPC UA e o timestamp de origem:

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 publicação de 500 ms
                await sub.subscribe_data_change(nodes)
                mq.publish(STATUS, "online", qos=1, retain=True)
                print("ponte ativa para", len(nodes), "nós")
                while True:
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:          # conexão perdida, reinício do servidor, node id inválido
            print("Erro 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

O que esse design oferece:

• Tráfego guiado por mudanças. O OPC UA envia notificações só quando os valores mudam, então o lado MQTT não carrega amostras redundantes. Ajuste o intervalo de publicação (500 ms aqui) e o intervalo de amostragem de cada item ao seu processo.
• A qualidade viaja com o valor. Um status code OPC UA ruim ou incerto é informação; mantenha-o no payload em vez de descartar a mensagem em silêncio.
• Timestamps de origem. Historiadores devem armazenar quando o valor mudou na origem, não quando a ponte o encaminhou.
• Valores retidos e tópico de status. Novos assinantes veem o estado atual na hora, e o Last Will muda o status da ponte para offline se o processo morrer. Mais em mensagens retidas.
• Recuperação automática. Se o servidor OPC UA reiniciar, o loop externo reconecta e recria a subscription. O paho reconecta sozinho ao broker após loop_start().

4. Teste a ponte

1. Inicie o simulador e a ponte. A ponte deve imprimir o número de nós ligados.
2. Abra o cliente MQTT online, conecte em broker.connectmq.com e assine testmqtt/acme/plant1/#. filler/counter deve subir, filler/running alternar e _bridge/status mostrar online.
3. Pare o contêiner do simulador. Em poucos segundos o status deve virar opcua-unreachable; reinicie-o e os valores devem voltar sem reiniciar a ponte.
4. Mate a ponte com kill -9. Depois do timeout do keep-alive, o broker publica o Last Will: offline.

Veja as tags OPC UA no navegador

Confira com o topic matcher se os filtros usados pelos consumidores batem com o que a ponte publica, e lembre dos valores estruturados grandes (arrays, extension objects): a calculadora de tamanho de pacote ajuda a estimar quanto uma notificação custa na rede.

Outras formas de fazer a ponte

Uma pequena ponte em Python é ideal para aprender, para um punhado de servidores e para ter controle total do mapeamento. Em ambientes maiores, considere:

Qualquer que seja a ferramenta, as regras são as mesmas: tópicos descrevem a planta, payloads carregam valor + qualidade + timestamp, a ponte anuncia a própria saúde e escritas de volta no OPC UA são um caminho separado e protegido de propósito.

Checklist de produção

• Use a segurança do OPC UA (Basic256Sha256 com assinatura e criptografia) e um certificado de aplicação dedicado para a ponte em servidores reais; a opção sem segurança do simulador é só para testes locais.
• Leia os limites do servidor de monitored items por subscription e divida listas grandes de tags em várias subscriptions.
• Rode uma ponte por área ou linha para que uma falha fique local.
• Versione o mapa nó→tópico e revise-o como código.
• Monitore o tópico de status da ponte e gere alerta quando não estiver online.

Uma ponte OPC UA para MQTT permite manter o OPC UA para o que ele faz de melhor e dar a todos os outros sistemas um fluxo simples e escalável. Mapeie nós para tópicos com o formato da planta, publique notificações de mudança com qualidade e timestamps, teste contra um simulador e o cliente online, e seus dados de TA viram algo que a empresa inteira pode assinar.

Perguntas frequentes