Gêmeo digital de braço robótico com MQTT: espelhe ângulos das juntas em 3D e envie comandos de volta

Um gêmeo digital de braço robótico é uma réplica 3D ao vivo de um braço que se move exatamente como o real, guiada pela telemetria dos ângulos das juntas e não por uma animação pré-gravada. Neste guia você vai transmitir ângulos das juntas via MQTT, animar um modelo 3D articulado no navegador com Three.js e enviar comandos de movimento de volta ao braço, com confirmações. Tudo roda contra um broker real em poucos minutos.

Se quiser primeiro o contexto mais amplo, o guia de tecnologia de gêmeo digital explica onde esse tipo de gêmeo 3D se encaixa entre os outros tipos de gêmeo.

Por que um braço robótico é o primeiro gêmeo 3D ideal

Braços robóticos são um ponto de partida perfeito para gêmeos 3D porque seu estado é pequeno e bem definido. Um braço de seis eixos é totalmente descrito por seis ângulos de junta e algumas flags, como estado da garra, programa em execução e código de falha. É um payload minúsculo, mas quando você aplica esses ângulos a uma cadeia cinemática em 3D, o resultado é imediatamente compreensível para qualquer pessoa que passe pela tela: o gêmeo estende, gira e agarra exatamente como a célula no chão de fábrica.

Essa compacidade importa para o transporte. Você pode publicar a pose completa várias vezes por segundo via MQTT sem sobrecarregar a rede, e todo assinante, seja um visualizador 3D, um historiador ou um script de análise, recebe o mesmo fluxo a partir do broker.

Arquitetura do gêmeo digital de braço robótico

O design usa três papéis conectados por um único broker:

1. Gateway do robô: lê as posições das juntas no controlador (ou as simula) e publica a pose. Ele também assina os topics de comando e encaminha ao controlador os comandos aceitos.
2. Visualizador 3D: uma página no navegador que assina a pose via Secure WebSocket e aplica cada ângulo de junta ao nó correspondente de um modelo 3D.
3. Operador ou aplicação: publica comandos como "mover para pose" ou "abrir garra" e aguarda as confirmações.

A telemetria das juntas usa QoS 0 porque uma pose mais nova sempre substitui a anterior; perder um quadro não causa dano. Comandos e confirmações usam QoS 1 para serem entregues pelo menos uma vez, e o id do comando permite que o gateway ignore duplicatas. Se precisar relembrar esse trade-off, veja níveis de QoS do MQTT explicados.

Neste tutorial usamos o servidor MQTT público (gratuito) em broker.connectmq.com com um prefixo de topic exclusivo. Lembre-se de que o servidor MQTT público pode ser lido por qualquer pessoa e serve apenas para testes e protótipos, então nunca envie dados reais de produção para lá.

Passo 1: publique os ângulos das juntas a partir do gateway

O script abaixo simula um braço de seis eixos que se aproxima suavemente dos ângulos-alvo, publica sua pose cinco vezes por segundo, mantém uma mensagem de estado retida e aceita comandos de movimento. Substitua ra-7f3k pelo seu próprio ID aleatório.

import json
import random
import time

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PREFIX = "connectmq/demo/ra-7f3k/cell1/robot1"
CLIENT_ID = f"ra-7f3k-gateway-{random.randint(1000, 9999)}"

joints = {f"j{i}": 0.0 for i in range(1, 7)}
targets = dict(joints)
LIMITS = {"j1": (-170, 170), "j2": (-90, 90), "j3": (-120, 120),
          "j4": (-180, 180), "j5": (-110, 110), "j6": (-360, 360)}
seen_commands = set()


def on_connect(client, userdata, flags, reason_code, properties):
    print("Connected:", reason_code)
    client.publish(f"{PREFIX}/status", "online", qos=1, retain=True)
    client.publish(f"{PREFIX}/state", json.dumps({"mode": "auto", "gripper": "open"}),
                   qos=1, retain=True)
    client.subscribe(f"{PREFIX}/cmd/move", qos=1)


def on_message(client, userdata, msg):
    try:
        cmd = json.loads(msg.payload)
        cmd_id = cmd["id"]
    except (ValueError, KeyError):
        return
    if cmd_id in seen_commands:
        return  # duplicata de QoS 1, já tratada
    seen_commands.add(cmd_id)

    result = "accepted"
    for name, value in cmd.get("target", {}).items():
        low, high = LIMITS.get(name, (0, 0))
        if name not in LIMITS or not low <= float(value) <= high:
            result = f"rejected: {name} out of range"
            break
    if result == "accepted":
        targets.update({k: float(v) for k, v in cmd["target"].items()})
    client.publish(f"{PREFIX}/cmd/ack", json.dumps({"id": cmd_id, "result": result}), qos=1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                     protocol=mqtt.MQTTv5)
