Endüstriyel Veri için OPC UA'dan MQTT'ye Köprü

Bir OPC UA MQTT köprüsü, bir OPC UA sunucusuna istemci olarak bağlanır, ilgilendiğiniz tag'lere subscribe olur ve her veri değişimini iyi adlandırılmış bir topic'te MQTT mesajı olarak yeniden publish eder. Tesis OPC UA'yı güçlü olduğu yerde tutar (zengin modeller, güvenlik, makine entegrasyonu); dashboard'lar, historian'lar ve bulut servisleri ise serbestçe ölçekleyebilecekleri hafif bir publish/subscribe akışı alır.

Bu eğitim böyle bir köprüyü Python'da asyncua ve paho-mqtt ile kuruyor, OPC UA node id'lerini bir Unified Namespace topic ağacına eşliyor ve simüle bir OPC UA sunucusuyla nasıl test edeceğinizi gösteriyor.

Doğrudan bağlanmak yerine neden köprü

OPC UA istemci/sunucu modelidir: her tüketici ihtiyaç duyduğu her sunucuya bir oturum açar. Beş tüketici ve yirmi sunucuyla bu yüz oturum eder; her birinin kendi subscription'ları, sertifikaları ve sunucu üzerindeki yükü vardır. Köprü bunu bir hub'a dönüştürür:

Köprü varsayılan olarak tek yönlüdür: OPC UA girer, MQTT çıkar. Böylece BT veriyi alırken OT tarafı korunur. İki protokolün daha geniş bir karşılaştırması için MQTT ve OPC UA yazısını okuyun.

1. Bir test OPC UA sunucusu başlatın

Üretim sunucusuna karşı geliştirme yapmayın. Microsoft'un açık kaynak OPC PLC simülatörü Docker'da çalışır ve değişen değerler sunar:

docker run --rm -p 50000:50000 mcr.microsoft.com/iotedge/opc-plc:latest \
  --pn=50000 --autoaccept --unsecuretransport

Endpoint opc.tcp://localhost:50000'dir. UaExpert gibi bir grafik istemciyle ya da asyncua ile gelen uals komutuyla gezinin ve köprülemek istediğiniz node id'lerini not edin. Simülatör kendi namespace'inde StepUp, AlternatingBoolean ve RandomSignedInt32 gibi node'lar sunar; namespace index'ini tarayıcınızda kontrol edin, çünkü sürümler arasında farklı olabilir.

2. Node'ları topic'lere eşleyin

Asıl tasarım işi eşlemedir. Bir topic, değerin tesiste nerede yaşadığını söylemeli; bugün onu hangi sunucunun veya node id'nin sağladığını değil. Eşlemeyi konfigürasyonda tutun:

BASE = "testmqtt/acme/plant1/packaging/line1"

NODES = {
    "ns=3;s=StepUp":             f"{BASE}/filler/counter",
    "ns=3;s=AlternatingBoolean": f"{BASE}/filler/running",
    "ns=3;s=RandomSignedInt32":  f"{BASE}/capper/torque",
}

İleride simüle sunucuyu gerçeğiyle değiştirdiğinizde ya da bir tag'i başka bir PLC'ye taşıdığınızda yalnızca eşlemenin sol tarafı değişir. Tüm subscriber'lar çalışmaya devam eder. Bir Unified Namespace'in temel fikri budur: topic ağacı sözleşmedir, kaynaklar ise uygulama ayrıntısıdır.

3. Köprü

asyncua'nın subscription API'si bir monitored item her değiştiğinde bir handler çağırır. Handler değeri, OPC UA kalitesini ve kaynak zaman damgasını publish eder:

import asyncio, json, os
import paho.mqtt.client as mqtt
from asyncua import Client
from nodes import BASE, NODES

OPC_URL = os.getenv("OPC_URL", "opc.tcp://localhost:50000")
STATUS = f"{BASE}/_bridge/status"

mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="opcua-bridge-line1")
mq.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
mq.tls_set()
mq.will_set(STATUS, "offline", qos=1, retain=True)
mq.connect("broker.connectmq.com", 8883, keepalive=30)
mq.loop_start()

class Handler:
    def datachange_notification(self, node, val, data):
        dv = data.monitored_item.Value
        ts = dv.SourceTimestamp or dv.ServerTimestamp
        payload = {
            "v": val,
            "good": dv.StatusCode.is_good(),
            "ts": ts.isoformat() if ts else None,
        }
        topic = NODES[node.nodeid.to_string()]
        mq.publish(topic, json.dumps(payload, default=str), qos=1, retain=True)

