Sparkplug B mit Python dekodieren: Protobuf-Payloads Schritt für Schritt lesen

Um Sparkplug B in Python zu dekodieren, brauchen Sie zwei Dinge: Klassen, die aus dem offiziellen Schema sparkplug_b.proto generiert wurden, und einen MQTT-Subscriber, der sich merkt, was jeder Edge Node in seinen Birth-Nachrichten angekündigt hat. In diesem Tutorial bauen Sie beides, erhalten am Ende ein Skript, das jede Sparkplug-Nachricht als lesbares JSON ausgibt, und lernen die drei Details kennen, die die meisten schnellen Decoder falsch machen.

Was Sie brauchen

Falls noch kein Edge Node publiziert, starten Sie den Simulator aus unserem Artikel über Sparkplug Birth- und Death-Zertifikate. Er veröffentlicht NBIRTH, DBIRTH und DDATA in eine zufällige Gruppe auf dem öffentlichen Broker, genau das, was dieser Decoder braucht.

Schritt 1: Die Python-Klassen generieren

Legen Sie einen Projektordner und eine virtuelle Umgebung an und generieren Sie dann sparkplug_b_pb2.py aus dem Schema, das das Eclipse-Tahu-Projekt veröffentlicht:

mkdir sparkplug-decoder && cd sparkplug-decoder
python -m venv .venv
source .venv/bin/activate        # unter Windows: .venv\Scripts\activate
pip install "paho-mqtt>=2.0" protobuf grpcio-tools
curl -O https://raw.githubusercontent.com/eclipse/tahu/master/sparkplug_b/sparkplug_b.proto
python -m grpc_tools.protoc -I. --python_out=. sparkplug_b.proto

Mit grpcio-tools vermeiden Sie Versionskonflikte zwischen einem separat installierten protoc und der Python-Protobuf-Laufzeit. Erscheint später ein Fehler, der generierte Code stamme von einem neueren protoc, aktualisieren Sie protobuf in derselben Umgebung und generieren Sie neu.

Schritt 2: Verstehen, was der Decoder leisten muss

ParseFromString auf die Bytes anzuwenden ist der einfache Teil. Ein brauchbarer Decoder muss außerdem drei Sparkplug-spezifische Details beherrschen:

1. Aliase. Edge Nodes senden vollständige Metriknamen oft nur in NBIRTH und DBIRTH und verweisen in NDATA und DDATA über einen numerischen Alias auf Metriken. Der Decoder muss die Alias-Zuordnung pro Node und Device speichern.
2. Datentypen nur in Births. Der Datentyp ist in Birth-Nachrichten Pflicht, in Datennachrichten aber optional. Der Decoder sollte sich daher auch den Datentyp jeder Metrik aus dem Birth merken.
3. Vorzeichenbehaftete Ganzzahlen. Int8, Int16 und Int32 werden in einem vorzeichenlosen 32-Bit-Feld übertragen, Int64 in einem vorzeichenlosen 64-Bit-Feld. Ein Wert von -5 kommt als 4294967291 an, wenn Sie ihn nicht per Zweierkomplement umrechnen.

Der Leitfaden zum Sparkplug B Payload erklärt die Struktur hinter diesen Regeln ausführlicher.

Schritt 3: Der vollständige Decoder

Speichern Sie dies als decode_sparkplug.py:

import json
import sys
import uuid

import paho.mqtt.client as mqtt
from google.protobuf.json_format import MessageToDict
from google.protobuf.message import DecodeError

import sparkplug_b_pb2 as spb

BROKER = "broker.connectmq.com"
PORT = 1883
GROUP = sys.argv[1] if len(sys.argv) > 1 else "+"

DATATYPES = {
    1: "Int8", 2: "Int16", 3: "Int32", 4: "Int64",
    5: "UInt8", 6: "UInt16", 7: "UInt32", 8: "UInt64",
    9: "Float", 10: "Double", 11: "Boolean", 12: "String",
    13: "DateTime", 14: "Text", 15: "UUID", 16: "DataSet",
    17: "Bytes", 18: "File", 19: "Template",
}

