Décoder Sparkplug B en Python : lire les payloads Protobuf pas à pas

Pour décoder Sparkplug B en Python, il vous faut deux choses : des classes générées à partir du schéma officiel sparkplug_b.proto et un abonné MQTT qui mémorise ce que chaque edge node a annoncé dans ses messages de birth. Dans ce tutoriel, vous construirez les deux, obtiendrez un script qui affiche chaque message Sparkplug en JSON lisible, et découvrirez les trois détails que la plupart des décodeurs écrits à la va-vite gèrent mal.

Ce dont vous avez besoin

Si vous n'avez pas encore d'edge node qui publie, lancez le simulateur de notre article sur les certificats birth et death Sparkplug. Il publie NBIRTH, DBIRTH et DDATA dans un groupe aléatoire sur le serveur MQTT public gratuit, exactement ce dont ce décodeur a besoin.

Étape 1 : générer les classes Python

Créez un dossier de projet et un environnement virtuel, puis générez sparkplug_b_pb2.py à partir du schéma publié par le projet Eclipse Tahu :

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

Utiliser grpcio-tools évite les incompatibilités de version entre un protoc installé séparément et le runtime protobuf de Python. Si vous voyez plus tard une erreur indiquant que le code généré provient d'un protoc plus récent, mettez à jour protobuf dans le même environnement et régénérez.

Étape 2 : comprendre ce que le décodeur doit gérer

Appeler ParseFromString sur les octets est la partie facile. Un décodeur utile doit aussi gérer trois détails propres à Sparkplug :

1. Les alias. Les edge nodes n'envoient souvent les noms complets des métriques que dans le NBIRTH et le DBIRTH, puis désignent les métriques par un alias numérique dans NDATA et DDATA. Le décodeur doit conserver la table des alias par nœud et par device.
2. Les types de données uniquement dans les births. Le type de données est obligatoire dans les messages de birth mais facultatif dans les messages de données : le décodeur doit donc aussi mémoriser le type de chaque métrique à partir du birth.
3. Les entiers signés. Int8, Int16 et Int32 voyagent dans un champ non signé de 32 bits, et Int64 dans un champ non signé de 64 bits. Une valeur de -5 arrive sous la forme 4294967291 si vous ne la convertissez pas par complément à deux.

Le guide du payload Sparkplug B explique plus en détail la structure qui sous-tend ces règles.

Étape 3 : le décodeur complet

Enregistrez ceci sous 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()

Lancez-le avec le group ID affiché par votre simulateur, par exemple python decode_sparkplug.py demo-3f9a1c. Préciser un groupe est une question de courtoisie sur le serveur MQTT public gratuit : l'espace public est partagé et lisible par tous, et s'abonner à tous les groupes y produit le bruit des autres utilisateurs.

Étape 4 : lire la sortie

Un message DDATA du simulateur s'affiche ainsi :

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

Si vous démarrez le décodeur après l'edge node, les métriques avec alias s'afficheront sous la forme <unknown alias N>, car le décodeur a manqué le birth. La solution consiste à envoyer une commande de rebirth au nœud, qui republie alors le NBIRTH et le DBIRTH. Les applications hôtes de production le font automatiquement ; un outil simple peut tout bonnement être démarré en premier.

Étape 5 : demander un rebirth automatiquement

Un décodeur passif qui affiche des alias inconnus convient pour un coup d'œil rapide. Un décodeur qui alimente une base de données doit réparer sa propre vue. La méthode standard est un message NCMD avec la métrique Node Control/Rebirth à true. Ajoutez cette fonction et appelez-la depuis metric_to_dict lorsqu'un alias manque :

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

Passez le client à metric_to_dict (ou conservez-le dans une variable de module) et supprimez l'entrée de requested à l'arrivée du NBIRTH, afin qu'un trou ultérieur puisse déclencher une nouvelle demande. Soyez prudent dans les environnements partagés : envoyer des commandes est une opération d'écriture, et dans une usine en production seule l'application hôte désignée devrait normalement émettre des rebirths. L'article sur l'espace de noms des topics Sparkplug liste les topics qui transportent des commandes.

Conseils de fiabilité pour un décodeur de longue durée