client.will_set(f"{PREFIX}/status", "offline", qos=1, retain=True)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=30)
client.loop_start()

try:
    while True:
        for name in joints:
            joints[name] += (targets[name] - joints[name]) * 0.15
        payload = {k: round(v, 2) for k, v in joints.items()}
        payload["ts"] = int(time.time() * 1000)
        client.publish(f"{PREFIX}/joints", json.dumps(payload), qos=0)
        time.sleep(0.2)  # 5 Hz fica bem abaixo do limite de taxa do broker público
except KeyboardInterrupt:
    client.publish(f"{PREFIX}/status", "offline", qos=1, retain=True)
    client.loop_stop()
    client.disconnect()

Instale a biblioteca com pip install "paho-mqtt>=2.0" e execute o script. Acompanhe o fluxo em outro terminal:

mosquitto_sub -h broker.connectmq.com -p 1883 -i ra-7f3k-watch -t "connectmq/demo/ra-7f3k/cell1/robot1/#" -v

Em uma célula real, substitua o laço de suavização por leituras da interface de controlador fornecida pelo fabricante do robô e encaminhe a ele os comandos aceitos. Mantenha a verificação de faixa no gateway de qualquer forma: o gêmeo nunca pode ser o responsável por levar um eixo além do seu limite.

Passo 2: construa o gêmeo digital 3D do braço robótico no navegador

Um braço robótico em Three.js é uma cadeia de grupos aninhados. Cada junta é um grupo que gira em torno de um eixo, e cada filho herda as transformações dos pais, que é exatamente como a cinemática direta funciona. Se você carregar um modelo glTF exportado do CAD em vez de primitivas, a mesma ideia vale: encontre o nó de cada junta pelo nome e gire-o. O guia para vincular dados de IoT a um modelo 3D trata das convenções de nomenclatura desses nós.

Crie um projeto com npm install three mqtt e um bundler como o Vite, e use este módulo como ponto de entrada:

import * as THREE from "three";
import mqtt from "mqtt";

const PREFIX = "connectmq/demo/ra-7f3k/cell1/robot1";
const scene = new THREE.Scene();
scene.background = new THREE.Color(0x1e1f24);
const camera = new THREE.PerspectiveCamera(50, innerWidth / innerHeight, 0.1, 100);
camera.position.set(3, 2.5, 3);
camera.lookAt(0, 1, 0);
const renderer = new THREE.WebGLRenderer({ antialias: true });
renderer.setSize(innerWidth, innerHeight);
document.body.appendChild(renderer.domElement);
scene.add(new THREE.HemisphereLight(0xffffff, 0x444444, 2.5));

const mat = new THREE.MeshStandardMaterial({ color: 0xf28c28 });
function link(length, radius = 0.08) {
  const mesh = new THREE.Mesh(new THREE.CylinderGeometry(radius, radius, length), mat);
  mesh.position.y = length / 2;
  return mesh;
}

// Cadeia cinemática: cada junta é um Group aninhado no seu pai.
const joints = {};
const axes = { j1: "y", j2: "z", j3: "z", j4: "y", j5: "z", j6: "y" };
const lengths = { j1: 0.4, j2: 0.9, j3: 0.7, j4: 0.2, j5: 0.2, j6: 0.1 };
let parent = scene;
for (const name of Object.keys(axes)) {
  const g = new THREE.Group();
  g.add(link(lengths[name]));
  parent.add(g);
  joints[name] = g;
  const tip = new THREE.Group();
  tip.position.y = lengths[name];
  g.add(tip);
  parent = tip;
}

const target = {};
const client = mqtt.connect("wss://broker.connectmq.com:8084/mqtt", {
  clientId: "ra-7f3k-viewer-" + Math.random().toString(16).slice(2, 8),
  protocolVersion: 5,
});
client.on("connect", () => client.subscribe(`${PREFIX}/#`, { qos: 1 }));
client.on("message", (topic, payload) => {
  if (topic.endsWith("/joints")) Object.assign(target, JSON.parse(payload.toString()));
  if (topic.endsWith("/status")) document.title = "Robot1: " + payload.toString();
});

const toRad = THREE.MathUtils.degToRad;
renderer.setAnimationLoop(() => {
  for (const [name, axis] of Object.entries(axes)) {
    if (target[name] === undefined) continue;
    const g = joints[name];
    g.rotation[axis] += (toRad(target[name]) - g.rotation[axis]) * 0.3; // suaviza entre quadros
  }
  renderer.render(scene, camera);
});

