Gemelo digital de brazo robótico con MQTT: replica los ángulos de las articulaciones en 3D y envía comandos

Un gemelo digital de brazo robótico es una réplica 3D en vivo de un brazo que se mueve exactamente igual que el real, controlada por la telemetría de los ángulos de sus articulaciones y no por una animación pregrabada. En esta guía transmitirás ángulos de articulación por MQTT, animarás un modelo 3D articulado en el navegador con Three.js y enviarás comandos de movimiento al brazo con confirmaciones. Todo funciona contra un broker real en pocos minutos.

Si antes quieres el contexto general, la guía de tecnología de gemelo digital explica dónde encaja este tipo de gemelo digital 3D entre los demás tipos de gemelos.

Por qué un brazo robótico es el primer gemelo 3D ideal

Los brazos robóticos son un punto de partida perfecto para los gemelos 3D porque su estado es pequeño y está bien definido. Un brazo de seis ejes queda totalmente descrito por seis ángulos de articulación más algunos indicadores como el estado de la pinza, el programa en ejecución y el código de fallo. Es un payload diminuto y, sin embargo, cuando aplicas esos ángulos a una cadena cinemática en 3D, el resultado lo entiende al instante cualquiera que pase por delante de la pantalla: el gemelo se estira, gira y agarra igual que la célula de la planta.

Esa compacidad importa para el transporte. Puedes publicar la pose completa varias veces por segundo por MQTT sin sobrecargar la red, y cada suscriptor, ya sea un visor 3D, un historiador o un script de analítica, recibe el mismo flujo desde el broker.

Arquitectura del gemelo digital de brazo robótico

El diseño usa tres roles conectados a través de un único broker:

1. Gateway del robot: lee las posiciones de las articulaciones desde el controlador (o las simula) y publica la pose. También se suscribe a los topics de comandos y reenvía al controlador los comandos aceptados.
2. Visor 3D: una página de navegador que se suscribe a la pose a través de WebSocket seguro y aplica cada ángulo al nodo correspondiente de un modelo 3D.
3. Operario o aplicación: publica comandos como "mover a la pose" o "abrir la pinza" y escucha las confirmaciones.

La telemetría de articulaciones usa QoS 0 porque una pose nueva siempre sustituye a la anterior; perder un fotograma es inofensivo. Los comandos y las confirmaciones usan QoS 1 para que se entreguen al menos una vez, y el id del comando permite al gateway ignorar duplicados. Si necesitas repasar ese compromiso, consulta los niveles de QoS de MQTT explicados.

En este tutorial usamos el servidor MQTT público gratuito en broker.connectmq.com con un prefijo de topic único. Recuerda que el servidor público lo puede leer cualquiera y es solo para pruebas y prototipos, así que nunca envíes allí datos reales de producción.

Paso 1: publica los ángulos de las articulaciones desde el gateway

El script siguiente simula un brazo de seis ejes que se aproxima suavemente a los ángulos objetivo, publica su pose cinco veces por segundo, mantiene un mensaje de estado retenido y acepta comandos de movimiento. Sustituye ra-7f3k por tu propio ID aleatorio.

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  # duplicado de QoS 1, ya procesado
    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 nos mantiene muy por debajo del límite del broker público
except KeyboardInterrupt:
    client.publish(f"{PREFIX}/status", "offline", qos=1, retain=True)
    client.loop_stop()
    client.disconnect()

Instala la librería con pip install "paho-mqtt>=2.0" y ejecuta el script. Observa el flujo desde otra terminal:

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

En una célula real, sustituye el bucle de aproximación por lecturas de la interfaz del controlador que proporciona el fabricante de tu robot y reenvíale los comandos aceptados. Mantén en cualquier caso la comprobación de rangos en el gateway: el gemelo nunca debe ser lo que lleve un eje más allá de su límite.

Paso 2: construye en el navegador el gemelo digital 3D del brazo robótico

Un brazo robótico en Three.js es una cadena de grupos anidados. Cada articulación es un grupo que gira en torno a un eje, y cada hijo hereda las transformaciones de sus padres, que es exactamente como funciona la cinemática directa. Si cargas un modelo glTF exportado desde CAD en lugar de primitivas, se aplica la misma idea: localiza por nombre el nodo de cada articulación y gíralo. La guía para vincular datos IoT a un modelo 3D trata las convenciones de nombres de esos nodos.

Crea un proyecto con npm install three mqtt y un bundler como Vite, y usa este módulo como punto 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;
}

