Python ile Sparkplug B Edge Node Simülatörü Oluşturun

Bir Sparkplug B edge node simülatörü; bir PLC, gateway veya bakım penceresi beklemeden SCADA host'ları, historian'lar ve dashboard'lar geliştirip test etmenizi sağlar. Bu eğitimde, dakikalar içinde erişebileceğiniz gerçek bir MQTT broker'a karşı doğru bir NBIRTH, DBIRTH ve DDATA dizisi yayınlayan, bir NDEATH will'i kaydeden ve rebirth isteklerine yanıt veren bir simülatörü Python'da oluşturacaksınız.

Simülatörün Doğru Yapması Gerekenler

Bir Sparkplug tüketicisi, düz bir MQTT subscriber'dan daha katıdır. Bir host uygulaması her edge node için oturum durumunu takip eder ve kuralları çiğneyen veriyi reddeder veya yok sayar. Bu yüzden kullanışlı bir Sparkplug B edge node simülatörü yalnızca topic adlarını değil, protokol davranışını da yeniden üretmek zorundadır.

Bu kavramlardan herhangi biri sizin için yeniyse, önce Sparkplug birth ve death sertifikaları ve Sparkplug topic namespace yazılarını okuyun. Aşağıdaki simülatör her ikisini de izliyor.

Ön Koşullar

Python 3.9 veya üzeri ve iki pakete ihtiyacınız var:

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

Ayrıca resmi Sparkplug B Protobuf tanımından üretilmiş Python modülü olan sparkplug_b_pb2.py dosyasına ihtiyacınız var. Eclipse Tahu projesi, Python core klasöründe hazır bir kopya sunar; bunu betiğinizle aynı dizine indirin. Eclipse Tahu rehberimiz dosyanın nerede bulunduğunu ve protobuf sürümünüz hata verirse protoc ile nasıl yeniden üretileceğini açıklıyor.

Broker olarak 1883 portunda broker.connectmq.com adresindeki herkese açık MQTT test broker'ı ile başlayın. Boş olmayan, benzersiz herhangi bir client ID ile anonim bağlantıları kabul eder. Sparkplug topic'leri spBv1.0 ile başlamak zorunda olduğundan özel bir önek ekleyemezsiniz; bunun yerine trafiğinizin başkalarınınkiyle çakışmaması için demo-x91q gibi rastgele bir grup ID'si kullanın. Herkese açık broker herkes tarafından okunabilir ve client başına saniyede 10 mesajla sınırlıdır; bu, birkaç saniyede bir yayın yapan bir simülatör için fazlasıyla yeterlidir.

Sparkplug B Edge Node Simülatörü Kodu

Bunu edge_sim.py olarak kaydedin. Tek bir cihaza sahip bir edge node'u simüle eder: hız, sıcaklık, çalışma bayrağı ve sayaç içeren bir dolum makinesi.

import os
import random
import ssl
import time

import paho.mqtt.client as mqtt
import sparkplug_b_pb2

# --- yapılandırma ------------------------------------------------------------
BROKER = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
USERNAME = os.getenv("MQTT_USERNAME")          # özel bir ConnectMQ broker için ayarlayın
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 veri tipi kodları
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):
        """Değerleri değiştirir ve yalnızca değişen metrikleri döndürür."""
        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:
    # Düzgün kapanışta will gönderilmediği için NDEATH açıkça yayınlanır
    client.publish(topic("NDEATH"), node.death_payload(), qos=1).wait_for_publish(5)
    client.loop_stop()
    client.disconnect()

Betiği python edge_sim.py ile çalıştırın. Bir birth satırı ve ardından her iki saniyede bir DDATA satırı görmelisiniz.

Çıktıyı İzlemek

İkinci bir terminalde mesajların geldiğini doğrulayın. Payload'lar binary Protobuf olduğu için mosquitto_sub okunamaz baytlar yazdıracaktır, ancak ilk kontrol için topic'ler ve zamanlama yeterlidir:

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

Çözülmüş değerler için Sparkplug metrik alias'ları rehberimizdeki alias'tan haberdar decoder'ı kullanın ya da trafiği bir ConnectMQ broker'ının portalında açın; oradaki canlı topic gezgini Sparkplug B payload'larını okunabilir metriklere çözer. Death sertifikasını test etmek için Ctrl+C'ye basmak yerine simülatör sürecini kill -9 ile sonlandırın veya terminal penceresini kapatın. Keep-alive süresi dolduktan sonra broker NDEATH will'ini yayınlar ve bir host uygulaması her metriği bayat olarak işaretlemelidir.

Simülatörle Host'ları ve Dashboard'ları Test Etmek

Bir simülatörün amacı tüketici tarafını sınamaktır. Bir host'a veya dashboard'a gerçek bir tesisi emanet etmeden önce şu senaryoları uygulayın.

