Allen-Bradley MQTT: ControlLogix ve CompactLogix Tag'lerini Yayınlama

Allen-Bradley MQTT entegrasyonu, ControlLogix ve CompactLogix tag verilerini SCADA sunucunuza yük eklemeden panolara, historian sistemlerine ve bulut analitiğine taşımanın en hızlı yoludur. Bu eğitimde Logix tag'lerini pycomm3 ile EtherNet/IP üzerinden okuyan, bunları ISA-95 topic'lerine eşleyen ve yalnızca anlamlı değişiklikleri bir MQTT broker'a yayınlayan küçük bir Python gateway kuracaksınız. Önce herkese açık bir broker'da test edecek, ardından TLS'li özel bir broker'a geçeceksiniz.

Allen-Bradley PLC'ler MQTT ile Nasıl Uyum Sağlar?

Logix kontrolörleri, CIP (Common Industrial Protocol) kullanarak EtherNet/IP üzerinden yerel olarak iletişim kurar. Studio 5000 projeleri veriyi, kontrolör kapsamlı veya program kapsamlı adlandırılmış tag'ler halinde; atomik tipler (BOOL, DINT, REAL), diziler ve kullanıcı tanımlı tipler (UDT) olarak düzenler. Bu adlandırılmış tag modeli MQTT entegrasyonu için bir nimettir: Modbus'tan farklı olarak bir register haritasına ihtiyacınız yoktur, çünkü tag adı zaten anlam taşır.

Bu tag'leri MQTT'ye taşımanın üç yaygın yolu vardır:

Bu eğitim, tüm hareketli parçaları gösterdiği için ilk yaklaşımı kullanıyor. Daha sonra ticari bir sürücüye veya MQTT ile Ignition çözümüne geçerseniz de aynı topic ve payload tasarımı geçerlidir.

Ön Koşullar

• Gateway'den TCP 44818 portu üzerinden erişilebilen bir ControlLogix veya CompactLogix kontrolör.
• Python 3.9 veya üzeri yüklü bir Linux veya Windows makine.
• Okunacak birkaç tag; örneğin Line1_Temp (REAL), Line1_Speed (DINT), Line1_Running (BOOL) ve program kapsamlı bir Program:MainProgram.Pump1_Pressure (REAL) tag'i.
• Test için mosquitto client'ları.

Kütüphaneleri kurun:

python -m venv abmqtt
source abmqtt/bin/activate
pip install pycomm3 "paho-mqtt>=2.0"

Adım 1: Tag Listesini Keşfedin

Gateway'i yazmadan önce bağlantıyı doğrulayın ve tag'lerin tam olarak nasıl adlandırıldığını görün. pycomm3 bağlandığında kontrolörün tag veritabanını yükler:

from pycomm3 import LogixDriver

# CompactLogix: yalnızca IP. CPU'su slot 3'te olan ControlLogix: "192.168.1.20/3"
PLC_PATH = "192.168.1.20"

with LogixDriver(PLC_PATH) as plc:
    print(plc.info)                       # kontrolör adı, revizyon, seri numarası
    for name, definition in plc.tags.items():
        print(name, definition["data_type_name"], definition.get("dim", 0))

Program kapsamlı tag'ler varsayılan olarak kontrolör kapsamlı listede yer almaz; okurken onları Program:MainProgram.TagName şeklinde adresleyin. ControlLogix şasilerinde eğik çizgiden sonraki yol, kontrolörün backplane'deki slotudur. Bunu yanlış girerseniz bağlantı Ethernet modülüne açılır ama tag okumaları başarısız olur.

pycomm3 dokümantasyonu yol sözdizimini, dizileri ve UDT işlemeyi ayrıntılı olarak ele alıyor.

Adım 2: ISA-95 Topic'lerini Tasarlayın

Her tag'i kontrolörü değil ekipmanı tanımlayan bir topic'e eşleyin. Gateway'deki basit bir eşleme tablosu, PLC adlandırma kurallarını MQTT isim alanınızın dışında tutar:

Bir saha ve alan önekiyle tam topic plant1/assembly/line1/oven/temperature olur. Tesis genelinde bir isim alanının işe yaramasını sağlayan bu tür tutarlı yollardır; seviyeler ve adlandırma hakkında daha derin bir anlatım için ISA-95 MQTT topic'leri yazısına bakın.

Adım 3: Allen-Bradley MQTT Gateway

Gateway, eşlenen tüm tag'leri her saniye tek bir çoklu tag isteğiyle okur, tag başına ölü bant uygular, değişiklikleri QoS 1 ile yayınlar, beş dakikada bir her şeyi yeniler ve Last Will içeren retained bir çevrimiçi/çevrimdışı durumu korur.

import json
import time
import uuid
from datetime import datetime, timezone

from pycomm3 import LogixDriver
import paho.mqtt.client as mqtt

PLC_PATH = "192.168.1.20"

