OPC UA'dan MQTT'ye Gateway: PLC Verisini Buluta Köprülemek

Bir OPC UA'dan MQTT'ye gateway, tesis ağındaki PLC'lerden ve OPC UA sunucularından tag'leri okur ve bunları panoların, historian'ların ve bulut uygulamalarının abone olabileceği bir broker'a MQTT mesajları olarak yayımlar. Bu eğitimde açık kaynak kütüphanelerle Python'da bir tane kuracaksınız: OPC UA node'larına abone olur, bunları ISA-95 tarzı topic'lere eşler ve bir bulut broker'a temiz JSON yayımlar. Her şeyi yaklaşık on beş dakikada bir dizüstü bilgisayarda çalıştırabilirsiniz.

OPC UA'yı MQTT'ye Köprülemek Neden Gerekli?

OPC UA fabrika sahasında mükemmeldir. Zengin bir bilgi modeline, yerleşik güvenliğe ve PLC üreticileri arasında geniş desteğe sahiptir. Ama bir client/server protokolüdür: her tüketici her sunucuya kendi oturumunu açar. Bu, on PLC ile konuşan bir SCADA sistemi için işe yarar; bazıları bulutta olan yirmi uygulama aynı veriyi istediğinde ise zahmetli hâle gelir.

MQTT bu ilişkiyi tersine çevirir. Gateway sunucu başına bir OPC UA oturumu ve tek bir dışa giden MQTT bağlantısı tutar; istenen sayıda tüketici de broker'a abone olur. Tesis güvenlik duvarının yalnızca 8883 portunda dışa giden bir bağlantıya ihtiyacı vardır ve PLC'ler hiçbir zaman bulut trafiği görmez. İki protokolü daha geniş açıdan tartıyorsanız OPC UA ve MQTT karşılaştırmamız ödünleşimleri ele alıyor; kısaca, çoğu mimari ikisini birlikte kullanır.

Gateway'in Mimarisi

 PLC / OPC UA server            Edge gateway (this tutorial)              MQTT broker
 opc.tcp://plc:4840   ---->  asyncua subscription  ---> paho-mqtt  ---->  broker.connectmq.com:8883
 (tags, monitored items)     map node -> ISA-95 topic   JSON payload       dashboards, historian, apps

Gateway'in dört sorumluluğu vardır:

1. Sorgulamayın, abone olun. OPC UA subscription'ları, sunucunun değer değişikliklerini sizin seçtiğiniz bir örnekleme aralığında göndermesini sağlar; bu, her tag'i bir zamanlayıcıyla okumaktan çok daha verimlidir.
2. Node'ları topic'lere eşleyin. ns=2;s=Line1.CNC1.SpindleTemp gibi node ID'leri bir bulut geliştiricisi için bir şey ifade etmez. Gateway bunları okunabilir bir hiyerarşiye çevirir.
3. Payload'ları normalleştirin. Her mesaj değeri, OPC UA durumunu, bir kaynak zaman damgasını ve bir mühendislik birimini içerir.
4. Kopmalara dayanın. İki taraf da bir noktada kopacaktır; gateway yapılandırmasını kaybetmeden yeniden bağlanmalıdır.

Ön Koşullar

• Python 3.10 veya daha yenisi.
• asyncua (açık kaynak FreeOpcUa client kütüphanesi) ve paho-mqtt 2.x.
• Bir OPC UA sunucusu. Elinizde bir PLC yoksa eğitimde küçük bir simüle sunucu bulunuyor.
• Bir MQTT broker. Test için broker.connectmq.com adresindeki public MQTT test broker'ı kullanın; gerçek tesis verisi için barındırılan bir MQTT broker'dan alınan özel kimlik bilgilerini kullanın.

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

1. Adım: Simüle Bir OPC UA Sunucusu (İsteğe Bağlı)

