Einen Sparkplug B Edge-Node-Simulator in Python bauen

Mit einem Sparkplug B Edge-Node-Simulator entwickeln und testen Sie SCADA-Hosts, Historians und Dashboards, ohne auf eine SPS, ein Gateway oder ein Wartungsfenster warten zu müssen. In diesem Tutorial bauen Sie einen in Python, der eine korrekte Abfolge aus NBIRTH, DBIRTH und DDATA veröffentlicht, einen NDEATH-Will registriert und auf Rebirth-Anforderungen antwortet, und zwar gegen einen echten MQTT Broker, den Sie in wenigen Minuten erreichen.

Was der Simulator richtig machen muss

Ein Sparkplug-Konsument ist strenger als ein einfacher MQTT-Subscriber. Eine Host-Anwendung verfolgt den Session-Zustand pro Edge Node und verwirft oder ignoriert Daten, die gegen die Regeln verstoßen. Ein brauchbarer Sparkplug B Edge-Node-Simulator muss daher das Protokollverhalten nachbilden, nicht nur die Topic-Namen.

Sind Ihnen einige dieser Konzepte neu, lesen Sie zuerst Sparkplug Birth- und Death-Zertifikate und den Sparkplug Topic-Namespace. Der Simulator unten folgt beidem.

Voraussetzungen

Sie benötigen Python 3.9 oder neuer und zwei Pakete:

python -m venv .venv
source .venv/bin/activate        # unter Windows: .venv\Scripts\activate
pip install "paho-mqtt>=2.0" protobuf

Außerdem brauchen Sie sparkplug_b_pb2.py, das Python-Modul, das aus der offiziellen Sparkplug-B-Protobuf-Definition generiert wird. Das Eclipse-Tahu-Projekt liefert eine fertige Kopie in seinem Python-Core-Ordner; laden Sie sie in dasselbe Verzeichnis wie Ihr Skript herunter. Unser Leitfaden zu Eclipse Tahu erklärt, wo sie liegt und wie Sie sie mit protoc neu generieren, falls Ihre protobuf-Version Probleme meldet.

Als Broker beginnen Sie mit dem öffentlichen MQTT-Broker unter broker.connectmq.com auf Port 1883. Er akzeptiert anonyme Verbindungen mit jeder nicht leeren, eindeutigen Client ID. Sparkplug-Topics müssen mit spBv1.0 beginnen, Sie können also kein privates Präfix voranstellen. Verwenden Sie stattdessen eine zufällige Group ID wie demo-x91q, damit Ihr Verkehr nicht mit dem anderer Nutzer kollidiert. Der öffentliche Broker ist für alle lesbar und auf 10 Nachrichten pro Sekunde und Client begrenzt, was für einen Simulator, der alle paar Sekunden veröffentlicht, mehr als genug ist.

Der Code des Sparkplug B Edge-Node-Simulators

Speichern Sie dies als edge_sim.py. Es simuliert einen Edge Node mit einem Device, einer Abfüllmaschine mit Geschwindigkeit, Temperatur, Laufstatus und Stückzahl.

import os
import random
import ssl
import time

import paho.mqtt.client as mqtt
import sparkplug_b_pb2

# --- configuration -----------------------------------------------------------
BROKER = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
USERNAME = os.getenv("MQTT_USERNAME")          # set for a private ConnectMQ broker
PASSWORD = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID", f"spb-sim-{random.randint(1000, 9999)}")
GROUP = os.getenv("SPB_GROUP", "demo-x91q")
NODE = os.getenv("SPB_NODE", "edge-sim-01")
DEVICE = os.getenv("SPB_DEVICE", "filler-01")
INTERVAL = float(os.getenv("SIM_INTERVAL", "2"))

# Sparkplug B datatype codes
INT64, FLOAT, BOOLEAN = 4, 9, 11

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

def now_ms():
    return int(time.time() * 1000)