# Önce herkese açık test broker'ı. Herkese açık alan herkes tarafından okunabilir: yalnızca test verisi.
BROKER, PORT, USE_TLS = "broker.connectmq.com", 1883, False
USERNAME, PASSWORD = None, None                  # üretim: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"ab-gw-{uuid.uuid4().hex[:8]}"      # üretim: "YOUR_CLIENT_ID"
BASE = f"connectmq/demo/{uuid.uuid4().hex[:6]}/plant1/assembly/line1"

TAGS = {
    "Line1_Temp": ("oven/temperature", "degC", 0.5),
    "Line1_Speed": ("conveyor/speed", "rpm", 5),
    "Line1_Running": ("line/running", None, 0),
    "Program:MainProgram.Pump1_Pressure": ("pump1/pressure", "bar", 0.05),
}


def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT connected:", reason_code)
    client.publish(f"{BASE}/gateway/status", "online", qos=1, retain=True)


def should_publish(tag, value, last, deadband):
    if tag not in last:
        return True
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return value != last[tag]
    return abs(value - last[tag]) >= deadband


def main():
    mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        mq.username_pw_set(USERNAME, PASSWORD)
    if USE_TLS:
        mq.tls_set()
    mq.will_set(f"{BASE}/gateway/status", "offline", qos=1, retain=True)
    mq.on_connect = on_connect
    mq.connect(BROKER, PORT, keepalive=30)
    mq.loop_start()
    print("Publishing under", BASE)

    plc = LogixDriver(PLC_PATH)
    last, last_full = {}, 0.0

    while True:
        try:
            if not plc.connected:
                plc.open()
            results = plc.read(*TAGS.keys())
            now = time.time()
            force = now - last_full > 300
            ts = datetime.now(timezone.utc).isoformat(timespec="seconds")

            for result in results:
                suffix, unit, deadband = TAGS[result.tag]
                if result.error:
                    mq.publish(f"{BASE}/{suffix}", json.dumps(
                        {"value": None, "quality": "bad", "error": result.error, "ts": ts}), qos=1)
                    continue
                value = result.value
                if isinstance(value, float):
                    value = round(value, 3)
                if force or should_publish(result.tag, value, last, deadband):
                    payload = {"value": value, "quality": "good", "ts": ts}
                    if unit:
                        payload["unit"] = unit
                    mq.publish(f"{BASE}/{suffix}", json.dumps(payload), qos=1)
                    last[result.tag] = value
            if force:
                last_full = now
        except Exception as exc:
            print("PLC communication failed:", exc)
            try:
                plc.close()
            except Exception:
                pass
            time.sleep(5)
        time.sleep(1)


if __name__ == "__main__":
    main()

Vurgulanmaya değer birkaç tasarım noktası:

• Tek bir çoklu tag okuması. pycomm3 birden fazla tag'i mümkün olan en az sayıda CIP isteğine paketler; bu, her tag'i ayrı ayrı okumaktan kontrolör için çok daha naziktir.
• Tag başına hatalar. Bir tag adındaki yazım hatası yalnızca o tag için hata döndürür. quality: bad yayınlamak, bir değeri sessizce dondurmak yerine sorunu görünür kılar.
• UDT'ler de çalışır. Bir UDT okumak bir Python sözlüğü döndürür ve json.dumps bunu yapılandırılmış bir payload'a dönüştürür. UDT payload'larını küçük tutun ve büyük dizileri her saniye yayınlamaktan kaçının.

Adım 4: Herkese Açık Broker'da Doğrulayın

Gateway'i başlatın ve önekine abone olun:

mosquitto_sub -h broker.connectmq.com -p 1883 -i ab-watch-01 \
  -t 'connectmq/demo/+/plant1/assembly/line1/#' -v

Retained gateway durumunu ve ilk tam değer setini hemen görmelisiniz. Studio 5000'den Line1_Running değerini değiştirin; değişiklik bir saniye içinde gelir. Değerler hassas olmadığı sürece bu adım için herkese açık MQTT test broker'ını kullanmak sorun değildir.

Adım 5: Özel Bir Broker'da Üretim

Üretim için ConnectMQ portalında özel bir broker oluşturun ve dört satırı değiştirin: port 8883, USE_TLS = True, verilen kullanıcı adı ve parola ve verilen client ID. Topic'lerin kuruluş veya saha seviyesinden başlaması için connectmq/demo/... önekini kaldırın. Client ID'ler kimlik bilgilerine bağlı olduğundan, sahalar arasında tek bir set paylaşmak yerine her gateway'e kendi setini verin.

Otomatik olarak yeniden başlaması için gateway'i bir servis olarak çalıştırın:

[Unit]
Description=Allen-Bradley MQTT gateway
After=network-online.target

[Service]
ExecStart=/opt/abmqtt/bin/python /opt/abmqtt/gateway.py
Restart=always
RestartSec=5
User=abmqtt

[Install]
WantedBy=multi-user.target

Veri akmaya başladığında ConnectMQ portalının cihazlar görünümü ISA-95 topic'lerini saha, alan ve hatta göre gruplar; eşik tabanlı alarm kuralları da sınırlarının dışındaki bir sıcaklığı veya basıncı işaretleyebilir. Historian'lar ve panolar, MQTT broker için diğer client'lar gibi abone olur.