Zaten bir OPC UA sunucunuz varsa bu adımı atlayın ve endpoint URL'sini ve node ID'lerini not edin. Aksi takdirde bu betik opc.tcp://localhost:4840 üzerinde değişen üç tag sunar.

# sim_server.py - değişen CNC değerlerine sahip küçük bir OPC UA sunucusu
import asyncio
import math
import random
from asyncua import Server, ua

async def main():
    server = Server()
    await server.init()
    server.set_endpoint("opc.tcp://0.0.0.0:4840/sim/")
    idx = await server.register_namespace("urn:demo:cnc")

    cnc = await server.nodes.objects.add_object(ua.NodeId("Line1.CNC1", idx), "CNC1")
    temp = await cnc.add_variable(ua.NodeId("Line1.CNC1.SpindleTemp", idx), "SpindleTemp", 40.0)
    speed = await cnc.add_variable(ua.NodeId("Line1.CNC1.SpindleSpeed", idx), "SpindleSpeed", 0.0)
    count = await cnc.add_variable(ua.NodeId("Line1.CNC1.PartCount", idx), "PartCount", 0)

    async with server:
        t = 0
        while True:
            t += 1
            await temp.write_value(round(45 + 5 * math.sin(t / 20) + random.random(), 2))
            await speed.write_value(float(random.choice([0, 8000, 12000])))
            if t % 10 == 0:
                await count.write_value(await count.read_value() + 1)
            await asyncio.sleep(1)

if __name__ == "__main__":
    asyncio.run(main())

Node ID'leri, kaydedilen namespace'te (burada indeks 2) string tanımlayıcılar kullandığından ns=2;s=Line1.CNC1.SpindleTemp vb. şeklinde görünür. Gerçek bir PLC'de tam node ID'lerini bulmak için UaExpert gibi bir client aracıyla adres alanında gezinin.

2. Adım: Tag Haritası

Eşlemeyi kodda değil yapılandırmada tutun. Her girdi bir OPC UA node'unu, tüketicilerin görmesini istediğiniz topic yolu ve birimle eşleştirir. Topic düzeni ISA-95 tarzı bir hiyerarşiyi (saha, alan, hat, hücre) izler; ISA-95 MQTT topic'leri rehberimiz bunu derinlemesine açıklıyor.

{
  "opcua_endpoint": "opc.tcp://localhost:4840/sim/",
  "topic_root": "connectmq/demo/a91c/plant1",
  "tags": [
    { "node": "ns=2;s=Line1.CNC1.SpindleTemp",  "path": "machining/line1/cnc1/spindle_temp",  "unit": "degC" },
    { "node": "ns=2;s=Line1.CNC1.SpindleSpeed", "path": "machining/line1/cnc1/spindle_speed", "unit": "rpm" },
    { "node": "ns=2;s=Line1.CNC1.PartCount",    "path": "machining/line1/cnc1/part_count",    "unit": "count" }
  ]
}

tags.json olarak kaydedin. Topic'lerinizin public broker'da başkalarınınkiyle çakışmaması için a91c değerini kendi rastgele dizenizle değiştirin. Ortaya çıkan topic'ler sekiz seviye derinliğindedir; bu, public broker'ın izin verdiği maksimumdur. Özel bir broker'da daha fazla alanınız olur.

3. Adım: OPC UA'dan MQTT'ye Gateway Kodu

# gateway.py - OPC UA subscription -> MQTT publish
import asyncio
import json
import logging
import ssl
import uuid
from datetime import timezone

import paho.mqtt.client as mqtt
from asyncua import Client

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("gateway")

CONFIG = json.load(open("tags.json"))
MQTT_HOST = "broker.connectmq.com"
MQTT_PORT = 8883
MQTT_USER = None          # üretim: "YOUR_USERNAME"
MQTT_PASS = None          # üretim: "YOUR_PASSWORD"
CLIENT_ID = f"opcua-gw-{uuid.uuid4().hex[:8]}"   # üretim: "YOUR_CLIENT_ID"
STATUS_TOPIC = f"{CONFIG['topic_root']}/gateway/status"


