Como Decodificar Sparkplug B em Python: Leia Payloads Protobuf Passo a Passo

Para decodificar Sparkplug B em Python você precisa de duas coisas: classes geradas a partir do schema oficial sparkplug_b.proto e um subscriber MQTT que se lembre do que cada edge node anunciou nas suas mensagens de birth. Neste tutorial você vai construir as duas, terminar com um script que imprime cada mensagem Sparkplug como JSON legível e aprender os três detalhes que a maioria dos decodificadores rápidos erra.

O que você precisa

Se você ainda não tem um edge node publicando, rode o simulador do nosso artigo sobre certificados de birth e death do Sparkplug. Ele publica NBIRTH, DBIRTH e DDATA em um group aleatório no servidor MQTT público, que é exatamente o que este decodificador precisa.

Passo 1: Gerar as classes Python

Crie uma pasta de projeto e um ambiente virtual e depois gere o sparkplug_b_pb2.py a partir do schema publicado pelo projeto Eclipse Tahu:

mkdir sparkplug-decoder && cd sparkplug-decoder
python -m venv .venv
source .venv/bin/activate        # no 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 o grpcio-tools evita incompatibilidades de versão entre um protoc instalado separadamente e o runtime protobuf do Python. Se mais tarde aparecer um erro dizendo que o código gerado foi criado por um protoc mais novo, atualize o protobuf no mesmo ambiente e gere as classes de novo.

Passo 2: Entender o que o decodificador precisa tratar

Chamar ParseFromString sobre os bytes é a parte fácil. Um decodificador útil também precisa lidar com três detalhes específicos do Sparkplug:

1. Aliases. Os edge nodes costumam enviar os nomes completos das métricas apenas no NBIRTH e no DBIRTH e, depois, se referem às métricas por um alias numérico no NDATA e no DDATA. O decodificador precisa armazenar o mapa de aliases por node e por device.
2. Tipos de dados só nos births. O datatype é obrigatório nas mensagens de birth, mas opcional nas mensagens de dados, então o decodificador também deve memorizar o datatype de cada métrica a partir do birth.
3. Inteiros com sinal. Int8, Int16 e Int32 trafegam em um campo de 32 bits sem sinal, e Int64 em um campo de 64 bits sem sinal. Um valor -5 chega como 4294967291, a menos que você o converta usando complemento de dois.

O guia de payload Sparkplug B explica com mais profundidade a estrutura por trás dessas regras.

Passo 3: O decodificador completo

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

Execute com o group ID que o seu simulador imprimiu, por exemplo python decode_sparkplug.py demo-3f9a1c. Informar um group é uma questão de boa convivência no servidor MQTT público: a área pública é compartilhada e pode ser lida por qualquer pessoa, então assinar todos os groups ali gera ruído vindo de outros usuários.

Passo 4: Ler a saída

Uma mensagem DDATA do simulador é impressa assim:

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

Se você iniciar o decodificador depois do edge node, todas as métricas com alias vão aparecer como <unknown alias N>, porque o decodificador perdeu o birth. A solução é enviar um comando de rebirth ao node, que republica o NBIRTH e o DBIRTH. Aplicações host de produção fazem isso automaticamente; uma ferramenta simples pode simplesmente ser iniciada primeiro.

Passo 5: Solicitar um rebirth automaticamente

Um decodificador que fica passivo imprimindo aliases desconhecidos serve para uma olhada rápida. Um decodificador que alimenta um banco de dados deve corrigir a sua própria visão. A forma padrão é uma mensagem NCMD com a métrica Node Control/Rebirth definida como true. Adicione este helper e chame-o a partir de metric_to_dict quando faltar um 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}")

Passe o client para metric_to_dict (ou mantenha-o como variável de módulo) e limpe a entrada em requested quando o NBIRTH chegar, para que uma lacuna posterior possa disparar outra solicitação. Tenha cuidado com isso em ambientes compartilhados: enviar comandos é uma operação de escrita, e em uma planta em produção normalmente só a aplicação host designada deve emitir rebirths. O artigo sobre o namespace de topics do Sparkplug lista quais topics transportam comandos.

