Simulador de edge node Sparkplug B en Python

Un simulador de edge node Sparkplug B te permite desarrollar y probar hosts SCADA, historians y dashboards sin esperar a un PLC, un gateway o una ventana de mantenimiento. En este tutorial construirás uno en Python que publica una secuencia correcta de NBIRTH, DBIRTH y DDATA, registra un will NDEATH y responde a las solicitudes de rebirth, todo ello contra un broker MQTT real al que puedes acceder en minutos.

Qué debe hacer bien el simulador

Un consumidor Sparkplug es más estricto que un simple suscriptor MQTT. Una aplicación host sigue el estado de sesión de cada edge node y rechazará o ignorará los datos que incumplan las reglas. Por eso, un simulador de edge node Sparkplug B útil tiene que reproducir el comportamiento del protocolo, no solo los nombres de los topics.

Requisito	Por qué importa	Cómo lo gestiona el simulador
NDEATH registrado como will MQTT	Los hosts saben que el nodo se desconectó aunque no haya una desconexión limpia	`will_set` antes de conectar, QoS 1, no retained
`bdSeq` en NBIRTH y NDEATH	Los hosts asocian un death con el birth al que pone fin	Mismo valor del contador en ambos, incrementado por sesión
NBIRTH antes que cualquier otro mensaje	Los hosts necesitan primero la lista completa de métricas	Se publica en `on_connect`
DBIRTH por dispositivo	Las métricas del dispositivo se declaran antes del DDATA	Se publica justo después del NBIRTH
`seq` de 0 a 255	Los hosts detectan mensajes perdidos	El NBIRTH usa 0, cada mensaje posterior lo incrementa y vuelve a empezar
Soporte de rebirth	Los hosts recuperan el estado bajo demanda	Un NCMD `Node Control/Rebirth` provoca births nuevos
Report by exception	Patrón de tráfico realista	Solo las métricas modificadas van en el DDATA

Si alguno de estos conceptos es nuevo para ti, lee primero certificados birth y death de Sparkplug y el namespace de topics de Sparkplug. El simulador siguiente respeta ambos.

Requisitos previos

Necesitas Python 3.9 o posterior y dos paquetes:

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

También necesitas sparkplug_b_pb2.py, el módulo de Python generado a partir de la definición Protobuf oficial de Sparkplug B. El proyecto Eclipse Tahu incluye una copia lista para usar en su carpeta de núcleo de Python; descárgala en el mismo directorio que tu script. Nuestra guía de Eclipse Tahu explica dónde se encuentra y cómo regenerarla con protoc si tu versión de protobuf da problemas.

Como broker, empieza con el servidor MQTT público gratuito en broker.connectmq.com, puerto 1883. Acepta conexiones anónimas con cualquier client ID único y no vacío. Los topics Sparkplug deben empezar por spBv1.0, así que no puedes añadir un prefijo privado; en su lugar, usa un group ID aleatorio como demo-x91q para que tu tráfico no choque con el de otros. El broker público es legible por todo el mundo y está limitado a 10 mensajes por segundo por cliente, más que suficiente para un simulador que publica cada par de segundos.

El código del simulador de edge node Sparkplug B

Guárdalo como edge_sim.py. Simula un edge node con un dispositivo, una llenadora con velocidad, temperatura, un indicador de funcionamiento y un contador.

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

Ejecútalo con python edge_sim.py. Deberías ver una línea de births seguida de una línea DDATA cada dos segundos.

Observar la salida

En una segunda terminal, confirma que llegan los mensajes. Los payloads son Protobuf binario, así que mosquitto_sub mostrará bytes ilegibles, pero los topics y la cadencia bastan para una primera comprobación:

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

Para ver los valores decodificados, usa el decodificador con soporte de alias de nuestra guía sobre alias de métricas de Sparkplug o abre el tráfico en el portal de un broker ConnectMQ, cuyo explorador de topics en vivo decodifica los payloads Sparkplug B en métricas legibles. Para probar el certificado de death, mata el proceso del simulador con kill -9 o cierra la ventana de la terminal en lugar de pulsar Ctrl+C. Cuando expire el intervalo de keep-alive, el broker publicará el will NDEATH y una aplicación host debería marcar todas las métricas como obsoletas.

Probar hosts y dashboards con el simulador

La razón de ser de un simulador es poner a prueba el lado consumidor. Recorre estos escenarios antes de confiarle una planta real a un host o un dashboard.