def build_mqtt() -> mqtt.Client:
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                         protocol=mqtt.MQTTv5)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
    client.will_set(STATUS_TOPIC, json.dumps({"online": False}), qos=1, retain=True)

    def on_connect(client, userdata, flags, reason_code, properties):
        log.info("MQTT connected: %s", reason_code)
        client.publish(STATUS_TOPIC, json.dumps({"online": True}), qos=1, retain=True)

    def on_disconnect(client, userdata, flags, reason_code, properties):
        log.warning("MQTT disconnected: %s", reason_code)

    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    client.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
    client.loop_start()
    return client


class DataChangeHandler:
    """asyncua, izlenen her öğe değişikliğinde datachange_notification'ı çağırır."""

    def __init__(self, mqtt_client, node_map):
        self.mqtt = mqtt_client
        self.node_map = node_map

    def datachange_notification(self, node, val, data):
        tag = self.node_map[node.nodeid.to_string()]
        dv = data.monitored_item.Value
        ts = dv.SourceTimestamp or dv.ServerTimestamp
        payload = {
            "value": val,
            "unit": tag["unit"],
            "quality": dv.StatusCode.name,
            "ts": ts.replace(tzinfo=timezone.utc).isoformat() if ts else None,
        }
        topic = f"{CONFIG['topic_root']}/{tag['path']}"
        self.mqtt.publish(topic, json.dumps(payload), qos=1, retain=True)
        log.info("%s -> %s", topic, payload["value"])


async def run_opcua(mqtt_client):
    node_map = {t["node"]: t for t in CONFIG["tags"]}
    while True:
        try:
            async with Client(url=CONFIG["opcua_endpoint"]) as opc:
                log.info("OPC UA connected to %s", CONFIG["opcua_endpoint"])
                handler = DataChangeHandler(mqtt_client, node_map)
                sub = await opc.create_subscription(500, handler)  # yayımlama aralığı, ms
                nodes = [opc.get_node(n) for n in node_map]
                await sub.subscribe_data_change(nodes)
                while True:                       # oturumu canlı tut, arızaları algıla
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:                  # sunucu yeniden başlatma, ağ kaybı vb.
            log.error("OPC UA error: %s - retrying in 5 s", exc)
            await asyncio.sleep(5)


if __name__ == "__main__":
    asyncio.run(run_opcua(build_mqtt()))

Birkaç tasarım kararı özellikle belirtilmeye değer:

• Retained message'lar, sonradan bağlanan bir panonun her tag'in son değerini hemen görmesi anlamına gelir.
• Durum topic'indeki Last Will, gateway temiz bir bağlantı kesme olmadan çökerse {"online": false} değerine döner; böylece tüketiciler bayat veriyi canlı veriden ayırt edebilir.
• OPC UA durum kodu her değerle birlikte taşınır. Good bir okuma ile BadCommunicationError bir okuma akış aşağısında asla aynı görünmemelidir.
• check_connection() oturum kaybolduğunda hata fırlatır; bu da bizi yeniden deneme döngüsüne düşürür ve subscription'ı yeniden kurar.

4. Adım: Çalıştırın ve Doğrulayın

İki terminalde önce simüle sunucuyu, ardından gateway'i başlatın:

python sim_server.py
python gateway.py

Üçüncü bir terminalde kökünüzün altındaki her şeye abone olun:

mosquitto_sub -h broker.connectmq.com -p 1883 -i opcua-viewer-a91c \
  -t "connectmq/demo/a91c/plant1/#" -v

Şuna benzer satırlar görmelisiniz:

connectmq/demo/a91c/plant1/machining/line1/cnc1/spindle_temp {"value": 47.31, "unit": "degC", "quality": "Good", "ts": "2026-10-16T09:12:04.118000+00:00"}
connectmq/demo/a91c/plant1/gateway/status {"online": true}

Üretime Geçiş