class EdgeNode:
    def __init__(self):
        self.bd_seq = 0
        self.seq = 0
        self.state = {"Speed": 120.0, "Temperature": 41.5, "Running": True, "Count": 0}
        self.aliases = {"Speed": 1, "Temperature": 2, "Running": 3, "Count": 4}

    def next_seq(self):
        value = self.seq
        self.seq = (self.seq + 1) % 256
        return value

    @staticmethod
    def add_metric(payload, name, datatype, value, alias=None):
        m = payload.metrics.add()
        if name:
            m.name = name
        if alias is not None:
            m.alias = alias
        m.timestamp = now_ms()
        m.datatype = datatype
        if datatype == INT64:
            m.long_value = int(value)
        elif datatype == FLOAT:
            m.float_value = float(value)
        elif datatype == BOOLEAN:
            m.boolean_value = bool(value)
        return m

    def death_payload(self):
        p = sparkplug_b_pb2.Payload()
        p.timestamp = now_ms()
        self.add_metric(p, "bdSeq", INT64, self.bd_seq)
        return p.SerializeToString()

    def nbirth(self):
        self.seq = 0
        p = sparkplug_b_pb2.Payload()
        p.timestamp = now_ms()
        p.seq = self.next_seq()
        self.add_metric(p, "bdSeq", INT64, self.bd_seq)
        self.add_metric(p, "Node Control/Rebirth", BOOLEAN, False)
        return p.SerializeToString()

    def dbirth(self):
        p = sparkplug_b_pb2.Payload()
        p.timestamp = now_ms()
        p.seq = self.next_seq()
        types = {"Speed": FLOAT, "Temperature": FLOAT, "Running": BOOLEAN, "Count": INT64}
        for name, value in self.state.items():
            self.add_metric(p, name, types[name], value, alias=self.aliases[name])
        return p.SerializeToString()

    def simulate(self):
        """Change values and return only the metrics that changed."""
        changed = {}
        if random.random() < 0.05:
            self.state["Running"] = not self.state["Running"]
            changed["Running"] = (BOOLEAN, self.state["Running"])
        if self.state["Running"]:
            self.state["Speed"] = round(max(0, self.state["Speed"] + random.uniform(-3, 3)), 1)
            self.state["Count"] += random.randint(3, 5)
            changed["Speed"] = (FLOAT, self.state["Speed"])
            changed["Count"] = (INT64, self.state["Count"])
        new_temp = round(self.state["Temperature"] + random.uniform(-0.4, 0.4), 1)
        if new_temp != self.state["Temperature"]:
            self.state["Temperature"] = new_temp
            changed["Temperature"] = (FLOAT, new_temp)
        return changed

    def ddata(self, changed):
        p = sparkplug_b_pb2.Payload()
        p.timestamp = now_ms()
        p.seq = self.next_seq()
        for name, (datatype, value) in changed.items():
            self.add_metric(p, None, datatype, value, alias=self.aliases[name])
        return p.SerializeToString()

node = EdgeNode()

def publish_births(client):
    client.publish(topic("NBIRTH"), node.nbirth(), qos=0, retain=False)
    client.publish(topic("DBIRTH", DEVICE), node.dbirth(), qos=0, retain=False)
    print(f"births published (bdSeq={node.bd_seq})")

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print("connect failed:", reason_code)
        return
    client.subscribe(topic("NCMD"), qos=1)
    client.subscribe(topic("DCMD", DEVICE), qos=1)
    publish_births(client)

def on_message(client, userdata, msg):
    p = sparkplug_b_pb2.Payload()
    p.ParseFromString(msg.payload)
    for m in p.metrics:
        if m.name == "Node Control/Rebirth" and m.boolean_value:
            print("rebirth requested")
            publish_births(client)

def on_disconnect(client, userdata, flags, reason_code, properties):
    print("disconnected:", reason_code)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID, clean_session=True)
if USERNAME:
    client.username_pw_set(USERNAME, PASSWORD)
if PORT == 8883:
    client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect
client.will_set(topic("NDEATH"), node.death_payload(), qos=1, retain=False)
client.connect(BROKER, PORT, keepalive=30)
client.loop_start()

try:
    while True:
        time.sleep(INTERVAL)
        if client.is_connected():
            changed = node.simulate()
            if changed:
                client.publish(topic("DDATA", DEVICE), node.ddata(changed), qos=0)
                print("DDATA", {k: v[1] for k, v in changed.items()})