Dicas de confiabilidade para um decodificador de longa duração

Um decodificador que roda por semanas tem necessidades diferentes de um que você inicia por cinco minutos:

• Valide o seq por edge node. Guarde o último valor e sinalize qualquer lacuna que não seja a passagem de 255 para 0; uma lacuna significa que o seu cache de aliases e valores pode estar desatualizado.
• Acompanhe o bdSeq do NBIRTH e só trate um NDEATH como válido quando o bdSeq dele corresponder. Caso contrário, um will atrasado de uma sessão antiga pode marcar como offline um node que está vivo.
• Marque todas as métricas de um node como desatualizadas no NDEATH, e todas as métricas de um device como desatualizadas no DDEATH, em vez de manter o último valor como se fosse atual.
• Mantenha o on_message rápido. Faça o parse e depois entregue o resultado a uma fila ou a uma thread de trabalho se for gravar em um banco de dados, para que o loop de rede nunca bloqueie.
• Registre erros de decodificação com o topic e o tamanho do payload, e não com o payload completo, para manter os logs pequenos.

Passo 6: Conectar a um broker de produção

O servidor MQTT público serve para testes e protótipos, não para dados de produção. Para dados reais de planta, como os de centros de usinagem CNC em uma fábrica em Joinville, use credenciais do seu próprio broker e TLS. Substitua as linhas de conexão no final do script:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # usa o repositório de CAs do sistema; o broker usa um certificado Let's Encrypt
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 8883, keepalive=60)
client.loop_forever()

No ConnectMQ, os client IDs ficam vinculados às credenciais emitidas no portal, então use exatamente o client ID mostrado ali.

Solução de problemas

Decodificação em Python versus decodificação embutida no broker

O seu próprio decodificador é a ferramenta certa quando você precisa alimentar um banco de dados, um historian ou uma aplicação personalizada. Para a inspeção do dia a dia, porém, escrever código só para ver um valor é lento. O ConnectMQ decodifica Sparkplug B (Protobuf e JSON) no broker, então o topic explorer do portal mostra diretamente nomes e valores das métricas. Ele também pode republicar dados Sparkplug em topics ISA-95 legíveis, por exemplo de spBv1.0/factory1/DDATA/line1/cnc1 para connectmq/factory1/production/line1/cnc1/telemetry. Um script Python, um fluxo Node-RED, um agente Telegraf ou um gêmeo digital mais adiante na cadeia pode então assinar esses topics como um cliente MQTT comum, sem precisar embutir o arquivo proto.

Uma divisão prática que muitas equipes adotam: decodificação embutida para comissionamento e solução de problemas, e um consumidor Python como o acima quando os dados precisam ser armazenados ou transformados. Para detalhes de tipos de dados, tenha à mão a referência de tipos de dados do Sparkplug B e, para a arquitetura mais ampla, leia o guia de IoT Industrial com MQTT.

Comece com o ConnectMQ

Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, e conecte este decodificador com as credenciais geradas. Quando estiver pronto para ir além dos testes, um broker MQTT gerenciado ou um broker MQTT na nuvem elimina o trabalho operacional de manter o seu próprio broker.

Perguntas frequentes

Existe um pacote pip para decodificar Sparkplug B?

Existem vários pacotes da comunidade, mas gerar as classes a partir do proto oficial do Tahu é a abordagem mais confiável, porque você controla a versão do schema. Basta um comando com o grpcio-tools.

Posso decodificar Sparkplug B sem o arquivo proto?

Não de forma útil. Mensagens Protobuf carregam números de campo em vez de nomes, então você precisa do schema para saber o que cada campo significa.

Por que o meu decodificador mostra aliases desconhecidos?

O edge node envia os nomes das métricas apenas nas suas mensagens de birth. Se o seu decodificador iniciar depois, ele nunca aprendeu o mapa de aliases; solicite um rebirth ou inicie o decodificador primeiro.

Este decodificador funciona com MQTT 5?

Sim. Os payloads Sparkplug são os mesmos no MQTT 3.1.1 e no 5.0. Para usar MQTT 5, passe protocol=mqtt.MQTTv5 ao criar o client.