# (group, node, device) -> {alias: (name, datatype)}
known_metrics = {}


def to_signed(value, bits):
    if value >= 1 << (bits - 1):
        value -= 1 << bits
    return value


def convert(value, datatype, kind):
    if kind == "int_value" and datatype in (1, 2, 3):
        value = to_signed(value & 0xFFFFFFFF, 32)
        if datatype == 1 and value > 127:
            value = to_signed(value, 8)
        if datatype == 2 and value > 32767:
            value = to_signed(value, 16)
    elif kind == "long_value" and datatype == 4:
        value = to_signed(value, 64)
    return value


def metric_to_dict(metric, key):
    table = known_metrics.setdefault(key, {})
    name, datatype = metric.name, metric.datatype
    if metric.HasField("alias"):
        if name:
            table[metric.alias] = (name, datatype)
        elif metric.alias in table:
            name, stored_type = table[metric.alias]
            datatype = datatype or stored_type
        else:
            name = f"<unknown alias {metric.alias}>"

    kind = metric.WhichOneof("value")
    if metric.is_null or kind is None:
        value = None
    elif kind in ("dataset_value", "template_value", "extension_value"):
        value = MessageToDict(getattr(metric, kind))
    elif kind == "bytes_value":
        value = metric.bytes_value.hex()
    else:
        value = convert(getattr(metric, kind), datatype, kind)

    result = {"name": name, "type": DATATYPES.get(datatype, datatype), "value": value}
    if metric.HasField("timestamp"):
        result["timestamp"] = metric.timestamp
    if metric.is_historical:
        result["historical"] = True
    return result


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print("connect failed:", reason_code)
        return
    topic = f"spBv1.0/{GROUP}/#"
    client.subscribe(topic, qos=0)
    print("subscribed to", topic)


def on_message(client, userdata, msg):
    parts = msg.topic.split("/")
    if len(parts) >= 2 and parts[1] == "STATE":
        print(json.dumps({"topic": msg.topic, "state": msg.payload.decode(errors="replace")}))
        return
    if len(parts) < 4:
        return
    group, msg_type, node = parts[1], parts[2], parts[3]
    device = parts[4] if len(parts) > 4 else None

    payload = spb.Payload()
    try:
        payload.ParseFromString(msg.payload)
    except DecodeError:
        print(f"{msg.topic}: not a valid Sparkplug B payload")
        return

    if msg_type == "NBIRTH":
        for key in [k for k in known_metrics if k[:2] == (group, node)]:
            del known_metrics[key]
    elif msg_type == "DBIRTH":
        known_metrics.pop((group, node, device), None)

    key = (group, node, device)
    decoded = {
        "topic": msg.topic,
        "type": msg_type,
        "timestamp": payload.timestamp,
        "seq": payload.seq if payload.HasField("seq") else None,
        "metrics": [metric_to_dict(m, key) for m in payload.metrics],
    }
    print(json.dumps(decoded, indent=2))


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"sp-decoder-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, keepalive=60)
client.loop_forever()

Starten Sie es mit der Group ID, die Ihr Simulator ausgegeben hat, zum Beispiel python decode_sparkplug.py demo-3f9a1c. Eine Gruppe anzugeben ist auf dem öffentlichen MQTT-Broker eine Frage der Rücksicht: Der öffentliche MQTT-Broker ist geteilt, für alle lesbar und nur für Tests gedacht, und wer dort alle Gruppen abonniert, erhält das Rauschen anderer Nutzer.

Schritt 4: Die Ausgabe lesen

Eine DDATA-Nachricht aus dem Simulator wird so ausgegeben:

