PLC'ler için Modbus'tan MQTT'ye Gateway: Python ile Modbus TCP Poller

Bir Modbus MQTT gateway, bir PLC'nin register'larını Modbus TCP üzerinden okur ve MQTT mesajları olarak publish eder; böylece dashboard'lar, historian'lar ve bulut servisleri PLC'ye hiç dokunmadan tesis verisine subscribe olabilir. pymodbus ve paho-mqtt ile yaklaşık 150 satır Python'da güvenilir bir gateway yazabilirsiniz.

Bu eğitim gateway'i adım adım kuruyor: bir tag haritası, okuma döngüsü, 32 bit değerlerin çözümlenmesi, değişimde publish (report by exception), Last Will ile online/offline durumu ve üretim hattına yaklaşmadan her şeyi test etmenin bir yolu.

Gateway nereye oturur

Modbus istek/yanıt protokolüdür: bir master (istemci) bir slave'den (sunucu) register ister, her seferinde bir istek. MQTT ise publish/subscribe'dır: üreticiler bir kez publish eder, istenen sayıda tüketici subscribe olur. Gateway iki dünya arasında çeviri yapar:

Gateway'i iki ağa da erişebilen bir edge cihazında çalıştırın (endüstriyel PC, Raspberry Pi, hat sunucusunda bir container). PLC hiçbir zaman MQTT trafiği görmez, MQTT tüketicileri de hiçbir zaman Modbus soketi açmaz; OT ekiplerinin istediği ayrım tam olarak budur. Genel resim için IIoT ve Endüstri 4.0 için MQTT yazısına bakın.

1. Tag'lerinizi tanımlayın

Modbus'ta isim yoktur, yalnızca adres vardır. İlk iş, her register'ın ne anlama geldiğini söyleyen bir tag haritasıdır. Konfigürasyon gibi gözden geçirilebilmesi için kodun yanında bir dosyada tutun:

# isim, fonksiyon, adres, veri tipi, ölçek, birim
TAGS = [
    ("temperature",  "holding", 0,  "float32", 1.0,  "degC"),
    ("pressure",     "holding", 2,  "float32", 1.0,  "bar"),
    ("motor_speed",  "holding", 4,  "uint16",  1.0,  "rpm"),
    ("part_count",   "holding", 5,  "uint32",  1.0,  "pcs"),
    ("tank_level",   "input",   0,  "int16",   0.1,  "%"),
    ("running",      "coil",    0,  "bool",    1,    ""),
]

Buradaki adresler sıfırdan başlayan protokol adresleridir. Üretici dokümanları genellikle birden başlayan "40001" numaralandırmasını kullanır: holding register 40001 adres 0'dır, 40003 adres 2'dir. Bir kayma (off-by-one) en yaygın Modbus hatasıdır; geri kalanına güvenmeden önce bilinen bir değeri kontrol edin.

2. Register'ları okuyun ve çözümleyin

Bir Modbus register'ı 16 bittir. Float'lar ve 32 bit sayaçlar iki register kaplar ve üreticiler hangi word'ün önce geldiği konusunda anlaşamaz. struct ile çözümlemek word sırasını açık hale getirir:

import struct
from pymodbus.client import ModbusTcpClient

WORD_ORDER_BIG = True   # little-endian "word swap" cihazlar için False yapın

def decode(regs, dtype):
    if dtype == "uint16":
        return regs[0]
    if dtype == "int16":
        return struct.unpack(">h", struct.pack(">H", regs[0]))[0]
    words = regs if WORD_ORDER_BIG else list(reversed(regs))
    raw = struct.pack(">HH", *words)
    return struct.unpack(">f" if dtype == "float32" else ">I", raw)[0]

def read_tag(client, unit, func, addr, dtype):
    count = 2 if dtype in ("float32", "uint32") else 1
    if func == "coil":
        rr = client.read_coils(addr, count=1, device_id=unit)
        return None if rr.isError() else bool(rr.bits[0])
    reader = client.read_holding_registers if func == "holding" else client.read_input_registers
    rr = reader(addr, count=count, device_id=unit)
    return None if rr.isError() else decode(rr.registers, dtype)

Üretimde tag başına bir istek yerine ardışık blokları okuyun (örneğin 0–5 register'larını tek istekte): PLC tek büyük isteğe altı küçük istekten daha hızlı yanıt verir. Yukarıdaki tag başına sürüm okunması daha kolaydır ve birkaç düzine tag için yeterlidir.

3. Değişimde MQTT'ye publish edin

Şimdi poller'ı bir MQTT istemcisiyle birleştirin. Gateway her tag için bir topic publish eder, bir değeri yalnızca ölü bandın ötesinde değiştiğinde gönderir, yeni subscriber'lar güncel durumu hemen alsın diye değerleri retained olarak işaretler ve offline olduğunu duyurmak için Last Will kullanır.

import json, os, time
import paho.mqtt.client as mqtt
from pymodbus.client import ModbusTcpClient
from modbus_read import read_tag
from tags import TAGS

PLC_HOST, PLC_UNIT = os.getenv("PLC_HOST", "192.168.10.20"), 1
BASE = "testmqtt/acme/plant1/line2/plc1"
POLL_S, DEADBAND = 1.0, 0.05            # saniye, göreli değişim (%5)

mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="gw-line2-plc1")
mq.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
mq.tls_set()
mq.will_set(f"{BASE}/status", "offline", qos=1, retain=True)

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

