Eclipse Tahu: Sparkplug in Java, Python und C

Eclipse Tahu ist die Open-Source-Referenzimplementierung der Sparkplug-Spezifikation, mit Bibliotheken und Beispielen in Java, Python und C. In diesem Leitfaden erfahren Sie, wofür die einzelnen Teile des Tahu-Repositorys da sind, wie Sie es bauen, an welchen Stellen man beim ersten Einsatz stolpert und wie Sie einen Tahu-basierten Edge Node mit einem echten MQTT-Broker verbinden und das Eintreffen seiner Births und Daten beobachten.

Was Eclipse Tahu ist und was nicht

Sparkplug wird bei der Eclipse Foundation verwaltet, und Tahu ist das Projekt, das die Spezifikation in funktionierenden Code übersetzt. Es liefert drei Dinge:

Die Protobuf-Definition für Sparkplug B Payloads, sparkplug_b.proto, sowie daraus generierten Code für jede Sprache.
Hilfsbibliotheken, die Payloads erzeugen und parsen: BIRTH- und DEATH-Payloads erstellen, Metriken mit den richtigen Datentypen hinzufügen, Sequenznummern verwalten.
Beispiele und Kompatibilitätswerkzeuge, die einen minimalen Edge Node und eine Host-Anwendung zeigen und Herstellern helfen, ihre Implementierungen gegen die Spezifikation zu prüfen.

Was Eclipse Tahu nicht ist: kein MQTT-Broker, kein SCADA-System und kein Gateway-Produkt. Jedes Tahu-Beispiel verbindet sich als gewöhnlicher MQTT-Client mit einem Broker, den Sie bereitstellen. Genau diese Trennung ist der Kern von Sparkplug. Edge Node und Host sprechen Sparkplug, und der MQTT-Broker dazwischen muss nur ein konformer MQTT-Server sein.

In Industrie-4.0-Architekturen ist dieser Broker meist das Herz eines Unified Namespace (UNS): einer zentralen Topic-Struktur, in der SPS, Edge-Gateways, SCADA, MES und ERP ihre aktuellen Daten teilen. Sparkplug liefert dafür die zuverlässige Zustandsverwaltung mit Births, Deaths und Sequenznummern, während die ISA-95-Hierarchie (Enterprise → Site → Area → Line → Cell) die fachliche Struktur des UNS vorgibt. Wie beides zusammenpasst, erklärt Sparkplug B und Unified Namespace.

Tahu ist außerdem Referenzcode. Er ist auf Klarheit und Korrektheit ausgelegt, nicht darauf, unverändert in ein produktives Gateway übernommen zu werden. Viele Teams verwenden die Protobuf-Definitionen und Payload-Helfer direkt und schreiben ihre eigene Session-Verwaltung darum herum, und genau das tun wir weiter unten.

Ein Rundgang durch das Repository

Der Quellcode liegt im Eclipse-Tahu-Repository auf GitHub. Klonen Sie es, um es zu erkunden:

git clone https://github.com/eclipse-tahu/tahu.git
cd tahu
ls

Die Struktur ist nach Sprachen gegliedert. Ordnernamen haben sich zwischen Releases verschoben. Betrachten Sie diese Tabelle daher als Orientierung, nicht als exakte Auflistung, und lesen Sie die README des Branches, den Sie klonen.

Bereich	Inhalt	Typische Verwendung
`sparkplug_b` (Proto)	Die `.proto`-Definition für Sparkplug B Payloads	Bindings für jede Sprache generieren
`java`	Core-Bibliothek, Payload-Builder und -Decoder, Edge- und Host-Beispiele, Kompatibilitätstests	JVM-Gateways und Host-Anwendungen
`python`	`core` mit den Helfern in `sparkplug_b.py` und dem generierten `sparkplug_b_pb2.py`, dazu Beispiele	Skripte, Simulatoren, Testwerkzeuge
`c`	Payload-Kodierung auf Basis von nanopb, dazu ein Beispiel-Client	Embedded-Gateways und ressourcenarme Geräte
`javascript`	Hilfen zur Payload-Kodierung für Node.js	Node-basierte Werkzeuge und Flows im Stil von Node-RED