{
  "topic": "spBv1.0/demo-3f9a1c/DDATA/line1/cnc1",
  "type": "DDATA",
  "timestamp": 1792137600123,
  "seq": 7,
  "metrics": [
    { "name": "Spindle/Speed", "type": "Double", "value": 1462.3, "timestamp": 1792137600121 },
    { "name": "Spindle/Running", "type": "Boolean", "value": true, "timestamp": 1792137600121 }
  ]
}

Starten Sie den Decoder nach dem Edge Node, erscheinen Metriken mit Alias als <unknown alias N>, weil der Decoder das Birth verpasst hat. Die Lösung ist ein Rebirth-Befehl an den Node, der daraufhin NBIRTH und DBIRTH erneut veröffentlicht. Produktive Host-Anwendungen erledigen das automatisch; ein einfaches Werkzeug startet man einfach zuerst.

Schritt 5: Automatisch ein Rebirth anfordern

Ein Decoder, der passiv dasitzt und unbekannte Aliase ausgibt, reicht für einen kurzen Blick. Ein Decoder, der eine Datenbank befüllt, sollte seine Sicht selbst reparieren. Der Standardweg ist eine NCMD-Nachricht, in der die Metrik Node Control/Rebirth auf true gesetzt ist. Fügen Sie diese Hilfsfunktion hinzu und rufen Sie sie aus metric_to_dict auf, wenn ein Alias fehlt:

requested = set()


def request_rebirth(client, group, node):
    if (group, node) in requested:
        return
    requested.add((group, node))
    cmd = spb.Payload()
    m = cmd.metrics.add()
    m.name = "Node Control/Rebirth"
    m.datatype = 11  # Boolean
    m.boolean_value = True
    client.publish(f"spBv1.0/{group}/NCMD/{node}", cmd.SerializeToString(), qos=0)
    print(f"rebirth requested from {group}/{node}")

Übergeben Sie den Client an metric_to_dict (oder halten Sie ihn als Modulvariable) und löschen Sie den Eintrag in requested, sobald das NBIRTH eintrifft, damit eine spätere Lücke eine neue Anforderung auslösen kann. Seien Sie damit in geteilten Umgebungen vorsichtig: Befehle zu senden ist ein Schreibvorgang, und in einer Produktionsanlage sollte normalerweise nur die dafür vorgesehene Host-Anwendung Rebirths auslösen. Der Artikel zum Sparkplug Topic-Namespace listet auf, welche Topics Befehle transportieren.

Tipps für einen dauerhaft laufenden Decoder

Ein Decoder, der wochenlang läuft, hat andere Anforderungen als einer, den Sie für fünf Minuten starten:

• Prüfen Sie seq pro Edge Node. Speichern Sie den letzten Wert und markieren Sie jede Lücke außer dem Sprung von 255 auf 0; eine Lücke bedeutet, dass Ihr Alias- und Wertecache veraltet sein kann.
• Verfolgen Sie bdSeq aus dem NBIRTH und werten Sie ein NDEATH nur dann als gültig, wenn sein bdSeq übereinstimmt. Andernfalls kann ein verspäteter Will aus einer alten Session einen aktiven Node als offline markieren.
• Markieren Sie bei einem NDEATH alle Metriken eines Nodes und bei einem DDEATH alle Metriken eines Devices als veraltet, statt den letzten Wert so zu behandeln, als wäre er aktuell.
• Halten Sie on_message schnell. Parsen Sie und übergeben Sie das Ergebnis an eine Queue oder einen Worker-Thread, wenn Sie in eine Datenbank schreiben, damit die Netzwerkschleife nie blockiert.
• Protokollieren Sie Dekodierfehler mit Topic und Payload-Länge, nicht mit dem vollständigen Payload, damit die Logs klein bleiben.

Schritt 6: Mit einem Produktiv-Broker verbinden

Der öffentliche MQTT-Broker dient nur für Tests und Prototypen. Für echte Anlagendaten verwenden Sie Zugangsdaten Ihres eigenen Brokers und TLS. Ersetzen Sie die Verbindungszeilen am Ende des Skripts:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # nutzt den CA-Speicher des Systems; der Broker verwendet ein Let's-Encrypt-Zertifikat
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 8883, keepalive=60)
client.loop_forever()