Arranque en frío. Inicia primero el host y después el simulador. El host debería crear el nodo y el dispositivo y mostrar las cuatro métricas.
Incorporación tardía. Inicia primero el simulador y el host un minuto después. Un host primario correcto envía un NCMD de rebirth porque nunca vio los births; el simulador muestra "rebirth requested".
Death no ordenado. Mata el proceso. El host debería marcar el nodo como offline en cuanto llegue el NDEATH.
Reconexión. Reinicia el simulador. En un despliegue de larga duración, incrementa bd_seq entre sesiones para que el host pueda asociar cada death con el birth correcto; el script simple lo reinicia en cada arranque, lo cual es aceptable para pruebas.
Escala. Ejecuta varias copias con distintos valores de SPB_NODE para ver cómo gestiona el host muchos edge nodes. En el broker público, mantén moderada la tasa total de mensajes; en un broker privado puedes apretar más dentro de los límites de tu plan.
Ruta de comandos. Envía un DCMD desde el host y confirma que llega a on_message. Amplía el manejador para cambiar Running cuando se le ordene.

Si estás evaluando un host completo, el artículo sobre la aplicación host primaria de Sparkplug explica los mensajes STATE que publica un host y por qué los edge nodes de Sparkplug 3.0 pueden configurarse para esperarlos antes de enviar sus births.

Pasar del broker público al tuyo propio

El broker público es para experimentos; cualquiera puede suscribirse a tu grupo. Para cualquier cosa parecida a datos reales, usa un broker privado. En ConnectMQ, regístrate, elige un plan, crea un broker y copia el nombre de usuario, la contraseña y el client ID generados. Después ejecuta:

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

El puerto 8883 activa TLS con el certificado Let's Encrypt del broker, en el que Python confía de serie. En ConnectMQ, los client IDs están vinculados a las credenciales emitidas, así que usa exactamente el client ID del portal. ConnectMQ también puede republicar Sparkplug B decodificado en topics ISA-95, de modo que tu DDATA simulado para filler-01 puede aparecer bajo una ruta legible a la que pueden suscribirse consumidores de JSON simple sin entender Protobuf.

Solución de problemas

ModuleNotFoundError: sparkplug_b_pb2: el módulo generado no está junto al script ni en el PYTHONPATH.
TypeError: Descriptors cannot be created directly: el módulo generado se creó con un protoc más antiguo de lo que espera tu runtime de protobuf. Regenéralo o, como solución temporal, establece PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python.
Conexión rechazada, no autorizado: en un broker privado, el nombre de usuario, la contraseña o el client ID no coinciden con los emitidos por el portal.
El host muestra el nodo pero no los valores: puede que el host exija nombres en el DDATA y no admita alias. Elimina los argumentos alias= de dbirth y pasa los nombres en ddata para confirmarlo.
Mensajes descartados en el broker público: has superado la tasa por cliente. Aumenta SIM_INTERVAL.

Para ver cómo encaja esto en una arquitectura de planta, consulta nuestra guía de MQTT para IoT industrial.

Ejecuta tu simulador Sparkplug en ConnectMQ

Un simulador es más útil contra el mismo tipo de broker que usarás en producción. ConnectMQ es un broker MQTT gestionado con MQTT 3.1.1 y 5.0, TLS, decodificación de Sparkplug B y republicación ISA-95 integradas en el portal. Empieza con el plan Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y pasa a un plan de pago o Dedicated cuando tu planta simulada se convierta en una real.

Preguntas frecuentes

¿Puedo usar este simulador con Ignition u otros hosts SCADA?

Sí, siempre que el host esté configurado con el mismo broker, group ID y namespace Sparkplug B. Empieza por los escenarios de incorporación tardía y rebirth, ya que son los que revelan la mayoría de los problemas de compatibilidad.

¿Por qué se publican NBIRTH y DDATA con QoS 0?

La especificación Sparkplug exige QoS 0 para los mensajes BIRTH y DATA y confía en los números de secuencia y los rebirths para la recuperación. Solo el will NDEATH usa QoS 1.

¿Por qué Ctrl+C no dispara el will NDEATH?

Una desconexión MQTT limpia descarta el mensaje will. Por eso el script publica explícitamente el NDEATH al apagarse, antes de desconectarse.

¿Puedo simular más de un dispositivo?

Sí. Publica un DBIRTH por dispositivo después del NBIRTH, mantén los alias únicos en todo el edge node y publica el DDATA en el topic propio de cada dispositivo.

Crea un simulador de edge node Sparkplug B en Python