1. Soğuk başlatma. Önce host'u, ardından simülatörü başlatın. Host, node'u ve cihazı oluşturmalı ve dört metriğin hepsini göstermelidir.
2. Geç katılım. Önce simülatörü, bir dakika sonra host'u başlatın. Doğru çalışan bir primary host, birth mesajlarını hiç görmediği için bir NCMD rebirth gönderir; simülatör "rebirth requested" yazdırır.
3. Düzensiz ölüm. Süreci sonlandırın. NDEATH geldiğinde host node'u çevrimdışı olarak işaretlemelidir.
4. Yeniden bağlanma. Simülatörü yeniden başlatın. Uzun süre çalışan bir kurulumda, host'un her death'i doğru birth ile eşleştirebilmesi için bd_seq değerini oturumlar arasında artırın; basit betik her başlangıçta onu sıfırlar, bu test için kabul edilebilir.
5. Ölçek. Host'un çok sayıda edge node'u nasıl ele aldığını görmek için farklı SPB_NODE değerleriyle birkaç kopya çalıştırın. Herkese açık broker'da toplam mesaj oranını makul tutun; özel bir broker'da planınızın sınırları dahilinde daha fazla zorlayabilirsiniz.
6. Komut yolu. Host'tan bir DCMD gönderin ve on_message fonksiyonuna ulaştığını doğrulayın. Handler'ı, komut geldiğinde Running değerini değiştirecek şekilde genişletin.

Tam bir host değerlendiriyorsanız, Sparkplug primary host uygulaması yazısı bir host'un yayınladığı STATE mesajlarını ve Sparkplug 3.0'da edge node'ların birth göndermeden önce bunları bekleyecek şekilde neden yapılandırılabildiğini açıklıyor.

Herkese Açık Broker'dan Kendi Broker'ınıza Geçmek

Herkese açık broker deneyler içindir; grubunuza herkes abone olabilir. Gerçek veriye benzeyen her şey için özel bir broker kullanın. ConnectMQ'da kaydolun, bir plan seçin, bir broker oluşturun ve oluşturulan kullanıcı adını, parolayı ve client ID'yi kopyalayın. Ardından şunu çalıştırın:

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

8883 portu, Python'un varsayılan olarak güvendiği broker'ın Let's Encrypt sertifikasıyla TLS'i etkinleştirir. ConnectMQ'da client ID'leri verilen kimlik bilgilerine bağlıdır; bu yüzden portaldaki client ID'yi birebir kullanın. ConnectMQ ayrıca çözülmüş Sparkplug B verisini ISA-95 topic'lerine yeniden yayınlayabilir; böylece filler-01 için simüle edilmiş DDATA, düz JSON tüketicilerinin Protobuf'u anlamadan abone olabileceği okunabilir bir yol altında görünebilir.

Sorun Giderme

• ModuleNotFoundError: sparkplug_b_pb2: üretilen modül betiğin yanında veya PYTHONPATH üzerinde değil.
• TypeError: Descriptors cannot be created directly: üretilen modül, protobuf çalışma zamanınızın beklediğinden daha eski bir protoc ile oluşturulmuş. Yeniden üretin veya geçici çözüm olarak PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python ayarlayın.
• Bağlantı reddedildi, yetkisiz: özel bir broker'da kullanıcı adı, parola veya client ID portalın verdiğiyle eşleşmiyor.
• Host node'u gösteriyor ama değer yok: host DDATA'da ad gerektiriyor ve alias'ları desteklemiyor olabilir. Doğrulamak için dbirth içindeki alias= argümanlarını kaldırın ve ddata içinde adları geçirin.
• Herkese açık broker'da mesajlar düşüyor: client başına oranı aştınız. SIM_INTERVAL değerini artırın.

Bunun bir tesis mimarisine nasıl oturduğunu görmek için endüstriyel IoT MQTT rehberimize bakın.

Sparkplug Simülatörünüzü ConnectMQ'da Çalıştırın

Bir simülatör, en çok üretimde çalıştıracağınız türden bir broker'a karşı kullanıldığında işe yarar. ConnectMQ; MQTT 3.1.1 ve 5.0, TLS, Sparkplug B çözme ve ISA-95 yeniden yayınlamayı portala yerleşik olarak sunan tam yönetilen bir MQTT servisidir. app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz Serverless planıyla başlayın ve simüle ettiğiniz tesis gerçeğe dönüştüğünde ücretli veya dedicated bir plana geçin.

Sık Sorulan Sorular

Bu simülatörü Ignition veya başka SCADA host'larıyla kullanabilir miyim?

Evet, host aynı broker, grup ID'si ve Sparkplug B namespace ile yapılandırıldığı sürece. Uyumluluk sorunlarının çoğunu ortaya çıkardıkları için geç katılım ve rebirth senaryolarıyla başlayın.

NBIRTH ve DDATA neden QoS 0 ile yayınlanıyor?

Sparkplug spesifikasyonu BIRTH ve DATA mesajları için QoS 0 gerektirir ve kurtarma için sıra numaralarına ve rebirth'lere dayanır. Yalnızca NDEATH will'i QoS 1 kullanır.

Ctrl+C neden NDEATH will'ini tetiklemiyor?

Düzgün bir MQTT bağlantı kesme işlemi will mesajını iptal eder. Betiğin kapanışta bağlantıyı kesmeden önce NDEATH'i açıkça yayınlamasının nedeni budur.

Birden fazla cihaz simüle edebilir miyim?

Evet. NBIRTH'ten sonra cihaz başına bir DBIRTH yayınlayın, alias'ları tüm edge node genelinde benzersiz tutun ve DDATA'yı her cihazın kendi topic'inde yayınlayın.