Bei ConnectMQ sind Client IDs an die im Portal ausgestellten Zugangsdaten gebunden. Verwenden Sie also genau die dort angezeigte Client ID.

Fehlerbehebung

Dekodieren mit Python oder direkt im Broker

Ein eigener Decoder ist das richtige Werkzeug, wenn Daten in eine Datenbank, einen Historian, ein MES oder eine eigene Anwendung fließen sollen, etwa für OEE-Kennzahlen oder Energiemonitoring. Für die tägliche Kontrolle ist es jedoch umständlich, Code zu schreiben, nur um einen Wert zu sehen. ConnectMQ dekodiert Sparkplug B (Protobuf und JSON) im Broker, sodass der Topic-Explorer im Portal Metriknamen und Werte direkt anzeigt. Außerdem kann ConnectMQ Sparkplug-Daten in lesbare ISA-95-Topics erneut veröffentlichen, etwa von spBv1.0/factory1/DDATA/line1/cnc1 nach connectmq/factory1/production/line1/cnc1/telemetry. Ein nachgelagertes Python-Skript, ein Node-RED-Flow oder ein Telegraf-Agent kann diese Topics dann als gewöhnlicher MQTT-Client abonnieren, ohne die Proto-Datei mitzuliefern.

Damit wird der Decoder zum Baustein eines Unified Namespace (UNS): Die ISA-95-Equipment-Hierarchie (Enterprise → Site → Area → Line → Cell) bildet die Topic-Struktur auf dem MQTT-Broker, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur, und MES, ERP oder Dashboards abonnieren lesbare Werte statt Protobuf. Wie Sparkplug B und UNS zusammenspielen, beschreibt der Artikel Sparkplug B und Unified Namespace.

Eine bewährte Aufteilung vieler Teams: integrierte Dekodierung für Inbetriebnahme und Fehlersuche, ein Python-Konsument wie der obige, wenn Daten gespeichert oder transformiert werden müssen. Für Details zu Datentypen halten Sie die Referenz zu den Sparkplug B Datentypen bereit, für die Gesamtarchitektur lesen Sie den Leitfaden zu Industrial IoT mit MQTT.

Starten Sie mit ConnectMQ

Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker, ganz ohne Kreditkarte, und verbinden Sie diesen Decoder mit Ihren generierten Zugangsdaten. Wenn Sie über Tests hinauswachsen, nimmt Ihnen ein verwalteter MQTT-Broker oder ein Cloud-MQTT-Broker den Betriebsaufwand eines eigenen Brokers ab.

Häufig gestellte Fragen

Gibt es ein pip-Paket zum Dekodieren von Sparkplug B?

Es gibt mehrere Community-Pakete, doch Klassen aus dem offiziellen Tahu-Proto zu generieren ist der zuverlässigste Weg, weil Sie die Schemaversion selbst kontrollieren. Mit grpcio-tools genügt dafür ein einziger Befehl.

Kann ich Sparkplug B ohne die Proto-Datei dekodieren?

Nicht auf sinnvolle Weise. Protobuf-Nachrichten enthalten Feldnummern statt Namen, daher brauchen Sie das Schema, um zu wissen, was jedes Feld bedeutet.

Warum zeigt mein Decoder unbekannte Aliase an?

Der Edge Node sendet Metriknamen nur in seinen Birth-Nachrichten. Startet Ihr Decoder später, hat er die Alias-Zuordnung nie erhalten; fordern Sie ein Rebirth an oder starten Sie den Decoder zuerst.

Funktioniert dieser Decoder mit MQTT 5?

Ja. Sparkplug-Payloads sind unter MQTT 3.1.1 und 5.0 identisch. Um MQTT 5 zu nutzen, übergeben Sie beim Erstellen des Clients protocol=mqtt.MQTTv5.