Jumeau numérique de bras robotique avec MQTT : refléter les angles des axes en 3D et renvoyer des commandes

Un jumeau numérique de bras robotique est une réplique 3D en direct d'un bras, qui bouge exactement comme le vrai, piloté par la télémétrie des angles d'axes plutôt que par une animation préenregistrée. Dans ce guide, vous allez diffuser les angles des axes via MQTT, animer un modèle 3D articulé dans le navigateur avec Three.js et renvoyer au bras des commandes de déplacement avec acquittement. Le tout fonctionne avec un vrai broker en quelques minutes.

Si vous souhaitez d'abord le contexte général, le guide de la technologie du jumeau numérique explique la place de ce type de jumeau 3D parmi les autres types de jumeaux.

Pourquoi un bras robotique est le premier jumeau 3D idéal

Les bras robotiques sont un point de départ parfait pour les jumeaux 3D, car leur état est réduit et bien défini. Un bras six axes est entièrement décrit par six angles d'axes plus quelques indicateurs comme l'état de la pince, le programme en cours et le code de défaut. C'est un payload minuscule, et pourtant, une fois ces angles appliqués à une chaîne cinématique en 3D, le résultat est immédiatement compréhensible pour quiconque passe devant l'écran : le jumeau s'étend, pivote et saisit exactement comme la cellule dans l'atelier.

Cette compacité compte pour le transport. Vous pouvez publier la pose complète plusieurs fois par seconde via MQTT sans surcharger le réseau, et chaque abonné, qu'il s'agisse d'une visionneuse 3D, d'un historien ou d'un script d'analyse, reçoit le même flux depuis le broker.

Architecture du jumeau numérique de bras robotique

La conception repose sur trois rôles reliés par un seul broker :

1. La passerelle du robot : lit les positions des axes depuis le contrôleur (ou les simule) et publie la pose. Elle s'abonne aussi aux topics de commande et transmet au contrôleur les commandes acceptées.
2. La visionneuse 3D : une page de navigateur qui s'abonne à la pose via WebSocket sécurisé et applique chaque angle au nœud correspondant d'un modèle 3D.
3. L'opérateur ou l'application : publie des commandes comme « aller à la pose » ou « ouvrir la pince » et écoute les acquittements.

La télémétrie des axes utilise la QoS 0, car une pose plus récente remplace toujours la précédente ; perdre une image est sans conséquence. Les commandes et les acquittements utilisent la QoS 1 pour être livrés au moins une fois, et l'id de commande permet à la passerelle d'ignorer les doublons. Pour vous rafraîchir la mémoire sur ce compromis, consultez les niveaux de QoS MQTT expliqués.

Pour ce tutoriel, nous utilisons le serveur MQTT public gratuit broker.connectmq.com avec un préfixe de topic unique. N'oubliez pas que la zone publique est lisible par tous : n'y envoyez jamais de vraies données de production.

Étape 1 : publier les angles des axes depuis la passerelle

Le script ci-dessous simule un bras six axes qui converge vers des angles cibles, publie sa pose cinq fois par seconde, maintient un message d'état retenu et accepte des commandes de déplacement. Remplacez ra-7f3k par votre propre identifiant aléatoire.

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  # doublon QoS 1, déjà traité
    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 nous maintient bien sous la limite de débit du broker public
except KeyboardInterrupt:
    client.publish(f"{PREFIX}/status", "offline", qos=1, retain=True)
    client.loop_stop()
    client.disconnect()

Installez la bibliothèque avec pip install "paho-mqtt>=2.0" et lancez le script. Observez le flux depuis un autre terminal :

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

Sur une vraie cellule, remplacez la boucle de convergence par des lectures via l'interface de contrôleur fournie par le fabricant de votre robot, et transmettez-lui les commandes acceptées. Conservez quoi qu'il arrive la vérification des plages dans la passerelle : le jumeau ne doit jamais être ce qui pousse un axe au-delà de sa limite.

Étape 2 : construire le jumeau numérique 3D du bras robotique dans le navigateur

Dans Three.js, un bras robotique est une chaîne de groupes imbriqués. Chaque axe est un groupe qui tourne autour d'un seul axe, et chaque enfant hérite des transformations de ses parents : c'est exactement le fonctionnement de la cinématique directe. Si vous chargez un modèle glTF exporté depuis la CAO au lieu de primitives, le principe est le même : retrouvez le nœud de chaque axe par son nom et faites-le pivoter. Le guide pour lier des données IoT à un modèle 3D couvre les conventions de nommage de ces nœuds.

Créez un projet avec npm install three mqtt et un bundler comme Vite, puis utilisez ce module comme point d'entrée :

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

// Chaîne cinématique : chaque axe est un Group imbriqué dans son parent.
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; // lissage entre les images
  }
  renderer.render(scene, camera);
});