async def run():
    while True:
        try:
            async with Client(url=OPC_URL, timeout=10) as opc:
                nodes = [opc.get_node(nid) for nid in NODES]
                sub = await opc.create_subscription(500, Handler())   # 500 ms publishing aralığı
                await sub.subscribe_data_change(nodes)
                mq.publish(STATUS, "online", qos=1, retain=True)
                print("köprülenen node sayısı:", len(nodes))
                while True:
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:          # bağlantı koptu, sunucu yeniden başladı, hatalı node id
            print("OPC UA hatası:", exc)
            mq.publish(STATUS, "opcua-unreachable", qos=1, retain=True)
            await asyncio.sleep(5)

asyncio.run(run())

pip install "asyncua>=1.1" "paho-mqtt>=2.0"
python opcua_mqtt_bridge.py

Bu tasarımın size kazandırdıkları:

• Değişim odaklı trafik. OPC UA yalnızca değerler değiştiğinde bildirim gönderir, dolayısıyla MQTT tarafında gereksiz örnek taşınmaz. Publishing aralığını (burada 500 ms) ve item başına sampling aralığını prosesinize göre ayarlayın.
• Kalite değerle birlikte taşınır. Kötü veya belirsiz bir OPC UA status code'u bir bilgidir; mesajı sessizce atmak yerine payload'da tutun.
• Kaynak zaman damgaları. Historian'lar değerin kaynakta ne zaman değiştiğini saklamalı, köprünün onu ne zaman ilettiğini değil.
• Retained değerler ve durum topic'i. Yeni subscriber'lar güncel durumu anında görür; süreç ölürse Last Will köprü durumunu offline yapar. Ayrıntılar için retained mesajlar.
• Otomatik toparlanma. OPC UA sunucusu yeniden başlarsa dış döngü yeniden bağlanır ve subscription'ı yeniden oluşturur. paho da loop_start() sonrasında broker'a kendiliğinden yeniden bağlanır.

4. Köprüyü test edin

1. Simülatörü ve köprüyü başlatın. Köprü, köprülenen node sayısını yazdırmalı.
2. Online MQTT istemcisini açın, broker.connectmq.com'a bağlanın ve testmqtt/acme/plant1/# topic'ine subscribe olun. filler/counter artmalı, filler/running değişmeli ve _bridge/status online göstermeli.
3. Simülatör container'ını durdurun. Birkaç saniye içinde durum opcua-unreachable olmalı; yeniden başlattığınızda değerler köprüyü yeniden başlatmadan geri gelmeli.
4. Köprüyü kill -9 ile öldürün. Keep-alive zaman aşımından sonra broker Last Will'i publish eder: offline.

Köprülenen OPC UA tag'lerini tarayıcıda izleyin

Tüketicilerin kullandığı topic filtrelerinin köprünün publish ettikleriyle eşleştiğini topic matcher ile kontrol edin ve büyük yapısal değerleri (diziler, extension object'ler) aklınızda tutun: paket boyutu hesaplayıcı, tek bir bildirimin ağda ne kadar yer tuttuğunu tahmin etmenize yardımcı olur.

Köprü kurmanın diğer yolları

Küçük bir Python köprüsü öğrenmek, birkaç sunucu ve eşleme üzerinde tam kontrol için idealdir. Daha büyük ortamlar için şunları değerlendirin:

Hangi aracı seçerseniz seçin kurallar aynıdır: topic'ler tesisi tanımlar, payload'lar değer + kalite + zaman damgası taşır, köprü kendi sağlığını duyurur ve OPC UA'ya geri yazmalar ayrı, bilinçli olarak güvenceye alınmış bir yoldur.

Üretim kontrol listesi

• Gerçek sunucularda OPC UA güvenliğini (Basic256Sha256 ile imzalama ve şifreleme) ve köprüye özel bir uygulama sertifikası kullanın; simülatördeki güvenliksiz ayar yalnızca yerel testler içindir.
• Sunucunun subscription başına monitored item limitlerini okuyun ve büyük tag listelerini birden fazla subscription'a bölün.
• Bir arızanın yerel kalması için alan veya hat başına bir köprü çalıştırın.
• Node→topic eşlemesini versiyonlayın ve kod gibi gözden geçirin.
• Köprünün durum topic'ini izleyin ve online olmadığında alarm üretin.

OPC UA MQTT köprüsü, OPC UA'yı en iyi yaptığı iş için tutarken diğer tüm sistemlere basit ve ölçeklenebilir bir akış sunmanızı sağlar. Node'ları tesis yapısındaki topic'lere eşleyin, değişim bildirimlerini kalite ve zaman damgasıyla publish edin, bir simülatör ve online istemciyle test edin; OT verileriniz tüm şirketin subscribe olabileceği bir şeye dönüşsün.

Sıkça sorulan sorular