Un décodeur qui tourne pendant des semaines n'a pas les mêmes besoins qu'un décodeur lancé cinq minutes :

• Validez seq par edge node. Conservez la dernière valeur et signalez tout trou autre que le passage de 255 à 0 ; un trou signifie que votre cache d'alias et de valeurs est peut-être périmé.
• Suivez le bdSeq du NBIRTH et ne considérez un NDEATH comme valide que si son bdSeq correspond. Sinon, un will tardif d'une ancienne session peut marquer un nœud actif comme hors ligne.
• Marquez toutes les métriques d'un nœud comme périmées sur NDEATH, et toutes celles d'un device sur DDEATH, au lieu de garder la dernière valeur comme si elle était à jour.
• Gardez on_message rapide. Analysez, puis confiez le résultat à une file d'attente ou à un thread de travail si vous écrivez en base, pour que la boucle réseau ne bloque jamais.
• Journalisez les erreurs de décodage avec le topic et la longueur du payload, pas le payload complet, pour garder des logs légers.

Étape 6 : se connecter à un broker de production

Le serveur MQTT public gratuit sert aux tests et aux prototypes. Pour de vraies données d'usine, utilisez les identifiants de votre propre broker et TLS. Remplacez les lignes de connexion en bas du script :

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # utilise le magasin de CA du système ; le broker utilise un certificat Let's Encrypt
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 8883, keepalive=60)
client.loop_forever()

Sur ConnectMQ, les client ID sont liés aux identifiants émis dans le portail : utilisez exactement le client ID qui y est affiché.

Dépannage

Décodage Python ou décodage intégré au broker

Votre propre décodeur est le bon outil lorsque vous devez alimenter une base de données, un historian ou une application sur mesure. Pour l'inspection quotidienne, en revanche, écrire du code juste pour voir une valeur est lent. ConnectMQ décode Sparkplug B (Protobuf et JSON) dans le broker : l'explorateur de topics du portail affiche directement les noms et valeurs des métriques. Il peut aussi republier les données Sparkplug dans des topics ISA-95 lisibles, par exemple spBv1.0/factory1/DDATA/line1/cnc1 vers connectmq/factory1/production/line1/cnc1/telemetry. Un script Python, un flow Node-RED ou un agent Telegraf en aval peut alors s'abonner à ces topics comme un client MQTT ordinaire, sans embarquer le fichier proto.

Une répartition pratique adoptée par de nombreuses équipes : le décodage intégré pour la mise en service et le dépannage, et un consommateur Python comme celui ci-dessus lorsque les données doivent être stockées ou transformées. Pour le détail des types, gardez sous la main la référence des types de données Sparkplug B, et pour l'architecture d'ensemble, lisez le guide de l'IoT industriel avec MQTT.

Démarrer avec ConnectMQ

Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire, et connectez ce décodeur avec vos identifiants générés. Lorsque vous êtes prêt à aller au-delà des tests, un broker MQTT managé ou un broker MQTT cloud vous libère du travail d'exploitation de votre propre infrastructure.

Questions fréquentes

Existe-t-il un paquet pip pour décoder Sparkplug B ?

Plusieurs paquets communautaires existent, mais générer les classes à partir du proto officiel de Tahu est l'approche la plus fiable, car vous maîtrisez la version du schéma. Cela ne prend qu'une commande avec grpcio-tools.

Peut-on décoder Sparkplug B sans le fichier proto ?

Pas de manière exploitable. Les messages Protobuf portent des numéros de champ plutôt que des noms : il faut le schéma pour savoir ce que signifie chaque champ.

Pourquoi mon décodeur affiche-t-il des alias inconnus ?

L'edge node n'envoie les noms des métriques que dans ses messages de birth. Si votre décodeur démarre plus tard, il n'a jamais appris la table des alias ; demandez un rebirth ou démarrez le décodeur en premier.

Ce décodeur fonctionne-t-il avec MQTT 5 ?

Oui. Les payloads Sparkplug sont identiques en MQTT 3.1.1 et 5.0. Pour utiliser MQTT 5, passez protocol=mqtt.MQTTv5 lors de la création du client.