Crie um Simulador de Edge Node Sparkplug B em Python

Um simulador de edge node Sparkplug B permite desenvolver e testar hosts SCADA, historians e dashboards sem esperar por um CLP, um gateway ou uma janela de manutenção. Neste tutorial você vai construir um em Python que publica uma sequência correta de NBIRTH, DBIRTH e DDATA, registra um will NDEATH e responde a solicitações de rebirth, tudo contra um broker MQTT real que você consegue acessar em minutos.

O que o simulador precisa acertar

Um consumidor Sparkplug é mais rigoroso do que um subscriber MQTT comum. Uma aplicação host acompanha o estado da sessão de cada edge node e rejeita ou ignora dados que quebram as regras. Um simulador de edge node Sparkplug B útil, portanto, precisa reproduzir o comportamento do protocolo, e não apenas os nomes dos topics.

Se algum desses conceitos for novo, leia primeiro certificados de birth e death do Sparkplug e o namespace de topics do Sparkplug. O simulador abaixo segue os dois.

Pré-requisitos

Você precisa do Python 3.9 ou mais recente e de dois pacotes:

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

Você também precisa do sparkplug_b_pb2.py, o módulo Python gerado a partir da definição Protobuf oficial do Sparkplug B. O projeto Eclipse Tahu traz uma cópia pronta na sua pasta Python core; baixe-a para o mesmo diretório do seu script. Nosso guia sobre o Eclipse Tahu explica onde ela fica e como gerá-la de novo com o protoc se a sua versão do protobuf reclamar.

Como broker, comece com o servidor MQTT público (gratuito) em broker.connectmq.com, na porta 1883. Ele aceita conexões anônimas com qualquer client ID único e não vazio. Os topics Sparkplug precisam começar com spBv1.0, então você não pode adicionar um prefixo privado; em vez disso, use um group ID aleatório como demo-x91q para que o seu tráfego não colida com o de outras pessoas. O servidor MQTT público pode ser lido por qualquer pessoa e é limitado a 10 mensagens por segundo por cliente, o que é mais do que suficiente para um simulador que publica a cada dois segundos.

O código do simulador de edge node Sparkplug B

Salve isto como edge_sim.py. Ele simula um edge node com um device, uma envasadora com velocidade, temperatura, um indicador de funcionamento e uma contagem.

import os
import random
import ssl
import time

import paho.mqtt.client as mqtt
import sparkplug_b_pb2

# --- configuração -------------------------------------------------------------
BROKER = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
USERNAME = os.getenv("MQTT_USERNAME")          # defina para um broker ConnectMQ privado
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"))

# Códigos de tipo de dado do 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):
        """Altera os valores e retorna apenas as métricas que mudaram."""
        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:
    # Um encerramento limpo publica o NDEATH explicitamente, porque o will não é enviado
    client.publish(topic("NDEATH"), node.death_payload(), qos=1).wait_for_publish(5)
    client.loop_stop()
    client.disconnect()

Execute com python edge_sim.py. Você deve ver uma linha de births seguida de uma linha DDATA a cada dois segundos.

Acompanhando a saída

Em um segundo terminal, confirme que as mensagens estão chegando. Os payloads são Protobuf binário, então o mosquitto_sub vai imprimir bytes ilegíveis, mas os topics e o ritmo bastam para uma primeira verificação:

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

Para ver os valores decodificados, use o decodificador com suporte a aliases do nosso guia sobre aliases de métricas do Sparkplug ou abra o tráfego no portal de um broker ConnectMQ, onde o topic explorer ao vivo decodifica payloads Sparkplug B em métricas legíveis. Para testar o certificado de death, mate o processo do simulador com kill -9 ou feche a janela do terminal em vez de pressionar Ctrl+C. Depois que a janela de keep-alive expirar, o broker publica o will NDEATH, e uma aplicação host deve marcar todas as métricas como desatualizadas.

Testando hosts e dashboards com o simulador

O objetivo de um simulador é exercitar o lado do consumidor. Passe por estes cenários antes de confiar um host ou dashboard a uma planta real.