A interpolação no laço de renderização é importante. A telemetria chega a 5 Hz enquanto a tela renderiza a 60 Hz, então saltar direto para cada novo ângulo fica travado. Aproximar-se suavemente do alvo mais recente esconde esse intervalo sem falsear a pose real. Para um passo a passo mais aprofundado sobre cenas, câmeras e loaders, veja o tutorial de gêmeo digital com Three.js e a documentação oficial do Three.js.

Passo 3: feche o ciclo com comandos

Um gêmeo que só escuta é uma sombra digital. Para torná-lo bidirecional, publique um comando e aguarde a confirmação correspondente:

mosquitto_pub -h broker.connectmq.com -p 1883 -i ra-7f3k-operator -q 1 \
  -t "connectmq/demo/ra-7f3k/cell1/robot1/cmd/move" \
  -m '{"id":"c-42","target":{"j1":45,"j2":-30,"j3":60,"j5":20}}'

O gateway verifica os limites, atualiza seus alvos e publica {"id":"c-42","result":"accepted"} em cmd/ack. O modelo 3D então se move para a nova pose porque segue a telemetria, não o comando. Essa distinção é a regra de design mais importante de um gêmeo digital de braço robótico: o visualizador sempre renderiza o que o robô informa, nunca o que alguém pediu. Se o controlador recusar um movimento, o gêmeo fica parado e o operador vê o motivo da rejeição.

Em produção, adicione estas salvaguardas:

• Condicione os comandos aos intertravamentos do controlador e ao próprio sistema de segurança do robô. Comandos MQTT devem solicitar movimentos, não contornar a segurança.
• Inclua uma validade nos comandos para que uma mensagem atrasada não seja executada tarde demais. Com MQTT 5, defina o message expiry interval na publicação; o guia de recursos do MQTT 5 mostra como.
• Use credenciais separadas para visualizadores e operadores, e registre em log cada comando e cada confirmação.

Da demonstração para a produção

O servidor MQTT público serve para experimentos: ele limita os clientes a 10 mensagens por segundo e 64 KB por mensagem, e tudo o que você publica lá fica visível para outras pessoas. Para uma célula real, como uma célula de solda ou de paletização em uma planta em Joinville, migre para um broker MQTT gerenciado com credenciais privadas e TLS. No ConnectMQ você cria um broker no portal, copia o usuário, a senha e o client ID gerados e muda duas coisas no código: conectar com TLS na porta 8883 (ou WSS na 8084 para o navegador) e passar as credenciais.

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID",
                     protocol=mqtt.MQTTv5)
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # o repositório de CAs do sistema valida o certificado Let's Encrypt
client.connect("YOUR_BROKER_HOST", 8883, keepalive=30)

Como no ConnectMQ os client IDs são vinculados às credenciais emitidas, dê ao gateway e a cada visualizador uma identidade própria em vez de compartilhar uma só. O explorador de topics ao vivo do portal é útil durante o comissionamento: você acompanha o fluxo das juntas e as confirmações lado a lado sem escrever ferramentas extras.

Checklist de solução de problemas

Rode seu gêmeo de braço robótico no ConnectMQ

Agora você tem um gêmeo digital de braço robótico funcionando: um gateway transmitindo ângulos das juntas, um visualizador 3D que os acompanha e um caminho de comandos com confirmações. Prototipe no servidor MQTT público em broker.connectmq.com e depois leve o mesmo código para um broker ConnectMQ privado quando a célula entrar em operação. Crie um broker Serverless gratuito, sem cartão de crédito, e você terá endpoints TLS e credenciais em minutos.

Perguntas frequentes

Com que frequência um gêmeo digital de braço robótico deve ser atualizado?

Cinco a dez atualizações de pose por segundo costumam parecer fluidas quando o visualizador interpola entre os quadros. Taxas maiores aumentam a carga sem benefício visível na maioria dos casos de monitoramento.

Posso usar um modelo CAD em vez de primitivas?

Sim. Exporte o braço para glTF com um nó por junta, carregue-o com o GLTFLoader e gire esses nós exatamente como os grupos deste tutorial. Garanta que o pivô de cada nó fique sobre o eixo da junta.

É seguro controlar um robô real via MQTT?

O MQTT pode transportar solicitações de movimento, mas a segurança deve permanecer no controlador do robô e no seu sistema de segurança. Valide os limites no gateway, use comandos com validade e clientes autenticados, e nunca deixe o gêmeo contornar os intertravamentos.

Por que o visualizador segue a telemetria e não os comandos?

Comandos podem ser rejeitados, atrasados ou executados parcialmente. Renderizar apenas a pose informada mantém o gêmeo fiel à realidade, que é justamente o propósito de um gêmeo digital.