Java

Die Java-Implementierung ist die vollständigste. Sie enthält Modellklassen für Payloads und Metriken, Encoder und Decoder sowie höherstufige Beispiele für Edge Nodes und Hosts. Gebaut wird mit Maven:

cd java
mvn clean install -DskipTests

Die resultierenden Artefakte können Sie in Ihrem eigenen Maven- oder Gradle-Projekt verwenden. Java ist die naheliegende Wahl, wenn Sie eine Host-Anwendung schreiben oder an ein JVM-basiertes SCADA- oder MES-System anbinden.

Python

Der Python-Ordner enthält zwei Dateien, die Sie ständig nutzen werden. sparkplug_b_pb2.py ist das generierte Protobuf-Modul, und sparkplug_b.py ergänzt Helfer wie getNodeBirthPayload, getDeviceBirthPayload, getDdataPayload, getNodeDeathPayload und addMetric sowie eine Enumeration MetricDataType. Der Beispielordner zeigt einen einfachen Edge Node, der Births veröffentlicht, Daten sendet und auf Befehle antwortet.

C

Die C-Implementierung kodiert und dekodiert Sparkplug-Payloads mit nanopb, einer kleinen Protobuf-Laufzeit für Mikrocontroller. Sie ist der Ausgangspunkt, wenn Sie Sparkplug auf einem Embedded-Linux-Gateway oder in Firmware mit knappem Speicher brauchen. Der Beispiel-Client kombiniert den Encoder mit einer MQTT-Bibliothek; Sie bauen ihn mit dem mitgelieferten Makefile, nachdem Sie die Entwicklungs-Header dieser Bibliothek installiert haben.

Typische Stolperfallen bei Eclipse Tahu

Die meisten Probleme am ersten Tag mit Tahu sind Umgebungsprobleme, keine Sparkplug-Probleme.

paho-mqtt 2.x bricht ältere Python-Beispiele. Die Tahu-Python-Beispiele wurden für paho-mqtt 1.x geschrieben. Mit paho-mqtt 2.x verlangt mqtt.Client() als erstes Argument eine Callback-API-Version. Pinnen Sie für die Beispiele entweder pip install "paho-mqtt<2" oder erstellen Sie den Client als mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, client_id=...), um die alten Callback-Signaturen beizubehalten.
Versionskonflikt der Protobuf-Laufzeit. Ein sparkplug_b_pb2.py, das mit einem alten protoc generiert wurde, kann mit einem neuen protobuf-Paket an „Descriptors cannot be created directly" scheitern. Generieren Sie es mit einem aktuellen Compiler über protoc --python_out=. sparkplug_b.proto neu.
Importpfade. Die Beispiele importieren die Core-Helfer über einen eingefügten relativen Pfad. Wenn Sie ein Beispiel an einen anderen Ort kopieren, kopieren Sie sparkplug_b.py und sparkplug_b_pb2.py mit.
Fest eingetragene Verbindungseinstellungen. Die Beispiele zeigen standardmäßig auf localhost. Passen Sie Broker-Host, Port, Group ID, Edge Node ID und Zugangsdaten am Anfang der Datei an.
Geteilte Group IDs auf öffentlichen Brokern. Sparkplug-Topics beginnen immer mit spBv1.0. Verwenden Sie auf einem öffentlichen Broker daher eine zufällige Group ID, um nicht mit den Tests anderer Leute zu kollidieren.

Einen Tahu-Edge-Node mit einem MQTT-Broker verbinden

Das folgende Skript nutzt die Python-Helfer von Tahu mit paho-mqtt 2.x. Kopieren Sie sparkplug_b.py und sparkplug_b_pb2.py aus dem Python-Core-Ordner von Tahu in ein leeres Verzeichnis, legen Sie diese Datei daneben und installieren Sie die Abhängigkeiten mit pip install "paho-mqtt>=2.0" protobuf.

import random
import time

import paho.mqtt.client as mqtt
from sparkplug_b import (
    MetricDataType,
    addMetric,
    getDdataPayload,
    getDeviceBirthPayload,
    getNodeBirthPayload,
    getNodeDeathPayload,
)