L'interpolation dans la boucle de rendu est importante. La télémétrie arrive à 5 Hz alors que l'écran s'affiche à 60 Hz : sauter directement à chaque nouvel angle donne un rendu saccadé. Converger progressivement vers la dernière cible masque cet écart sans travestir la pose réelle. Pour une présentation plus approfondie des scènes, caméras et loaders, consultez le tutoriel jumeau numérique avec Three.js et la documentation officielle de Three.js.

Étape 3 : boucler la boucle avec des commandes

Un jumeau qui ne fait qu'écouter est une ombre numérique. Pour le rendre bidirectionnel, publiez une commande et attendez l'acquittement correspondant :

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

La passerelle vérifie les limites, met à jour ses cibles et publie {"id":"c-42","result":"accepted"} sur cmd/ack. Le modèle 3D pivote alors vers la nouvelle pose, car il suit la télémétrie et non la commande. Cette distinction est la règle de conception la plus importante d'un jumeau numérique de bras robotique : la visionneuse affiche toujours ce que le robot rapporte, jamais ce que quelqu'un a demandé. Si le contrôleur refuse un mouvement, le jumeau reste immobile et l'opérateur voit le motif du refus.

En production, ajoutez ces garde-fous :

• Soumettez les commandes aux verrouillages côté contrôleur et au système de sécurité propre du robot. Les commandes MQTT doivent demander des mouvements, pas contourner la sécurité.
• Donnez une date d'expiration aux commandes pour qu'un message retardé ne soit pas exécuté trop tard. Avec MQTT 5, définissez l'intervalle d'expiration du message lors de la publication ; le guide des fonctionnalités de MQTT 5 montre comment.
• Utilisez des identifiants distincts pour les visionneuses et pour les opérateurs, et journalisez chaque commande et chaque acquittement.

De la démo à la production

Le serveur MQTT public gratuit est fait pour les tests et les prototypes : il limite les clients à 10 messages par seconde et 64 Ko par message, et tout ce que vous y publiez est visible par les autres. Pour une vraie cellule, passez à un broker MQTT managé avec des identifiants privés et TLS. Sur ConnectMQ, vous créez un broker dans le portail, copiez le nom d'utilisateur, le mot de passe et le client ID générés, puis modifiez deux choses dans le code : connectez-vous en TLS sur le port 8883 (ou en WSS sur le 8084 pour le navigateur) et transmettez les identifiants.

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()  # le magasin de CA du système valide le certificat Let's Encrypt
client.connect("YOUR_BROKER_HOST", 8883, keepalive=30)

Comme les client IDs sont liés aux identifiants délivrés sur ConnectMQ, donnez à la passerelle et à chaque visionneuse leur propre identité plutôt que d'en partager une. L'explorateur de topics en direct du portail est pratique pendant la mise en service : vous pouvez suivre côte à côte le flux des axes et les acquittements sans écrire d'outil supplémentaire.

Liste de dépannage

Faites tourner votre jumeau de bras robotique sur ConnectMQ

Vous disposez maintenant d'un jumeau numérique de bras robotique fonctionnel : une passerelle qui diffuse les angles des axes, une visionneuse 3D qui les suit et un circuit de commande avec acquittements. Prototypez sur le serveur MQTT public gratuit broker.connectmq.com, puis transférez le même code vers un broker ConnectMQ privé quand la cellule passe en production. Créez un broker Serverless gratuit, sans carte bancaire, et vous aurez des points d'accès TLS et des identifiants en quelques minutes.

Questions fréquentes

À quelle fréquence un jumeau numérique de bras robotique doit-il se mettre à jour ?

Cinq à dix mises à jour de pose par seconde donnent généralement un rendu fluide quand la visionneuse interpole entre les images. Des fréquences plus élevées ajoutent de la charge sans bénéfice visible pour la plupart des usages de supervision.

Peut-on utiliser un modèle CAO au lieu de primitives ?

Oui. Exportez le bras en glTF avec un nœud par axe, chargez-le avec GLTFLoader et faites pivoter ces nœuds exactement comme les groupes de ce tutoriel. Veillez à ce que le pivot de chaque nœud se trouve sur l'axe de l'articulation.

Est-il sûr de piloter un vrai robot via MQTT ?

MQTT peut transporter des demandes de mouvement, mais la sécurité doit rester dans le contrôleur du robot et son système de sécurité. Validez les limites dans la passerelle, utilisez des commandes à expiration et des clients authentifiés, et ne laissez jamais le jumeau contourner les verrouillages.

Pourquoi la visionneuse suit-elle la télémétrie plutôt que les commandes ?

Les commandes peuvent être rejetées, retardées ou partiellement exécutées. N'afficher que la pose rapportée garde le jumeau fidèle à la réalité, ce qui est toute la raison d'être d'un jumeau numérique.