Decode Sparkplug B Python Tutorial: Read Protobuf Payloads Step by Step

To decode Sparkplug B in Python you need two things: classes generated from the official sparkplug_b.proto schema and an MQTT subscriber that remembers what each edge node announced in its birth messages. In this tutorial you will build both, end up with a script that prints every Sparkplug message as readable JSON, and learn the three details that most quick decoders get wrong.

What You Need

If you do not have an edge node publishing yet, run the simulator from our article on Sparkplug birth and death certificates. It publishes NBIRTH, DBIRTH and DDATA to a random group on the public broker, which is exactly what this decoder needs.

Step 1: Generate the Python Classes

Create a project folder and a virtual environment, then generate sparkplug_b_pb2.py from the schema published by the Eclipse Tahu project:

mkdir sparkplug-decoder && cd sparkplug-decoder
python -m venv .venv
source .venv/bin/activate        # on 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

Using grpcio-tools avoids version mismatches between a separately installed protoc and the Python protobuf runtime. If you later see an error about the generated code being made by a newer protoc, upgrade protobuf in the same environment and regenerate.

Step 2: Understand What the Decoder Must Handle

Calling ParseFromString on the bytes is the easy part. A useful decoder also has to deal with three Sparkplug-specific details:

1. Aliases. Edge nodes often send full metric names only in NBIRTH and DBIRTH, then refer to metrics by a numeric alias in NDATA and DDATA. The decoder must store the alias map per node and device.
2. Datatypes in births only. The datatype is required in birth messages but optional in data messages, so the decoder should also remember each metric's datatype from the birth.
3. Signed integers. Int8, Int16 and Int32 travel in an unsigned 32-bit field and Int64 in an unsigned 64-bit field. A value of -5 arrives as 4294967291 unless you convert it using two's complement.

The Sparkplug B payload guide explains the structure behind these rules in more depth.

Step 3: The Complete Decoder

Save this as 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()

Run it with the group ID your simulator printed, for example python decode_sparkplug.py demo-3f9a1c. Passing a group is polite on the public MQTT test broker: the public area is shared and readable by everyone, so subscribing to every group there produces noise from other users.

Step 4: Read the Output

A DDATA message from the simulator prints like this:

{
  "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 }
  ]
}

If you start the decoder after the edge node, any aliased metrics will show as <unknown alias N> because the decoder missed the birth. The fix is to send a rebirth command to the node, which republishes NBIRTH and DBIRTH. Production host applications do that automatically; a simple tool can just be started first.

Step 5: Request a Rebirth Automatically

A decoder that sits passively and prints unknown aliases is fine for a quick look. A decoder that feeds a database should repair its own view. The standard way is an NCMD message with the Node Control/Rebirth metric set to true. Add this helper and call it from metric_to_dict when an alias is missing:

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}")

Pass the client into metric_to_dict (or keep it as a module-level variable) and clear the requested entry when the NBIRTH arrives so a later gap can trigger another request. Be careful with this in shared environments: sending commands is a write operation, and in a production plant only the designated host application should normally issue rebirths. The Sparkplug topic namespace article lists which topics carry commands.

Reliability Tips for a Long-Running Decoder

A decoder that runs for weeks has different needs from one you start for five minutes:

• Validate seq per edge node. Store the last value and flag any gap other than 255 to 0; a gap means your alias and value cache may be stale.
• Track bdSeq from NBIRTH and only treat an NDEATH as valid when its bdSeq matches. Otherwise a late will from an old session can mark a live node offline.
• Mark every metric of a node as stale on NDEATH, and every metric of a device as stale on DDEATH, instead of keeping the last value as if it were current.
• Keep on_message fast. Parse, then hand the result to a queue or a worker thread if you write to a database, so the network loop never blocks.
• Log decode errors with the topic and payload length, not the full payload, to keep logs small.

Step 6: Connect to a Production Broker

The public broker is for testing. For real plant data, use credentials from your own broker and TLS. Replace the connection lines at the bottom of the script:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # uses the system CA store; the broker uses a Let's Encrypt certificate
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 8883, keepalive=60)
client.loop_forever()

On ConnectMQ, client IDs are bound to the credentials issued in the portal, so use the exact client ID shown there.

Troubleshooting

Python Decoding Versus Built-In Broker Decoding

Your own decoder is the right tool when you need to feed data into a database, a historian or a custom application. For everyday inspection, though, writing code just to see a value is slow. ConnectMQ decodes Sparkplug B (Protobuf and JSON) in the broker, so the portal topic explorer shows metric names and values directly. It can also republish Sparkplug data into readable ISA-95 topics, for example spBv1.0/factory1/DDATA/line1/cnc1 to connectmq/factory1/production/line1/cnc1/telemetry. A downstream Python script, Node-RED flow or Telegraf agent can then subscribe to those topics as an ordinary MQTT client without bundling the proto file.

A practical split many teams use: built-in decoding for commissioning and troubleshooting, and a Python consumer like the one above when data must be stored or transformed. For datatype details, keep the Sparkplug B data types reference at hand, and for the bigger architecture read the Industrial IoT with MQTT guide.

Get Started with ConnectMQ

Create a free Serverless broker at app.connectmq.com, no credit card required, and connect this decoder with your generated credentials. When you are ready to scale beyond tests, a fully managed MQTT service or a cloud MQTT broker removes the operational work of running your own.

Frequently Asked Questions

Is there a pip package for Sparkplug B decoding?

Several community packages exist, but generating classes from the official Tahu proto is the most reliable approach because you control the schema version. It takes one command with grpcio-tools.

Can I decode Sparkplug B without the proto file?

Not in a meaningful way. Protobuf messages carry field numbers rather than names, so you need the schema to know what each field means.

Why does my decoder show unknown aliases?

The edge node sends metric names only in its birth messages. If your decoder starts later, it never learned the alias map; request a rebirth or start the decoder first.

Does this decoder work with MQTT 5?

Yes. Sparkplug payloads are the same over MQTT 3.1.1 and 5.0. To use MQTT 5, pass protocol=mqtt.MQTTv5 when creating the client.