Python ile Sparkplug B Çözme: Protobuf Payload'larını Adım Adım Okuyun

Python'da Sparkplug B çözmek için iki şeye ihtiyacınız var: resmi sparkplug_b.proto şemasından üretilmiş sınıflar ve her edge node'un birth mesajlarında neyi duyurduğunu hatırlayan bir MQTT subscriber. Bu eğitimde ikisini de oluşturacak, her Sparkplug mesajını okunabilir JSON olarak yazdıran bir betik elde edecek ve hızlıca yazılmış decoder'ların çoğunun yanlış yaptığı üç ayrıntıyı öğreneceksiniz.

Neye İhtiyacınız Var?

Henüz yayın yapan bir edge node'unuz yoksa, Sparkplug birth ve death sertifikaları yazımızdaki simülatörü çalıştırın. Herkese açık broker'da rastgele bir gruba NBIRTH, DBIRTH ve DDATA yayınlar; bu decoder'ın ihtiyaç duyduğu da tam olarak budur.

Adım 1: Python Sınıflarını Üretin

Bir proje klasörü ve sanal ortam oluşturun, ardından Eclipse Tahu projesinin yayınladığı şemadan sparkplug_b_pb2.py dosyasını üretin:

mkdir sparkplug-decoder && cd sparkplug-decoder
python -m venv .venv
source .venv/bin/activate        # Windows'ta: .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

grpcio-tools kullanmak, ayrıca kurulmuş bir protoc ile Python protobuf çalışma zamanı arasındaki sürüm uyumsuzluklarını önler. Daha sonra üretilen kodun daha yeni bir protoc ile oluşturulduğuna dair bir hata görürseniz, aynı ortamda protobuf paketini yükseltin ve sınıfları yeniden üretin.

Adım 2: Decoder'ın Neleri Ele Alması Gerektiğini Anlayın

Baytlar üzerinde ParseFromString çağırmak işin kolay kısmıdır. Kullanışlı bir decoder, Sparkplug'a özgü üç ayrıntıyla da başa çıkmak zorundadır:

1. Alias'lar. Edge node'lar çoğu zaman tam metrik adlarını yalnızca NBIRTH ve DBIRTH'te gönderir, ardından NDATA ve DDATA'da metriklere sayısal bir alias ile başvurur. Decoder, alias eşlemesini her node ve cihaz için saklamalıdır.
2. Veri tipleri yalnızca birth mesajlarında. Veri tipi birth mesajlarında zorunlu, veri mesajlarında ise isteğe bağlıdır; bu yüzden decoder her metriğin veri tipini de birth'ten hatırlamalıdır.
3. İşaretli tamsayılar. Int8, Int16 ve Int32 işaretsiz 32 bit bir alanda, Int64 ise işaretsiz 64 bit bir alanda taşınır. İkiye tümleyen ile dönüştürmezseniz -5 değeri 4294967291 olarak gelir.

Sparkplug B payload rehberi, bu kuralların arkasındaki yapıyı daha ayrıntılı açıklıyor.

Adım 3: Eksiksiz Decoder

Bunu decode_sparkplug.py olarak kaydedin:

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

Betiği simülatörünüzün yazdırdığı grup ID'siyle çalıştırın; örneğin python decode_sparkplug.py demo-3f9a1c. Herkese açık MQTT test broker'ında bir grup belirtmek nezaket gereğidir: herkese açık alan paylaşımlıdır ve herkes tarafından okunabilir, bu yüzden oradaki tüm gruplara abone olmak diğer kullanıcılardan gelen gürültü üretir.

Adım 4: Çıktıyı Okuyun

Simülatörden gelen bir DDATA mesajı şöyle yazdırılır:

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

Decoder'ı edge node'dan sonra başlatırsanız, alias kullanılan metrikler <unknown alias N> olarak görünür, çünkü decoder birth mesajını kaçırmıştır. Çözüm, node'a NBIRTH ve DBIRTH'ü yeniden yayınlatan bir rebirth komutu göndermektir. Üretimdeki host uygulamaları bunu otomatik olarak yapar; basit bir araç ise önce başlatılabilir.

Adım 5: Rebirth'ü Otomatik Olarak İsteyin

Pasif şekilde bekleyip bilinmeyen alias'ları yazdıran bir decoder, hızlı bir göz atma için yeterlidir. Bir veritabanını besleyen decoder ise kendi görünümünü onarmalıdır. Standart yöntem, Node Control/Rebirth metriği true olarak ayarlanmış bir NCMD mesajıdır. Bu yardımcı fonksiyonu ekleyin ve bir alias eksik olduğunda metric_to_dict içinden çağırın:

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

Client'ı metric_to_dict fonksiyonuna parametre olarak geçirin (veya modül seviyesinde bir değişken olarak tutun) ve NBIRTH geldiğinde requested kaydını temizleyin; böylece daha sonraki bir boşluk yeni bir isteği tetikleyebilir. Paylaşımlı ortamlarda buna dikkat edin: komut göndermek bir yazma işlemidir ve bir üretim tesisinde normalde yalnızca belirlenmiş host uygulaması rebirth istemelidir. Sparkplug topic namespace yazısı, hangi topic'lerin komut taşıdığını listeliyor.