BROKER = "broker.connectmq.com"
PORT = 1883
GROUP = "tahu-demo-5mz8"       # zufällige Group ID für den öffentlichen Broker
NODE = "tahu-edge-01"
DEVICE = "pump-01"
CLIENT_ID = f"tahu-edge-{random.randint(1000, 9999)}"

def t(msg_type, device=None):
    base = f"spBv1.0/{GROUP}/{msg_type}/{NODE}"
    return f"{base}/{device}" if device else base

def publish_births(client):
    nbirth = getNodeBirthPayload()
    addMetric(nbirth, "Node Control/Rebirth", None, MetricDataType.Boolean, False)
    client.publish(t("NBIRTH"), bytearray(nbirth.SerializeToString()), 0, False)

    dbirth = getDeviceBirthPayload()
    addMetric(dbirth, "Flow", None, MetricDataType.Float, 12.5)
    addMetric(dbirth, "Pressure", None, MetricDataType.Float, 3.1)
    addMetric(dbirth, "Running", None, MetricDataType.Boolean, True)
    client.publish(t("DBIRTH", DEVICE), bytearray(dbirth.SerializeToString()), 0, False)
    print("NBIRTH and DBIRTH published")

def on_connect(client, userdata, flags, reason_code, properties):
    print("connected:", reason_code)
    client.subscribe(t("NCMD"))
    publish_births(client)

def on_message(client, userdata, msg):
    print("command received on", msg.topic)
    publish_births(client)   # in dieser Demo wird jede NCMD als Rebirth-Anforderung behandelt

# Der Death-Payload muss vor dem Birth erstellt werden, damit beide denselben bdSeq tragen
death = getNodeDeathPayload()

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
client.on_connect = on_connect
client.on_message = on_message
client.will_set(t("NDEATH"), bytearray(death.SerializeToString()), 1, False)
client.connect(BROKER, PORT, 60)
client.loop_start()

try:
    while True:
        time.sleep(3)
        ddata = getDdataPayload()
        addMetric(ddata, "Flow", None, MetricDataType.Float, round(random.uniform(11, 14), 2))
        addMetric(ddata, "Pressure", None, MetricDataType.Float, round(random.uniform(2.8, 3.4), 2))
        client.publish(t("DDATA", DEVICE), bytearray(ddata.SerializeToString()), 0, False)
        print("DDATA published")
except KeyboardInterrupt:
    client.loop_stop()
    client.disconnect()

Die Reihenfolge von getNodeDeathPayload() und getNodeBirthPayload() ist wichtig, weil die Tahu-Helfer den bdSeq-Zähler intern verwalten und das als Will registrierte NDEATH denselben Wert tragen muss wie das folgende NBIRTH. Verhält sich Ihre Kopie der Helfer anders, prüfen Sie die bdSeq-Metrik in beiden Payloads und passen Sie das Skript an.

Starten Sie das Skript und abonnieren Sie die Gruppe in einem anderen Terminal:

mosquitto_sub -h broker.connectmq.com -p 1883 -i tahu-watch-5mz8 -t "spBv1.0/tahu-demo-5mz8/#" -v

Sie sehen die Topics mit binären Payloads eintreffen. Um die Metriken zu lesen, öffnen Sie den Verkehr in einem ConnectMQ-Portal: Der Live-Topic-Explorer dekodiert Sparkplug B Payloads, ob Protobuf oder JSON, in benannte Metriken mit Datentypen und Werten. So bestätigen Sie schnell, dass Tahu das Erwartete erzeugt hat, bevor ein SCADA-Host ins Spiel kommt. Der eben verwendete Broker ist der kostenlose öffentliche MQTT-Broker; er ist für Tests und Prototypen gedacht und für alle lesbar, halten Sie echte Tag-Namen und Werte also davon fern.

Vom Referenzcode zum echten Edge Node

Sobald das Beispiel funktioniert, braucht ein produktiver Edge Node mehr, als Tahu von Haus aus mitbringt:

