Decodificar Sparkplug B con Python: lee payloads Protobuf paso a paso

Para decodificar Sparkplug B con Python necesitas dos cosas: clases generadas a partir del esquema oficial sparkplug_b.proto y un suscriptor MQTT que recuerde lo que cada edge node anunció en sus mensajes de birth. En este tutorial construirás ambas cosas, terminarás con un script que imprime cada mensaje Sparkplug como JSON legible y aprenderás los tres detalles en los que fallan la mayoría de los decodificadores improvisados.

Qué necesitas

Si todavía no tienes un edge node publicando, ejecuta el simulador de nuestro artículo sobre certificados birth y death de Sparkplug. Publica NBIRTH, DBIRTH y DDATA en un grupo aleatorio del broker público, que es exactamente lo que necesita este decodificador.

Paso 1: generar las clases de Python

Crea una carpeta de proyecto y un entorno virtual, y genera sparkplug_b_pb2.py a partir del esquema publicado por el proyecto Eclipse Tahu:

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

Usar grpcio-tools evita incompatibilidades de versión entre un protoc instalado por separado y el runtime de protobuf de Python. Si más adelante ves un error que indica que el código generado procede de un protoc más reciente, actualiza protobuf en el mismo entorno y vuelve a generarlo.

Paso 2: entender qué debe gestionar el decodificador

Llamar a ParseFromString sobre los bytes es la parte fácil. Un decodificador útil también tiene que ocuparse de tres detalles propios de Sparkplug:

1. Alias. Los edge nodes suelen enviar los nombres completos de las métricas solo en NBIRTH y DBIRTH, y después se refieren a ellas mediante un alias numérico en NDATA y DDATA. El decodificador debe guardar el mapa de alias por nodo y dispositivo.
2. Tipos de datos solo en los births. El tipo de datos es obligatorio en los mensajes de birth pero opcional en los de datos, así que el decodificador también debe recordar el tipo de cada métrica a partir del birth.
3. Enteros con signo. Int8, Int16 e Int32 viajan en un campo de 32 bits sin signo e Int64 en uno de 64 bits sin signo. Un valor de -5 llega como 4294967291 a menos que lo conviertas usando complemento a dos.

La guía del payload Sparkplug B explica con más detalle la estructura que hay detrás de estas reglas.

Paso 3: el decodificador completo

Guarda esto como 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()

Ejecútalo con el group ID que mostró tu simulador, por ejemplo python decode_sparkplug.py demo-3f9a1c. Indicar un grupo es lo correcto en el servidor MQTT público gratuito: el área pública es compartida y legible por todo el mundo, así que suscribirse a todos los grupos genera ruido de otros usuarios.

Paso 4: leer la salida

Un mensaje DDATA del simulador se imprime así:

{
  "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 arrancas el decodificador después del edge node, las métricas con alias aparecerán como <unknown alias N> porque el decodificador se perdió el birth. La solución es enviar al nodo un comando de rebirth, que vuelve a publicar NBIRTH y DBIRTH. Las aplicaciones host de producción lo hacen automáticamente; una herramienta sencilla basta con arrancarla primero.

Paso 5: solicitar un rebirth automáticamente

Un decodificador que se limita a imprimir alias desconocidos sirve para echar un vistazo rápido. Un decodificador que alimenta una base de datos debe reparar su propia vista. La forma estándar es un mensaje NCMD con la métrica Node Control/Rebirth a true. Añade esta función auxiliar y llámala desde metric_to_dict cuando falte un alias:

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

Pasa el cliente a metric_to_dict (o mantenlo como variable a nivel de módulo) y borra la entrada de requested cuando llegue el NBIRTH, para que un salto posterior pueda desencadenar otra solicitud. Ten cuidado con esto en entornos compartidos: enviar comandos es una operación de escritura y, en una planta en producción, normalmente solo la aplicación host designada debería emitir rebirths. El artículo sobre el namespace de topics de Sparkplug enumera qué topics transportan comandos.

Consejos de fiabilidad para un decodificador de larga duración

Un decodificador que funciona durante semanas tiene necesidades distintas de uno que arrancas durante cinco minutos:

• Valida seq por cada edge node. Guarda el último valor y marca cualquier salto que no sea de 255 a 0; un salto significa que tu caché de alias y valores puede estar desactualizada.
• Registra el bdSeq del NBIRTH y trata un NDEATH como válido solo cuando su bdSeq coincida. De lo contrario, un will tardío de una sesión antigua puede marcar como offline un nodo que está vivo.
• Marca como obsoletas todas las métricas de un nodo al recibir un NDEATH, y todas las de un dispositivo al recibir un DDEATH, en lugar de conservar el último valor como si fuera actual.
• Mantén on_message rápido. Analiza el mensaje y entrega el resultado a una cola o a un hilo de trabajo si escribes en una base de datos, para que el bucle de red nunca se bloquee.
• Registra los errores de decodificación con el topic y la longitud del payload, no con el payload completo, para que los logs sean pequeños.

Paso 6: conectarse a un broker de producción

El servidor público es para pruebas y prototipos, no para datos de producción. Para datos reales de planta, usa credenciales de tu propio broker y TLS. Sustituye las líneas de conexión al final del 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()

En ConnectMQ, los client IDs están vinculados a las credenciales emitidas en el portal, así que usa exactamente el client ID que aparece allí.

Solución de problemas

Decodificación en Python frente a decodificación integrada en el broker

Tu propio decodificador es la herramienta adecuada cuando necesitas enviar datos a una base de datos, un historian o una aplicación a medida. Sin embargo, para la inspección diaria, escribir código solo para ver un valor es lento. ConnectMQ decodifica Sparkplug B (Protobuf y JSON) en el broker, de modo que el explorador de topics del portal muestra directamente los nombres y valores de las métricas. También puede republicar datos Sparkplug en topics ISA-95 legibles, por ejemplo de spBv1.0/factory1/DDATA/line1/cnc1 a connectmq/factory1/production/line1/cnc1/telemetry. Un script de Python, un flujo de Node-RED o un agente Telegraf aguas abajo pueden suscribirse a esos topics como un cliente MQTT corriente sin incluir el archivo proto.

Un reparto práctico que usan muchos equipos: decodificación integrada para la puesta en marcha y la resolución de problemas, y un consumidor en Python como el anterior cuando hay que almacenar o transformar datos. Para los detalles de los tipos de datos, ten a mano la referencia de tipos de datos de Sparkplug B y, para la arquitectura general, lee la guía de IoT industrial con MQTT.

Empieza con ConnectMQ

Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y conecta este decodificador con tus credenciales generadas. Cuando estés listo para ir más allá de las pruebas, un broker MQTT gestionado o un broker MQTT en la nube te ahorra el trabajo operativo de mantener uno propio.

Preguntas frecuentes

¿Existe un paquete pip para decodificar Sparkplug B?

Existen varios paquetes de la comunidad, pero generar las clases a partir del proto oficial de Tahu es el enfoque más fiable porque controlas la versión del esquema. Basta un comando con grpcio-tools.

¿Puedo decodificar Sparkplug B sin el archivo proto?

No de forma útil. Los mensajes Protobuf llevan números de campo en lugar de nombres, así que necesitas el esquema para saber qué significa cada campo.

¿Por qué mi decodificador muestra alias desconocidos?

El edge node envía los nombres de las métricas solo en sus mensajes de birth. Si tu decodificador arranca más tarde, nunca aprendió el mapa de alias; solicita un rebirth o arranca primero el decodificador.

¿Este decodificador funciona con MQTT 5?

Sí. Los payloads Sparkplug son iguales en MQTT 3.1.1 y 5.0. Para usar MQTT 5, pasa protocol=mqtt.MQTTv5 al crear el cliente.