except KeyboardInterrupt:
    # A clean shutdown publishes NDEATH explicitly, because the will is not sent
    client.publish(topic("NDEATH"), node.death_payload(), qos=1).wait_for_publish(5)
    client.loop_stop()
    client.disconnect()

Starten Sie ihn mit python edge_sim.py. Sie sollten eine Zeile zu den Births sehen und danach alle zwei Sekunden eine DDATA-Zeile.

Die Ausgabe beobachten

Prüfen Sie in einem zweiten Terminal, ob Nachrichten ankommen. Die Payloads sind binäres Protobuf, mosquitto_sub gibt also unlesbare Bytes aus, doch Topics und Timing reichen für eine erste Kontrolle:

mosquitto_sub -h broker.connectmq.com -p 1883 -i spb-watch-x91q -t "spBv1.0/demo-x91q/#" -v

Für dekodierte Werte nutzen Sie den aliasfähigen Decoder aus unserem Leitfaden zu Sparkplug Metrik-Aliasen oder öffnen den Verkehr im Portal eines ConnectMQ-Brokers, dessen Live-Topic-Explorer Sparkplug B Payloads in lesbare Metriken dekodiert. Um das Death-Zertifikat zu testen, beenden Sie den Simulator-Prozess mit kill -9 oder schließen das Terminalfenster, statt Strg+C zu drücken. Nach Ablauf des Keepalive-Fensters veröffentlicht der Broker den NDEATH-Will, und eine Host-Anwendung sollte jede Metrik als veraltet markieren.

Hosts und Dashboards mit dem Simulator testen

Der Sinn eines Simulators ist es, die Konsumentenseite auf die Probe zu stellen. Gehen Sie diese Szenarien durch, bevor Sie einem Host oder Dashboard eine echte Anlage anvertrauen.

1. Kaltstart. Starten Sie zuerst den Host, dann den Simulator. Der Host sollte Node und Device anlegen und alle vier Metriken anzeigen.
2. Später Beitritt. Starten Sie zuerst den Simulator und eine Minute später den Host. Ein korrekter Primary Host sendet ein NCMD-Rebirth, weil er die Births nie gesehen hat; der Simulator gibt „rebirth requested" aus.
3. Unsauberes Ende. Beenden Sie den Prozess hart. Der Host sollte den Node als offline kennzeichnen, sobald das NDEATH eintrifft.
4. Neuverbindung. Starten Sie den Simulator neu. Erhöhen Sie bd_seq in einem dauerhaft laufenden Einsatz zwischen Sessions, damit der Host jedes Death dem richtigen Birth zuordnen kann; das einfache Skript setzt den Wert bei jedem Start zurück, was zum Testen akzeptabel ist.
5. Skalierung. Starten Sie mehrere Kopien mit unterschiedlichen SPB_NODE-Werten, um zu sehen, wie der Host mit vielen Edge Nodes umgeht. Halten Sie auf dem öffentlichen Broker die gesamte Nachrichtenrate moderat; auf einem privaten Broker können Sie innerhalb der Limits Ihres Plans mehr Last erzeugen.
6. Befehlspfad. Senden Sie vom Host eine DCMD und prüfen Sie, ob sie on_message erreicht. Erweitern Sie den Handler, damit er Running auf Befehl ändert.

Wenn Sie einen vollständigen Host evaluieren, erklärt der Artikel zur Sparkplug Primary Host Application die STATE-Nachrichten, die ein Host veröffentlicht, und warum Edge Nodes in Sparkplug 3.0 so konfiguriert werden können, dass sie vor dem Senden von Births darauf warten.

Vom öffentlichen Broker zum eigenen wechseln

Der öffentliche MQTT-Broker ist für Tests und Prototypen gedacht; jeder kann Ihre Gruppe abonnieren. Für alles, was echten Maschinendaten ähnelt, verwenden Sie einen privaten Broker. Bei ConnectMQ registrieren Sie sich, wählen einen Plan, erstellen einen Broker und kopieren den generierten Benutzernamen, das Passwort und die Client ID. Dann führen Sie aus:

export MQTT_HOST=YOUR_BROKER_HOST
export MQTT_PORT=8883
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
python edge_sim.py