Persistenter bdSeq. Speichern Sie den Zähler auf der Festplatte, damit er über Prozessneustarts hinweg hochzählt.
Report by Exception. Veröffentlichen Sie nur geänderte Werte, mit Totbändern für verrauschte analoge Signale.
Store and Forward. Puffern Sie Daten bei Broker-Ausfällen und spielen Sie sie mit Historical-Kennzeichnung erneut ab.
Aliase. Aktivieren Sie auf eingeschränkten Verbindungen Sparkplug Metrik-Aliase.
Primary-Host-Erkennung. In Sparkplug 3.0 können Edge Nodes auf die STATE-Nachricht des Primary Hosts warten, bevor sie Births veröffentlichen.
TLS und Zugangsdaten. Wechseln Sie auf Port 8883 mit Zugangsdaten pro Client.

Wenn Sie ein vollständig eigenständiges Skript ohne die Tahu-Helfer möchten, erstellt unser Sparkplug B Edge-Node-Simulator Payloads direkt aus dem generierten Protobuf-Modul und behandelt Sequenznummern und Rebirths explizit. Zu den Interna der Payloads siehe Sparkplug B Datentypen. Die Spezifikation selbst veröffentlicht die Eclipse Sparkplug Working Group.

Im Produktivbetrieb verbinden Sie Tahu-basierte Nodes mit einem privaten Broker. Bei ConnectMQ erstellen Sie im Portal einen Broker, kopieren die generierten Werte für YOUR_USERNAME, YOUR_PASSWORD und YOUR_CLIENT_ID, setzen die Client ID im Skript, ergänzen client.username_pw_set(...) und client.tls_set() und verbinden sich mit Port 8883. ConnectMQ kann die dekodierten Sparkplug-B-Daten dann in ISA-95-Topics wie connectmq/factory1/production/line1/cnc1/telemetry erneut veröffentlichen, die Konsumenten ohne Sparkplug-Unterstützung als einfache Daten lesen können. Damit wird aus der flachen Sparkplug-Struktur (Group, Edge Node, Device) ein UNS-Zweig entlang der ISA-95-Ebenen, den auch MES, Historian oder ein BI-Werkzeug direkt abonnieren können. Die übergreifende Architektur behandelt unser Leitfaden zu Industrial IoT mit MQTT.

Eclipse Tahu mit ConnectMQ verbinden

Tahu liefert Ihnen korrekte Payloads; zwischen Edge und Host brauchen Sie trotzdem einen verlässlichen Broker. ConnectMQ ist ein verwalteter MQTT-Broker mit MQTT 3.1.1 und 5.0, TLS auf 8883, Sparkplug-B-Dekodierung und ISA-95-Republishing. Erstellen Sie unter app.connectmq.com einen Broker im kostenlosen Serverless-Plan, ganz ohne Kreditkarte, und richten Sie Ihren Tahu-Edge-Node in wenigen Minuten darauf aus.

Häufig gestellte Fragen

Ist Eclipse Tahu ein Broker?

Nein. Eclipse Tahu stellt Sparkplug-Bibliotheken und Beispiele bereit. Damit gebaute Edge Nodes und Hosts verbinden sich als MQTT-Clients mit einem beliebigen konformen MQTT-Broker.

Welche Tahu-Sprache sollte ich verwenden?

Java für Host-Anwendungen und JVM-Gateways, Python für Skripte, Simulatoren und Tests und C für Embedded-Gateways mit knappem Speicherbudget.

Warum scheitern die Tahu-Python-Beispiele mit paho-mqtt 2.x?

paho-mqtt 2.x verlangt beim Erstellen des Clients eine Callback-API-Version. Pinnen Sie paho-mqtt für die Beispiele unter 2.0 oder übergeben Sie mqtt.CallbackAPIVersion.VERSION1, um deren Callback-Signaturen beizubehalten.

Ist Tahu produktionsreif?

Es ist Referenzcode mit Fokus auf Korrektheit. Viele Teams nutzen die Protobuf-Definitionen und Payload-Helfer im Produktivbetrieb und ergänzen eigene Session-Verwaltung, Pufferung und Sicherheit.

Eclipse Tahu: Sparkplug-Referenzimplementierungen in Java, Python und C