Créer un simulateur d'edge node Sparkplug B en Python

Un simulateur d'edge node Sparkplug B vous permet de développer et tester des hôtes SCADA, des historians et des tableaux de bord sans attendre un automate, une passerelle ou une fenêtre de maintenance. Dans ce tutoriel, vous en construirez un en Python qui publie une séquence NBIRTH, DBIRTH et DDATA correcte, enregistre un will NDEATH et répond aux demandes de rebirth, le tout sur un vrai broker MQTT accessible en quelques minutes.

Ce que le simulateur doit respecter

Un consommateur Sparkplug est plus strict qu'un simple abonné MQTT. Une application hôte suit l'état de session de chaque edge node et rejettera ou ignorera les données qui enfreignent les règles. Un simulateur d'edge node Sparkplug B utile doit donc reproduire le comportement du protocole, pas seulement les noms de topics.

Si certaines de ces notions sont nouvelles pour vous, lisez d'abord certificats birth et death Sparkplug et l'espace de noms des topics Sparkplug. Le simulateur ci-dessous respecte les deux.

Prérequis

Il vous faut Python 3.9 ou plus récent et deux paquets :

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

Il vous faut aussi sparkplug_b_pb2.py, le module Python généré à partir de la définition Protobuf officielle de Sparkplug B. Le projet Eclipse Tahu en fournit une copie prête à l'emploi dans son dossier Python core ; téléchargez-la dans le même répertoire que votre script. Notre guide Eclipse Tahu explique où elle se trouve et comment la régénérer avec protoc si votre version de protobuf proteste.

Pour le broker, commencez par le serveur MQTT public gratuit sur broker.connectmq.com, port 1883. Il accepte les connexions anonymes avec n'importe quel client ID unique non vide. Les topics Sparkplug doivent commencer par spBv1.0 : vous ne pouvez donc pas ajouter de préfixe privé. Utilisez plutôt un group ID aléatoire comme demo-x91q pour que votre trafic n'entre pas en collision avec celui des autres. Le broker public est lisible par tous et limité à 10 messages par seconde et par client, ce qui suffit largement pour un simulateur qui publie toutes les deux secondes.

Le code du simulateur d'edge node Sparkplug B

Enregistrez ceci sous edge_sim.py. Il simule un edge node avec un device, une remplisseuse dotée d'une vitesse, d'une température, d'un indicateur de marche et d'un compteur.

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")          # à définir pour un broker ConnectMQ privé
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"))

# Codes de types de données Sparkplug B
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):
        """Modifie les valeurs et ne renvoie que les métriques modifiées."""
        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:
    # Un arrêt propre publie explicitement le NDEATH, car le will n'est pas envoyé
    client.publish(topic("NDEATH"), node.death_payload(), qos=1).wait_for_publish(5)
    client.loop_stop()
    client.disconnect()

Lancez-le avec python edge_sim.py. Vous devriez voir une ligne de births, puis une ligne DDATA toutes les deux secondes.

Observer la sortie

Dans un second terminal, vérifiez que les messages arrivent. Les payloads sont en Protobuf binaire, donc mosquitto_sub affichera des octets illisibles, mais les topics et le rythme suffisent pour une première vérification :

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

Pour des valeurs décodées, utilisez le décodeur compatible alias de notre guide sur les alias de métriques Sparkplug, ou ouvrez le trafic dans le portail d'un broker ConnectMQ, dont l'explorateur de topics en direct décode les payloads Sparkplug B en métriques lisibles. Pour tester le certificat de death, tuez le processus du simulateur avec kill -9 ou fermez la fenêtre du terminal au lieu d'appuyer sur Ctrl+C. Une fois la fenêtre de keep-alive expirée, le broker publie le will NDEATH, et une application hôte doit marquer toutes les métriques comme périmées.

Tester hôtes et tableaux de bord avec le simulateur

L'intérêt d'un simulateur est de mettre à l'épreuve le côté consommateur. Déroulez ces scénarios avant de confier une vraie usine à un hôte ou à un tableau de bord.

