Digitaler Zwilling für Roboterarme mit MQTT: Gelenkwinkel in 3D spiegeln und Befehle zurücksenden

Ein digitaler Zwilling eines Roboterarms ist eine Live-3D-Nachbildung, die sich exakt wie der reale Arm bewegt – gesteuert von Gelenkwinkel-Telemetrie statt von einer vorgefertigten Animation. In diesem Leitfaden streamen Sie Gelenkwinkel über MQTT, animieren mit Three.js ein gegliedertes 3D-Modell im Browser und senden Bewegungsbefehle mit Bestätigungen an den Arm zurück. Alles läuft in wenigen Minuten gegen einen echten Broker.

Wenn Sie zuerst den größeren Zusammenhang möchten, erklärt der Leitfaden zur Digital-Twin-Technologie, wo diese Art von 3D-Zwilling zwischen anderen Zwillingstypen einzuordnen ist.

Warum ein Roboterarm der ideale erste 3D-Zwilling ist

Roboterarme sind ein perfekter Einstieg in 3D-Zwillinge, weil ihr Zustand klein und klar definiert ist. Ein Sechsachs-Arm wird vollständig durch sechs Gelenkwinkel plus einige Flags wie Greiferzustand, laufendes Programm und Fehlercode beschrieben. Das ist ein winziger Payload, doch wendet man diese Winkel in 3D auf eine kinematische Kette an, ist das Ergebnis für jeden, der am Bildschirm vorbeigeht, sofort verständlich: Der Zwilling greift, dreht und fasst genau wie die Zelle in der Fertigung.

Diese Kompaktheit zählt beim Transport. Sie können die vollständige Pose mehrmals pro Sekunde über MQTT veröffentlichen, ohne das Netzwerk zu belasten, und jeder Subscriber – ob 3D-Viewer, Historian oder Analyse-Skript – erhält vom Broker denselben Datenstrom.

Architektur des digitalen Zwillings für den Roboterarm

Das Design nutzt drei Rollen, die über einen Broker verbunden sind:

1. Roboter-Gateway: liest Gelenkpositionen aus der Steuerung (oder simuliert sie) und veröffentlicht die Pose. Es abonniert außerdem Befehls-Topics und leitet akzeptierte Befehle an die Steuerung weiter.
2. 3D-Viewer: eine Browserseite, die die Pose über Secure WebSocket abonniert und jeden Gelenkwinkel auf den passenden Knoten eines 3D-Modells anwendet.
3. Bediener oder Anwendung: veröffentlicht Befehle wie „zu Pose fahren“ oder „Greifer öffnen“ und wartet auf Bestätigungen.

Die Gelenk-Telemetrie nutzt QoS 0, weil eine neuere Pose eine ältere immer ersetzt; der Verlust eines Frames ist harmlos. Befehle und Bestätigungen nutzen QoS 1, damit sie mindestens einmal zugestellt werden, und über die Befehls-id kann das Gateway Duplikate ignorieren. Eine Auffrischung zu dieser Abwägung bietet MQTT-QoS-Stufen erklärt.

Im Werk sitzt robot1 nicht isoliert, sondern in einem Unified Namespace (UNS): Die ISA-95-Hierarchie (Enterprise → Site → Area → Line → Cell) bildet die Topic-Struktur auf dem MQTT-Broker, etwa acme/werk-stuttgart/karosseriebau/linie-1/zelle-3/roboter-1/joints. So finden MES, SCADA und der 3D-Zwilling die Roboterdaten an derselben Stelle; mehr dazu unter Sparkplug B und Unified Namespace.

Für dieses Tutorial nutzen wir den öffentlichen MQTT-Broker unter broker.connectmq.com mit einem eindeutigen Topic-Präfix. Er ist für Tests und Prototypen gedacht, und der öffentliche Bereich ist für jeden lesbar; senden Sie dort also niemals echte Produktionsdaten.

Schritt 1: Gelenkwinkel vom Gateway veröffentlichen