Public broker herkes tarafından okunabilir; bu yüzden yalnızca örnek veri içindir. Gerçek PLC verisi için özel bir broker oluşturun, ardından MQTT_USER, MQTT_PASS ve CLIENT_ID değerlerini portaldaki kimlik bilgileriyle ayarlayın. ConnectMQ client ID'leri verilen kimlik bilgilerine bağlar; bu nedenle rastgele bir ID yerine oluşturulan client ID'yi kullanın.

Gateway'i bir tesis ağına koymadan önce kontrol listesi:

ConnectMQ üzerinde portalın canlı topic explorer'ı ISA-95 ağacınızı doldukça gösterir; cihaz görünümü de makineleri ISA-95 seviyelerine göre gruplar. Bu, topic eşlemenizin hatlar arasında tutarlı olduğunu doğrulamanın hızlı bir yoludur. Eşik uyarı kuralları da örneğin ek kod yazmadan bir sınırın üzerindeki iş mili sıcaklığını işaretleyebilir. Bu gateway'in büyük resimde nereye oturduğunu görmek için endüstriyel IoT MQTT rehberimize bakın.

Sorun Giderme

• BadIdentityTokenRejected veya BadUserAccessDenied: sunucu anonim erişime izin vermiyor. Client.set_user() ve set_password() ile bir kullanıcı geçirin veya sertifikaları yapılandırın.
• BadNodeIdUnknown: node ID dizesi yanlış; çoğu zaman namespace indeksi. Namespace indeksleri sunucu sürümleri arasında değişebilir; başlangıçta get_namespace_index() ile namespace URI'sine göre çözümleyin.
• Değerler geliyor ama zaman damgaları null: bazı sunucular kaynak zaman damgası ayarlamaz. Kod, sunucu zaman damgasına geri döner.
• MQTT bağlanıyor, ardından hemen kopuyor: public broker'da client ID'nizin benzersiz ve boş olmadığını kontrol edin. Özel bir broker'da client ID'nin verilen kimlik bilgileriyle eşleştiğini kontrol edin.
• Topic'ler public broker'da reddediliyor: sekiz topic seviyesini aştınız. Kökü veya yolu kısaltın.

OPC UA Gateway'inizi ConnectMQ'da Çalıştırın

app.connectmq.com adresinde kredi kartı gerektirmeden Serverless planda ücretsiz bir broker oluşturun. Oluşturulan kullanıcı adını, şifreyi ve client ID'yi gateway.py içine kopyalayın; dakikalar içinde PLC tag'leriniz panolar ve historian'lar için hazır, TLS korumalı özel bir bulut MQTT broker'da görünsün.

Sık Sorulan Sorular

OPC UA'dan MQTT'ye gateway için ticari bir ürüne ihtiyacım var mı?

Hayır. asyncua ve paho-mqtt gibi açık kaynak kütüphaneler birçok proje için yeterlidir. Ticari gateway'ler arayüz, store and forward ve üretici desteği ekler; bunlar ölçek büyüdükçe önem kazanabilir.

Gateway OPC UA tag'lerini sorgulamalı mı, yoksa subscription mı kullanmalı?

Subscription kullanın. Sunucu değerleri örnekler ve yalnızca değişiklikleri gönderir; bu, döngüsel okumalara kıyasla PLC ve ağ üzerindeki yükü azaltır.

Gateway PLC'ye değer geri yazabilir mi?

Teknik olarak evet; bir komut topic'ine abone olup write_value çağırarak. Ancak bunu ayrı ve dikkatle güvenceye alınmış bir özellik olarak ele alın. Çoğu kurulum gateway'i salt okunur tutar.

PLC tag verisi için hangi MQTT QoS seviyesini kullanmalıyım?

QoS 1 olağan tercihtir: iletim en az bir kez garanti edilir ve ek yük düşüktür. Yeni tüketicilerin güncel durumu hemen alması için retained message'larla birleştirin.