// Cadena cinemática: cada articulación es un Group anidado dentro de su padre.
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; // suavizar entre fotogramas
  }
  renderer.render(scene, camera);
});

La interpolación en el bucle de renderizado es importante. La telemetría llega a 5 Hz mientras la pantalla renderiza a 60 Hz, así que saltar directamente a cada nuevo ángulo da un aspecto entrecortado. Aproximarse suavemente al último objetivo oculta el hueco sin falsear la pose real. Para un recorrido más profundo por escenas, cámaras y loaders, consulta el tutorial de gemelo digital con Three.js y la documentación oficial de Three.js.

Paso 3: cierra el bucle con comandos

Un gemelo que solo escucha es una sombra digital. Para hacerlo bidireccional, publica un comando y espera la confirmación correspondiente:

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}}'

El gateway comprueba los límites, actualiza sus objetivos y publica {"id":"c-42","result":"accepted"} en cmd/ack. El modelo 3D gira entonces hasta la nueva pose porque sigue la telemetría, no el comando. Esa distinción es la regla de diseño más importante de un gemelo digital de brazo robótico: el visor siempre renderiza lo que informa el robot, nunca lo que alguien ha pedido. Si el controlador rechaza un movimiento, el gemelo se queda quieto y el operario ve el motivo del rechazo.

En producción, añade estas salvaguardas:

• Condiciona los comandos a los enclavamientos del controlador y al propio sistema de seguridad del robot. Los comandos MQTT deben solicitar movimientos, no eludir la seguridad.
• Incluye una caducidad en los comandos para que un mensaje retrasado no se ejecute tarde. Con MQTT 5, establece el intervalo de caducidad del mensaje al publicar; la guía de características de MQTT 5 muestra cómo.
• Usa credenciales distintas para visores y operarios, y registra cada comando y cada confirmación.

Pasar de la demo a producción

El servidor público gratuito es para experimentos: limita a los clientes a 10 mensajes por segundo y 64 KB por mensaje, y todo lo que publicas allí es visible para otros. Para una célula real, pasa a un broker MQTT gestionado con credenciales privadas y TLS. En ConnectMQ creas un broker en el portal, copias el nombre de usuario, la contraseña y el client ID generados, y cambias dos cosas en el código: conectar con TLS en el puerto 8883 (o WSS en el 8084 para el navegador) y pasar las credenciales.

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()  # el almacén de CA del sistema valida el certificado de Let's Encrypt
client.connect("YOUR_BROKER_HOST", 8883, keepalive=30)

Como en ConnectMQ los client ID están ligados a las credenciales emitidas, da al gateway y a cada visor su propia identidad en lugar de compartir una. El explorador de topics en vivo del portal resulta útil durante la puesta en marcha: puedes ver el flujo de articulaciones y las confirmaciones en paralelo sin escribir herramientas adicionales.

Lista de comprobación para resolver problemas

Ejecuta tu gemelo de brazo robótico en ConnectMQ

Ya tienes un gemelo digital de brazo robótico funcional: un gateway que transmite ángulos de articulación, un visor 3D que los sigue y una ruta de comandos con confirmaciones. Haz prototipos en broker.connectmq.com y después traslada el mismo código a un broker privado de ConnectMQ cuando la célula entre en producción. Crea un broker Serverless gratuito, sin tarjeta de crédito, y tendrás endpoints TLS y credenciales en minutos.

Preguntas frecuentes

¿Con qué frecuencia debe actualizarse un gemelo digital de brazo robótico?

De cinco a diez actualizaciones de pose por segundo suelen verse fluidas cuando el visor interpola entre fotogramas. Frecuencias mayores añaden carga sin beneficio visible en la mayoría de casos de monitorización.

¿Puedo usar un modelo CAD en lugar de primitivas?

Sí. Exporta el brazo a glTF con un nodo por articulación, cárgalo con GLTFLoader y gira esos nodos exactamente igual que los grupos de este tutorial. Asegúrate de que el pivote de cada nodo esté sobre el eje de la articulación.

¿Es seguro controlar un robot real por MQTT?

MQTT puede transportar peticiones de movimiento, pero la seguridad debe permanecer en el controlador del robot y en su sistema de seguridad. Valida los límites en el gateway, usa comandos con caducidad y clientes autenticados, y nunca permitas que el gemelo eluda los enclavamientos.

¿Por qué el visor sigue la telemetría en lugar de los comandos?

Los comandos pueden rechazarse, retrasarse o ejecutarse parcialmente. Renderizar solo la pose informada mantiene fiel al gemelo, que es precisamente el sentido de un gemelo digital.