1. Partida a frio. Inicie o host primeiro e depois o simulador. O host deve criar o node e o device e mostrar as quatro métricas.
2. Entrada tardia. Inicie o simulador primeiro e o host um minuto depois. Um host primário correto envia um NCMD de rebirth porque nunca viu os births; o simulador imprime "rebirth requested".
3. Death sem desconexão limpa. Mate o processo. O host deve marcar o node como offline assim que o NDEATH chegar.
4. Reconexão. Reinicie o simulador. Em uma implantação de longa duração, incremente o bd_seq entre sessões para que o host possa associar cada death ao birth certo; o script simples o zera a cada início, o que é aceitável para testes.
5. Escala. Rode várias cópias com valores diferentes de SPB_NODE para ver como o host lida com muitos edge nodes. No servidor MQTT público, mantenha a taxa total de mensagens moderada; em um broker privado você pode ir além, dentro dos limites do seu plano.
6. Caminho de comandos. Envie um DCMD a partir do host e confirme que ele chega ao on_message. Estenda o handler para alterar Running quando receber o comando.

Se você está avaliando um host completo, o artigo sobre a Primary Host Application do Sparkplug explica as mensagens STATE que um host publica e por que, no Sparkplug 3.0, os edge nodes podem ser configurados para esperá-las antes de enviar os births.

Saindo do servidor MQTT público para o seu próprio broker

O servidor MQTT público serve para experimentos; qualquer pessoa pode assinar o seu group. Para qualquer coisa parecida com dados reais, use um broker privado. No ConnectMQ, cadastre-se, escolha um plano, crie um broker e copie o username, a senha e o client ID gerados. Depois execute:

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

A porta 8883 habilita TLS com o certificado Let's Encrypt do broker, no qual o Python confia sem configuração adicional. No ConnectMQ, os client IDs ficam vinculados às credenciais emitidas, então use exatamente o client ID do portal. O ConnectMQ também pode republicar Sparkplug B decodificado em topics ISA-95, de modo que o DDATA simulado do filler-01 pode aparecer em um caminho legível que consumidores de JSON simples conseguem assinar sem entender Protobuf.

Solução de problemas

• ModuleNotFoundError: sparkplug_b_pb2: o módulo gerado não está ao lado do script nem no PYTHONPATH.
• TypeError: Descriptors cannot be created directly: o módulo gerado foi construído com um protoc mais antigo do que o runtime protobuf espera. Gere-o de novo ou defina PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python como solução temporária.
• Connection refused, not authorized: em um broker privado, o username, a senha ou o client ID não correspondem ao que o portal emitiu.
• O host mostra o node, mas nenhum valor: o host pode exigir nomes no DDATA e não suportar aliases. Remova os argumentos alias= de dbirth e passe os nomes em ddata para confirmar.
• Mensagens descartadas no servidor MQTT público: você excedeu a taxa por cliente. Aumente o SIM_INTERVAL.

Para ver como isso se encaixa em uma arquitetura de planta, consulte o nosso guia de IoT industrial com MQTT.

Rode o seu simulador Sparkplug no ConnectMQ

Um simulador é mais útil quando roda contra o mesmo tipo de broker que você vai usar em produção. O ConnectMQ é um broker MQTT gerenciado com MQTT 3.1.1 e 5.0, TLS, decodificação de Sparkplug B e republicação ISA-95 embutidas no portal, e os mesmos dados simulados ajudam a validar um gêmeo digital antes de a linha real existir. Comece no plano gratuito Serverless em app.connectmq.com, sem cartão de crédito, e passe para um plano pago ou dedicado quando a sua planta simulada virar uma planta real.

Perguntas frequentes

Posso usar este simulador com o Ignition ou outros hosts SCADA?

Sim, desde que o host esteja configurado com o mesmo broker, o mesmo group ID e o namespace Sparkplug B. Comece pelos cenários de entrada tardia e rebirth, porque eles revelam a maioria dos problemas de compatibilidade.

Por que NBIRTH e DDATA são publicados com QoS 0?

A especificação Sparkplug exige QoS 0 para mensagens BIRTH e DATA e depende dos números de sequência e dos rebirths para a recuperação. Apenas o will NDEATH usa QoS 1.

Por que o Ctrl+C não dispara o will NDEATH?

Uma desconexão MQTT limpa descarta a mensagem will. É por isso que o script publica o NDEATH explicitamente no encerramento, antes de desconectar.

Posso simular mais de um device?

Sim. Publique um DBIRTH por device depois do NBIRTH, mantenha os aliases únicos em todo o edge node e publique o DDATA no topic de cada device.