Build a Sparkplug B Edge Node Simulator in Python

A Sparkplug B edge node simulator lets you develop and test SCADA hosts, historians and dashboards without waiting for a PLC, a gateway or a maintenance window. In this tutorial you will build one in Python that publishes a correct NBIRTH, DBIRTH and DDATA sequence, registers an NDEATH will, and answers rebirth requests, all against a real MQTT broker you can reach in minutes.

What the Simulator Must Get Right

A Sparkplug consumer is stricter than a plain MQTT subscriber. A host application tracks session state per edge node and will reject or ignore data that breaks the rules. A useful Sparkplug B edge node simulator therefore has to reproduce the protocol behaviour, not just the topic names.

If any of these concepts are new, read Sparkplug birth and death certificates and the Sparkplug topic namespace first. The simulator below follows both.

Prerequisites

You need Python 3.9 or newer and two packages:

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

You also need sparkplug_b_pb2.py, the Python module generated from the official Sparkplug B Protobuf definition. The Eclipse Tahu project ships a ready-made copy in its Python core folder; download it into the same directory as your script. Our Eclipse Tahu guide explains where it lives and how to regenerate it with protoc if your protobuf version complains.

For the broker, start with the public MQTT test broker at broker.connectmq.com on port 1883. It accepts anonymous connections with any non-empty unique client ID. Sparkplug topics must start with spBv1.0, so you cannot add a private prefix; instead use a random group ID such as demo-x91q so your traffic does not collide with anyone else's. The public broker is readable by everyone and limited to 10 messages per second per client, which is plenty for a simulator publishing every couple of seconds.

The Sparkplug B Edge Node Simulator Code

Save this as edge_sim.py. It simulates one edge node with one device, a filling machine with a speed, a temperature, a running flag and a count.

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()

Run it with python edge_sim.py. You should see a births line followed by a DDATA line every two seconds.

Watching the Output

In a second terminal, confirm that messages arrive. The payloads are binary Protobuf, so mosquitto_sub will print unreadable bytes, but the topics and timing are enough for a first check:

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

For decoded values, use the alias-aware decoder from our Sparkplug metric aliases guide, or open the traffic in a ConnectMQ broker's portal, where the live topic explorer decodes Sparkplug B payloads into readable metrics. To test the death certificate, kill the simulator process with kill -9 or close the terminal window instead of pressing Ctrl+C. After the keep-alive window expires, the broker publishes the NDEATH will, and a host application should mark every metric stale.

Testing Hosts and Dashboards With the Simulator

The point of a simulator is to exercise the consumer side. Work through these scenarios before trusting a host or dashboard with a real plant.

1. Cold start. Start the host first, then the simulator. The host should create the node and device and show all four metrics.
2. Late join. Start the simulator first and the host a minute later. A correct primary host sends an NCMD rebirth because it never saw the births; the simulator prints "rebirth requested".
3. Ungraceful death. Kill the process. The host should flag the node offline once the NDEATH arrives.
4. Reconnect. Restart the simulator. Increase bd_seq between sessions in a long-running deployment so the host can match each death to the right birth; the simple script resets it on every start, which is acceptable for testing.
5. Scale. Run several copies with different SPB_NODE values to see how the host handles many edge nodes. On the public broker, keep the total message rate modest; on a private broker you can push harder within your plan's limits.
6. Command path. Send a DCMD from the host and confirm it reaches on_message. Extend the handler to change Running when commanded.

If you are evaluating a full host, the Sparkplug primary host application article explains the STATE messages a host publishes and why edge nodes in Sparkplug 3.0 can be configured to wait for them before sending births.

Moving From the Public Broker to Your Own

The public broker is for experiments; anyone can subscribe to your group. For anything resembling real data, use a private broker. On ConnectMQ, sign up, choose a plan, create a broker, and copy the generated username, password and client ID. Then run:

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 enables TLS with the broker's Let's Encrypt certificate, which Python trusts out of the box. Client IDs on ConnectMQ are bound to the issued credentials, so use the exact client ID from the portal. ConnectMQ can also republish decoded Sparkplug B into ISA-95 topics, so your simulated DDATA for filler-01 can appear under a readable path that plain JSON consumers can subscribe to without understanding Protobuf.

Troubleshooting

• ModuleNotFoundError: sparkplug_b_pb2: the generated module is not next to the script or not on PYTHONPATH.
• TypeError: Descriptors cannot be created directly: the generated module was built with an older protoc than your protobuf runtime expects. Regenerate it, or set PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python as a stopgap.
• Connection refused, not authorized: on a private broker, the username, password or client ID does not match what the portal issued.
• Host shows the node but no values: the host may require names in DDATA and not support aliases. Remove the alias= arguments from dbirth and pass names in ddata to confirm.
• Messages dropped on the public broker: you exceeded the per-client rate. Increase SIM_INTERVAL.

For how this fits into a plant architecture, see our industrial IoT MQTT guide.

Run Your Sparkplug Simulator on ConnectMQ

A simulator is most useful against the same kind of broker you will run in production. ConnectMQ is a fully managed MQTT service with MQTT 3.1.1 and 5.0, TLS, Sparkplug B decoding and ISA-95 republishing built into the portal. Start on the free Serverless plan at app.connectmq.com, no credit card required, and move to a paid or dedicated plan when your simulated plant becomes a real one.

Frequently Asked Questions

Can I use this simulator with Ignition or other SCADA hosts?

Yes, as long as the host is configured with the same broker, group ID and Sparkplug B namespace. Start with the late-join and rebirth scenarios, since those reveal most compatibility issues.

Why are NBIRTH and DDATA published with QoS 0?

The Sparkplug specification requires QoS 0 for BIRTH and DATA messages and relies on sequence numbers and rebirths for recovery. Only the NDEATH will uses QoS 1.

Why does Ctrl+C not trigger the NDEATH will?

A clean MQTT disconnect discards the will message. That is why the script publishes NDEATH explicitly on shutdown before disconnecting.

Can I simulate more than one device?

Yes. Publish one DBIRTH per device after the NBIRTH, keep aliases unique across the whole edge node, and publish DDATA on each device's own topic.