1. Démarrage à froid. Démarrez d'abord l'hôte, puis le simulateur. L'hôte doit créer le nœud et le device et afficher les quatre métriques.
2. Arrivée tardive. Démarrez d'abord le simulateur, puis l'hôte une minute plus tard. Un hôte primaire correct envoie un NCMD de rebirth parce qu'il n'a jamais vu les births ; le simulateur affiche « rebirth requested ».
3. Death non propre. Tuez le processus. L'hôte doit signaler le nœud hors ligne dès l'arrivée du NDEATH.
4. Reconnexion. Redémarrez le simulateur. Dans un déploiement de longue durée, incrémentez bd_seq entre les sessions pour que l'hôte associe chaque death au bon birth ; le script simple le réinitialise à chaque démarrage, ce qui est acceptable pour des tests.
5. Montée en charge. Lancez plusieurs copies avec des valeurs SPB_NODE différentes pour voir comment l'hôte gère de nombreux edge nodes. Sur le broker public, gardez un débit total modéré ; sur un broker privé, vous pouvez pousser davantage dans les limites de votre offre.
6. Chemin des commandes. Envoyez un DCMD depuis l'hôte et vérifiez qu'il atteint on_message. Étendez le gestionnaire pour modifier Running sur commande.

Si vous évaluez un hôte complet, l'article sur l'application hôte primaire Sparkplug explique les messages STATE qu'un hôte publie et pourquoi les edge nodes Sparkplug 3.0 peuvent être configurés pour les attendre avant d'envoyer leurs births.

Passer du broker public à votre propre broker

Le broker public sert aux expériences ; n'importe qui peut s'abonner à votre groupe. Pour tout ce qui ressemble à des données réelles, utilisez un broker privé. Sur ConnectMQ, inscrivez-vous, choisissez une offre, créez un broker et copiez le nom d'utilisateur, le mot de passe et le client ID générés. Lancez ensuite :

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

Le port 8883 active TLS avec le certificat Let's Encrypt du broker, auquel Python fait confiance d'emblée. Sur ConnectMQ, les client ID sont liés aux identifiants émis : utilisez exactement le client ID indiqué dans le portail. ConnectMQ peut aussi republier le Sparkplug B décodé dans des topics ISA-95, si bien que vos DDATA simulés pour filler-01 peuvent apparaître sous un chemin lisible auquel des consommateurs JSON classiques peuvent s'abonner sans comprendre Protobuf.

Dépannage

• ModuleNotFoundError: sparkplug_b_pb2 : le module généré n'est pas à côté du script ni dans le PYTHONPATH.
• TypeError: Descriptors cannot be created directly : le module généré a été produit avec un protoc plus ancien que ce qu'attend votre runtime protobuf. Régénérez-le, ou définissez PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python comme solution provisoire.
• Connexion refusée, non autorisée : sur un broker privé, le nom d'utilisateur, le mot de passe ou le client ID ne correspond pas à ce que le portail a émis.
• L'hôte affiche le nœud mais aucune valeur : l'hôte exige peut-être des noms dans les DDATA et ne prend pas en charge les alias. Retirez les arguments alias= de dbirth et passez les noms dans ddata pour le vérifier.
• Messages perdus sur le broker public : vous avez dépassé le débit par client. Augmentez SIM_INTERVAL.

Pour situer ce simulateur dans une architecture d'usine, consultez notre guide de l'IoT industriel avec MQTT.

Exécuter votre simulateur Sparkplug sur ConnectMQ

Un simulateur est le plus utile face au même type de broker que celui que vous exploiterez en production. ConnectMQ est un broker MQTT managé avec MQTT 3.1.1 et 5.0, TLS, décodage Sparkplug B et republication ISA-95 intégrés au portail. Commencez avec l'offre Serverless gratuite sur app.connectmq.com, sans carte bancaire, et passez à une offre payante ou dédiée lorsque votre usine simulée devient bien réelle.

Questions fréquentes

Puis-je utiliser ce simulateur avec Ignition ou d'autres hôtes SCADA ?

Oui, à condition que l'hôte soit configuré avec le même broker, le même group ID et l'espace de noms Sparkplug B. Commencez par les scénarios d'arrivée tardive et de rebirth, qui révèlent la plupart des problèmes de compatibilité.

Pourquoi NBIRTH et DDATA sont-ils publiés en QoS 0 ?

La spécification Sparkplug impose la QoS 0 pour les messages BIRTH et DATA et s'appuie sur les numéros de séquence et les rebirths pour la récupération. Seul le will NDEATH utilise la QoS 1.

Pourquoi Ctrl+C ne déclenche-t-il pas le will NDEATH ?

Une déconnexion MQTT propre supprime le message will. C'est pourquoi le script publie explicitement le NDEATH à l'arrêt, avant de se déconnecter.

Puis-je simuler plusieurs devices ?

Oui. Publiez un DBIRTH par device après le NBIRTH, gardez des alias uniques sur l'ensemble de l'edge node, et publiez les DDATA sur le topic propre à chaque device.