Okuma Hızlarını ve Ölü Bantları Ayarlamak

Okuma aralığı ve ölü bantlar birlikte gateway'in hem kontrol ağında hem de broker'da ne kadar trafik oluşturacağını belirler. Bunları ayarlamanın pratik bir yolu:

• Tüketiciden başlayın. Bir operatörün birkaç saniyede bir yenilediği bir pano 100 ms'lik veriye ihtiyaç duymaz. Enerji raporları için kullanılan bir historian bir dakikalık ortalamalarla yetinebilir.
• Ölü bantları sensör gürültüsüne göre boyutlandırın. Bir değeri Studio 5000 trend görünümünde birkaç dakika izleyin. Gürültü bandından biraz daha geniş bir ölü bant, gerçek değişiklikleri gizlemeden gereksiz dalgalanmaları bastırır.
• Durumlara ve alarmlara asla ölü bant uygulamayın. Boolean'lar, mod word'leri ve alarm bitleri her değişiklikte yayınlanmalıdır; bunların topic'leri, yeni bir abonenin mevcut durumu hemen görmesi için retained mesajlara iyi adaydır.
• Periyodik yenilemeyi koruyun. O olmadan, saatlerce gerçekten sabit kalan bir değer, sessizce durmuş bir gateway ile aynı görünür.

İleride tek bir gateway yetersiz kalırsa, tüm tesis için dev bir süreç yerine alan veya hat başına bir gateway çalıştırın. Arızalar sınırlı kalır ve her örneğin küçük, gözden geçirilebilir bir tag listesi olur.

Güvenlik Hususları

EtherNet/IP üzerinden CIP çoğu kurulumda kimlik doğrulama içermez; bu yüzden gateway fiilen güvenilen bir mühendislik client'ıdır. Onu bir güvenlik duvarının arkasında kontrol ağına yerleştirin, ondan yalnızca broker'a giden TLS trafiğine izin verin ve açık bir izin listesi, aralık kontrolleri ve PLC mantığında bir kilitleme (interlock) olmadan MQTT komutlarının PLC tag'lerine yazmasına asla izin vermeyin. Okumayı otomatikleştirmek güvenlidir; yazma ise bir tasarım incelemesini hak eder. EtherNet/IP MQTT dönüşümü rehberi CIP trafiği için ağ segmentasyonunu daha ayrıntılı ele alıyor, MQTT ile endüstriyel IoT rehberi ise gateway'lerin eksiksiz bir mimarideki yerini açıklıyor.

Sorun Giderme

1. Bağlantı zaman aşımına uğruyor. Yönlendirmeyi ve gateway ile kontrolör arasında TCP 44818'in açık olduğunu kontrol edin.
2. Bağlandı ama tüm tag'ler başarısız. ControlLogix yolunda yanlış slot ya da tag program kapsamlı ve Program: önekine ihtiyaç duyuyor.
3. Uzun ondalıklı REAL değerler. Bu normal float gösterimidir; yayınlamadan önce yuvarlayın.
4. MQTT birkaç saniyede bir bağlantıyı kesiyor. Başka bir client aynı client ID'yi kullanıyor.
5. Yüksek kontrolör iletişim yükü. Okuma aralığını artırın veya tag listesini azaltın; çoğu pano saniye altı veriye ihtiyaç duymaz.

Logix Kontrolörlerinizi ConnectMQ'ya Bağlayın

Gateway'iniz hazır; ona yayın yapacağı güvenli bir bulut MQTT broker verin. Kredi kartı gerektirmeyen ücretsiz Serverless plan için app.connectmq.com adresinden kaydolun, üretilen kimlik bilgilerini betiğe yapıştırın ve Allen-Bradley tag'lerinizin canlı olarak belirmesini izleyin.

Sık Sorulan Sorular

Bir ControlLogix PLC yerel olarak MQTT yayınlayabilir mi?

Logix kontrolörlerinin kendisinde genel amaçlı bir MQTT client bulunmaz. Olağan yaklaşım, tag'leri EtherNet/IP üzerinden okuyup yayınlayan bir edge gateway, bağlantı sunucusu veya SCADA modülüdür.

pycomm3, Micro800 veya SLC 500 ile çalışır mı?

pycomm3'ün LogixDriver'ı bazı sınırlamalarla ControlLogix, CompactLogix ve Micro800'ü hedefler; eski PLC-5 ve SLC 500 ayrı bir sürücü kullanır. Aile başına desteklenen özellikler için kütüphane dokümantasyonuna bakın.

Tag'leri ne sıklıkta okumalıyım?

Çoğu izleme için bir saniye fazlasıyla yeterlidir. Yalnızca gerçek değişikliklerin yayınlanması için ölü bantlar ve buna ek olarak periyodik tam yenileme kullanın.

JSON yerine Sparkplug B mi kullanmalıyım?

Sparkplug B, birth ve death sertifikaları ile kompakt Protobuf kodlaması ekler; bu da çok sayıda tag ve SCADA entegrasyonunda işe yarar. Düz JSON ise başlangıç için daha basittir ve incelemesi kolaydır.