mq.on_connect = on_connect
mq.connect("broker.connectmq.com", 8883, keepalive=30)
mq.loop_start()

plc = ModbusTcpClient(PLC_HOST, port=502, timeout=2)
last = {}

def changed(name, value):
    old = last.get(name)
    if old is None or isinstance(value, bool):
        return old != value
    return abs(value - old) > abs(old) * DEADBAND

while True:
    if not plc.connected and not plc.connect():
        mq.publish(f"{BASE}/status", "plc-unreachable", qos=1, retain=True)
        time.sleep(5)
        continue
    for name, func, addr, dtype, scale, unit in TAGS:
        value = read_tag(plc, PLC_UNIT, func, addr, dtype)
        if value is None:
            continue
        if not isinstance(value, bool):
            value = round(value * scale, 3)
        if changed(name, value):
            last[name] = value
            payload = {"v": value, "unit": unit, "ts": int(time.time() * 1000)}
            mq.publish(f"{BASE}/{name}", json.dumps(payload), qos=1, retain=True)
    time.sleep(POLL_S)

Açıklamaya değer birkaç karar:

• Tag başına bir topic (.../plc1/temperature) tüketicilerin tam olarak ihtiyaç duydukları şeye subscribe olmasını sağlar; .../plc1/+ gibi wildcard'lar yine her şeyi verir. Kalıpları topic matcher ile test edin.
• Retained + QoS 1, gece 3'te açılan bir dashboard'un son bilinen değeri anında göstermesi ve mesajların kısa ağ kesintilerinden sağ çıkması demektir. Retained mesajlar yazısına bakın.
• Last Will'li durum topic'i, "sıcaklık sabit" ile "gateway öldü" durumlarını ayırt eder. Gateway'in bağlantısı temiz bir disconnect olmadan koparsa broker sizin adınıza offline publish eder.
• Payload içindeki zaman damgası, değerin ne zaman ulaştığını değil ne zaman okunduğunu kaydeder; historian'lar için önemlidir.

Topic yapısı ISA-95 tarzı bir hiyerarşiyi (şirket / tesis / hat / cihaz) izler; bu da bir Unified Namespace için başlangıç noktasıdır.

4. Üretim PLC'si olmadan test edin

Asla çalışan bir hatta karşı geliştirme yapmayın. Bir Modbus simülatörüyle başlayın: pymodbus bir tane içerir; diagslave veya ModbusPal gibi araçlar da 502 portunda (root değilseniz 5020'de) yapılandırılabilir register'lar sunar. PLC_HOST'u ona yönlendirin ve gateway'i çalıştırın.

Ardından MQTT tarafını kontrol edin:

1. Online MQTT istemcisini açın, broker.connectmq.com'a bağlanın ve testmqtt/acme/plant1/# topic'ine subscribe olun. Retained status = online mesajını ve her tag için bir mesajı hemen görmelisiniz.
2. Simülatörde bir register'ı değiştirin. Yalnızca o tag yeni bir mesaj publish etmeli; değişmeyen tag'ler sessiz kalmalı.
3. Gateway'i kill -9 ile durdurun (temiz disconnect yok). Keep-alive süresinin yaklaşık bir buçuk katı sonra status offline olmalı.
4. Simülatörü durdurun. MQTT bağlı kalırken status plc-unreachable göstermeli.

Gateway topic'lerini tarayıcıda izleyin

Terminalde aynı kontrol tek bir komuttur:

mosquitto_sub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t 'testmqtt/acme/plant1/line2/#' -v

Üretim için sıkılaştırma

Modbus'ta hiçbir kimlik doğrulama yoktur; bu yüzden gateway çoğu zaman PLC ile geri kalan her şey arasındaki tek kontrollü sınırdır. MQTT yetkilerini dar, yazma yolunu ise gerçekten ihtiyacınız olmadıkça kapalı tutun.

Tesisleriniz ileride Sparkplug B'ye geçerse veya çok sayıda cihaz için standart durum yönetimi gerekirse, aynı okuma çekirdeği düz JSON topic'leri yerine bir Sparkplug B edge node'unu besleyebilir.

Modbus MQTT gateway küçük bir kod parçasıdır ama etkisi büyüktür: yalnızca PLC programcısının anladığı register'ları isimli, zaman damgalı ve subscribe olunabilir veriye dönüştürür. Net bir tag haritasıyla başlayın, retained değerler ve Last Will ile değişimde publish edin, bir simülatör ve online istemciyle test edin; hatta ancak ondan sonra bağlayın.

Sıkça sorulan sorular