Port 8883 aktiviert TLS mit dem Let's-Encrypt-Zertifikat des Brokers, dem Python ohne weitere Einrichtung vertraut. Client IDs sind bei ConnectMQ an die ausgestellten Zugangsdaten gebunden, verwenden Sie also genau die Client ID aus dem Portal. ConnectMQ kann dekodiertes Sparkplug B außerdem in ISA-95-Topics erneut veröffentlichen, sodass Ihre simulierten DDATA für filler-01 unter einem lesbaren Pfad erscheinen, den reine JSON-Konsumenten abonnieren können, ohne Protobuf zu verstehen.

Damit testen Sie mit dem Simulator auch Ihren Unified Namespace (UNS). Im UNS wird die ISA-95-Equipment-Hierarchie (Enterprise → Site → Area → Line → Cell) zur Topic-Struktur auf dem MQTT-Broker: Die simulierte Abfüllmaschine erscheint dann etwa unter acme/werk-muenchen/abfuellung/linie-1/filler-01/..., wo MES, OEE-Dashboards oder ein Historian sie abonnieren. Wie Sparkplug B und UNS zusammenspielen, erklärt der Artikel Sparkplug B und Unified Namespace.

Fehlerbehebung

• ModuleNotFoundError: sparkplug_b_pb2: Das generierte Modul liegt nicht neben dem Skript oder nicht im PYTHONPATH.
• TypeError: Descriptors cannot be created directly: Das generierte Modul wurde mit einem älteren protoc erstellt, als Ihre protobuf-Laufzeit erwartet. Generieren Sie es neu oder setzen Sie übergangsweise PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python.
• Verbindung abgelehnt, nicht autorisiert: Auf einem privaten Broker stimmen Benutzername, Passwort oder Client ID nicht mit den Angaben aus dem Portal überein.
• Host zeigt den Node, aber keine Werte: Der Host verlangt möglicherweise Namen in DDATA und unterstützt keine Aliase. Entfernen Sie zur Bestätigung die alias=-Argumente aus dbirth und übergeben Sie Namen in ddata.
• Nachrichten auf dem öffentlichen MQTT-Broker verworfen: Sie haben die Rate pro Client überschritten. Erhöhen Sie SIM_INTERVAL.

Wie sich das in eine Anlagenarchitektur einfügt, zeigt unser Leitfaden zu Industrial IoT mit MQTT.

Ihren Sparkplug-Simulator auf ConnectMQ betreiben

Am nützlichsten ist ein Simulator gegen dieselbe Art von Broker, die Sie auch produktiv betreiben. ConnectMQ ist ein verwalteter MQTT-Broker mit MQTT 3.1.1 und 5.0, TLS, Sparkplug-B-Dekodierung und ISA-95-Republishing direkt im Portal. Starten Sie mit dem kostenlosen Serverless-Plan unter app.connectmq.com, ganz ohne Kreditkarte, und wechseln Sie zu einem kostenpflichtigen oder Dedicated-Plan, sobald aus Ihrer simulierten Anlage eine echte wird.

Häufig gestellte Fragen

Kann ich diesen Simulator mit Ignition oder anderen SCADA-Hosts nutzen?

Ja, solange der Host mit demselben Broker, derselben Group ID und dem Sparkplug-B-Namespace konfiguriert ist. Beginnen Sie mit den Szenarien „später Beitritt" und Rebirth, da diese die meisten Kompatibilitätsprobleme aufdecken.

Warum werden NBIRTH und DDATA mit QoS 0 veröffentlicht?

Die Sparkplug-Spezifikation verlangt QoS 0 für BIRTH- und DATA-Nachrichten und setzt zur Wiederherstellung auf Sequenznummern und Rebirths. Nur der NDEATH-Will nutzt QoS 1.

Warum löst Strg+C den NDEATH-Will nicht aus?

Ein sauberes MQTT-Trennen verwirft die Will-Nachricht. Deshalb veröffentlicht das Skript NDEATH beim Herunterfahren explizit, bevor es die Verbindung trennt.

Kann ich mehr als ein Device simulieren?

Ja. Veröffentlichen Sie nach dem NBIRTH ein DBIRTH pro Device, halten Sie Aliase im gesamten Edge Node eindeutig und veröffentlichen Sie DDATA auf dem jeweils eigenen Topic jedes Devices.