Uzun Süre Çalışan Bir Decoder İçin Güvenilirlik İpuçları

Haftalarca çalışan bir decoder'ın ihtiyaçları, beş dakikalığına başlattığınız bir decoder'dan farklıdır:

• seq değerini her edge node için doğrulayın. Son değeri saklayın ve 255'ten 0'a dönüş dışındaki her boşluğu işaretleyin; bir boşluk, alias ve değer önbelleğinizin bayat olabileceği anlamına gelir.
• NBIRTH'ten gelen bdSeq değerini takip edin ve bir NDEATH'i yalnızca bdSeq'i eşleştiğinde geçerli sayın. Aksi takdirde eski bir oturumdan gelen gecikmiş bir will, canlı bir node'u çevrimdışı olarak işaretleyebilir.
• Son değeri güncelmiş gibi tutmak yerine, NDEATH'te bir node'un tüm metriklerini, DDEATH'te ise bir cihazın tüm metriklerini bayat olarak işaretleyin.
• on_message fonksiyonunu hızlı tutun. Bir veritabanına yazıyorsanız parse edin, ardından sonucu bir kuyruğa veya worker thread'e verin; böylece ağ döngüsü hiçbir zaman bloklanmaz.
• Logları küçük tutmak için çözme hatalarını tam payload yerine topic ve payload uzunluğuyla loglayın.

Adım 6: Bir Üretim Broker'ına Bağlanın

Herkese açık broker test içindir. Gerçek tesis verisi için kendi broker'ınızın kimlik bilgilerini ve TLS kullanın. Betiğin altındaki bağlantı satırlarını değiştirin:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # sistem CA deposunu kullanır; broker bir Let's Encrypt sertifikası kullanır
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 8883, keepalive=60)
client.loop_forever()

ConnectMQ'da client ID'leri portalda verilen kimlik bilgilerine bağlıdır; bu yüzden orada gösterilen client ID'yi birebir kullanın.

Sorun Giderme

Python ile Çözme ve Broker'da Yerleşik Çözme

Veriyi bir veritabanına, historian'a veya özel bir uygulamaya aktarmanız gerektiğinde kendi decoder'ınız doğru araçtır. Ancak günlük incelemelerde, yalnızca bir değeri görmek için kod yazmak yavaştır. ConnectMQ, Sparkplug B'yi (Protobuf ve JSON) broker içinde çözer; böylece portaldaki topic gezgini metrik adlarını ve değerlerini doğrudan gösterir. Ayrıca Sparkplug verisini okunabilir ISA-95 topic'lerine yeniden yayınlayabilir; örneğin spBv1.0/factory1/DDATA/line1/cnc1 topic'ini connectmq/factory1/production/line1/cnc1/telemetry topic'ine. Böylece aşağı akıştaki bir Python betiği, Node-RED akışı veya Telegraf agent'ı, proto dosyasını paketlemeden sıradan bir MQTT client olarak bu topic'lere abone olabilir.

Birçok ekibin kullandığı pratik bir ayrım şudur: devreye alma ve sorun giderme için yerleşik çözme, verinin saklanması veya dönüştürülmesi gerektiğinde ise yukarıdaki gibi bir Python tüketicisi. Veri tipi ayrıntıları için Sparkplug B veri tipleri referansını elinizin altında tutun; daha büyük mimari için MQTT ile Endüstriyel IoT rehberini okuyun.

ConnectMQ ile Başlayın

app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz bir Serverless broker oluşturun ve bu decoder'ı oluşturulan kimlik bilgilerinizle bağlayın. Testlerin ötesine ölçeklenmeye hazır olduğunuzda, tam yönetilen bir MQTT servisi veya bir bulut MQTT broker, kendi broker'ınızı işletmenin operasyonel yükünü ortadan kaldırır.

Sık Sorulan Sorular

Sparkplug B çözmek için bir pip paketi var mı?

Birkaç topluluk paketi mevcut, ancak resmi Tahu proto dosyasından sınıf üretmek en güvenilir yaklaşımdır, çünkü şema sürümünü siz kontrol edersiniz. grpcio-tools ile tek bir komut yeterlidir.

Proto dosyası olmadan Sparkplug B çözebilir miyim?

Anlamlı bir şekilde hayır. Protobuf mesajları ad yerine alan numaraları taşır; bu yüzden her alanın ne anlama geldiğini bilmek için şemaya ihtiyacınız vardır.

Decoder'ım neden bilinmeyen alias'lar gösteriyor?

Edge node metrik adlarını yalnızca birth mesajlarında gönderir. Decoder'ınız daha sonra başlarsa alias eşlemesini hiç öğrenmemiş olur; bir rebirth isteyin veya decoder'ı önce başlatın.

Bu decoder MQTT 5 ile çalışır mı?

Evet. Sparkplug payload'ları MQTT 3.1.1 ve 5.0 üzerinde aynıdır. MQTT 5 kullanmak için client'ı oluştururken protocol=mqtt.MQTTv5 parametresini geçirin.