Das folgende Skript simuliert einen Sechsachs-Arm, der sich sanft auf Zielwinkel zubewegt, seine Pose fünfmal pro Sekunde veröffentlicht, eine retained Zustandsnachricht vorhält und Bewegungsbefehle annimmt. Ersetzen Sie ra-7f3k durch Ihre eigene zufällige ID.

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  # QoS-1-Duplikat, bereits verarbeitet
    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 bleibt deutlich unter dem Ratenlimit des öffentlichen Brokers
except KeyboardInterrupt:
    client.publish(f"{PREFIX}/status", "offline", qos=1, retain=True)
    client.loop_stop()
    client.disconnect()

Installieren Sie die Bibliothek mit pip install "paho-mqtt>=2.0" und starten Sie das Skript. Beobachten Sie den Datenstrom in einem anderen Terminal:

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

In einer realen Zelle ersetzen Sie die Glättungsschleife durch Lesezugriffe auf die Steuerungsschnittstelle Ihres Roboterherstellers und leiten akzeptierte Befehle dorthin weiter. Behalten Sie die Bereichsprüfung im Gateway auf jeden Fall bei: Der Zwilling darf niemals derjenige sein, der eine Achse über ihre Grenze hinaus fährt.

Schritt 2: Den 3D-Zwilling des Roboterarms im Browser bauen

Ein Roboterarm in Three.js ist eine Kette verschachtelter Gruppen. Jedes Gelenk ist eine Gruppe, die sich um eine Achse dreht, und jedes Kind erbt die Transformationen seiner Eltern – genau so funktioniert Vorwärtskinematik. Wenn Sie statt Primitiven ein aus CAD exportiertes glTF-Modell laden, gilt dieselbe Idee: Finden Sie den Knoten jedes Gelenks über seinen Namen und drehen Sie ihn. Der Leitfaden zum Binden von IoT-Daten an ein 3D-Modell behandelt Namenskonventionen für diese Knoten.

Legen Sie ein Projekt mit npm install three mqtt und einem Bundler wie Vite an und verwenden Sie dann dieses Modul als Einstiegspunkt:

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

// Kinematische Kette: Jedes Gelenk ist eine Group, die in ihrem Elternteil verschachtelt ist.
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; // zwischen Frames glätten
  }
  renderer.render(scene, camera);
});

Die Interpolation in der Render-Schleife ist wichtig. Telemetrie trifft mit 5 Hz ein, der Bildschirm rendert aber mit 60 Hz; springt man direkt zu jedem neuen Winkel, wirkt das ruckelig. Das Gleiten zum jeweils neuesten Zielwert überbrückt die Lücke, ohne die reale Pose zu verfälschen. Eine ausführlichere Einführung in Szenen, Kameras und Loader bieten das Three.js-Digital-Twin-Tutorial und die offizielle Three.js-Dokumentation.

Schritt 3: Den Kreis mit Befehlen schließen

Ein Zwilling, der nur zuhört, ist ein digitaler Schatten. Um ihn bidirektional zu machen, veröffentlichen Sie einen Befehl und warten auf die passende Bestätigung:

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

Das Gateway prüft die Grenzen, aktualisiert seine Zielwerte und veröffentlicht {"id":"c-42","result":"accepted"} auf cmd/ack. Das 3D-Modell schwenkt dann in die neue Pose, weil es der Telemetrie folgt, nicht dem Befehl. Diese Unterscheidung ist die wichtigste Designregel für den digitalen Zwilling eines Roboterarms: Der Viewer rendert immer, was der Roboter meldet, nie, was jemand angefordert hat. Verweigert die Steuerung eine Bewegung, bleibt der Zwilling stehen, und der Bediener sieht den Ablehnungsgrund.

Ergänzen Sie in der Produktion diese Schutzmaßnahmen:

• Sichern Sie Befehle über steuerungsseitige Verriegelungen und das eigene Sicherheitssystem des Roboters ab. MQTT-Befehle sollten Bewegungen anfordern, nicht die Sicherheit umgehen.
• Versehen Sie Befehle mit einem Ablaufzeitpunkt, damit eine verzögerte Nachricht nicht verspätet ausgeführt wird. Mit MQTT 5 setzen Sie beim Publish das Message Expiry Interval; wie das geht, zeigt der Leitfaden zu MQTT-5-Funktionen.
• Verwenden Sie getrennte Zugangsdaten für Viewer und Bediener und protokollieren Sie jeden Befehl und jede Bestätigung.

Von der Demo zur Produktion

Der öffentliche Broker ist für Experimente gedacht: Er begrenzt Clients auf 10 Nachrichten pro Sekunde und 64 KB pro Nachricht, und alles, was Sie dort veröffentlichen, ist für andere sichtbar. Für eine echte Zelle wechseln Sie zu einem verwalteten MQTT-Broker mit privaten Zugangsdaten und TLS. Bei ConnectMQ erstellen Sie im Portal einen Broker, kopieren den generierten Benutzernamen, das Passwort und die Client ID und ändern zwei Dinge im Code: Verbindung per TLS auf Port 8883 (oder WSS auf 8084 für den Browser) und Übergabe der Zugangsdaten.

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()  # der System-CA-Store validiert das Let's-Encrypt-Zertifikat
client.connect("YOUR_BROKER_HOST", 8883, keepalive=30)

Da Client IDs bei ConnectMQ an ausgegebene Zugangsdaten gebunden sind, geben Sie dem Gateway und jedem Viewer eine eigene Identität, statt eine gemeinsam zu nutzen. Der Live-Topic-Explorer im Portal ist bei der Inbetriebnahme praktisch: Sie können den Gelenkdatenstrom und die Bestätigungen nebeneinander beobachten, ohne zusätzliche Tools zu schreiben.

Checkliste zur Fehlerbehebung

Betreiben Sie Ihren Roboterarm-Zwilling auf ConnectMQ

Sie haben jetzt einen funktionierenden digitalen Zwilling eines Roboterarms: ein Gateway, das Gelenkwinkel streamt, einen 3D-Viewer, der ihnen folgt, und einen Befehlspfad mit Bestätigungen. Entwickeln Sie den Prototyp auf broker.connectmq.com und übertragen Sie denselben Code auf einen privaten ConnectMQ-Broker, sobald die Zelle in Betrieb geht. Erstellen Sie einen kostenlosen Serverless-Broker, ohne Kreditkarte, und Sie haben in wenigen Minuten TLS-Endpunkte und Zugangsdaten.

Häufig gestellte Fragen

Wie oft sollte sich der digitale Zwilling eines Roboterarms aktualisieren?

Fünf bis zehn Pose-Updates pro Sekunde wirken in der Regel flüssig, wenn der Viewer zwischen den Frames interpoliert. Höhere Raten erzeugen Last ohne sichtbaren Nutzen für die meisten Monitoring-Anwendungsfälle.

Kann ich ein CAD-Modell statt Primitiven verwenden?

Ja. Exportieren Sie den Arm als glTF mit einem Knoten pro Gelenk, laden Sie ihn mit dem GLTFLoader und drehen Sie diese Knoten genau wie die Gruppen in diesem Tutorial. Achten Sie darauf, dass der Pivot jedes Knotens auf der Gelenkachse liegt.

Ist es sicher, einen echten Roboter über MQTT zu steuern?

MQTT kann Bewegungsanforderungen transportieren, die Sicherheit muss aber in der Robotersteuerung und ihrem Sicherheitssystem bleiben. Prüfen Sie Grenzen im Gateway, verwenden Sie ablaufende Befehle und authentifizierte Clients, und lassen Sie den Zwilling niemals Verriegelungen umgehen.

Warum folgt der Viewer der Telemetrie statt den Befehlen?

Befehle können abgelehnt, verzögert oder nur teilweise ausgeführt werden. Nur die gemeldete Pose darzustellen, hält den Zwilling wahrheitsgetreu – und genau darum geht es bei einem digitalen Zwilling.