Modbus TCP'den MQTT'ye: Python ile Register Okuma ve Veri Yayımlama

Modbus TCP'den MQTT'ye köprü, endüstriyel IoT'deki en yaygın köprüdür; çünkü Modbus her yerdedir ve bu verinin panolara ve buluta ulaşma yolu MQTT'dir. Bu eğitimde holding ve input register'larını okuyan, onları doğru şekilde çözüp ölçekleyen ve yalnızca anlamlı değişiklikleri bir MQTT broker'a yayımlayan bir Python poller yazacaksınız. Sonuç, herhangi bir Linux makinesinde veya endüstriyel PC'de çalıştırabileceğiniz küçük, yapılandırılabilir bir gateway'dir.

Modbus TCP, MQTT'den Nasıl Ayrılır?

Modbus bir istek/yanıt protokolüdür. Bir client (tarihsel olarak "master") bir sunucudan (tarihsel olarak "slave") bir register bloğu ister ve onları geri alır. Hiçbir şey kendiliğinden gönderilmez: bir değer istiyorsanız onu sorgularsınız. Modbus'ta 16 bit register'lar ve tekil bitler dışında ad, birim veya veri tipi kavramı yoktur.

MQTT bunun tam tersidir: bir broker üzerinden, anlam taşıyan topic'lerle olay güdümlü publish/subscribe. Köprünün işi periyodik sorgulamaları anlamlı olaylara dönüştürmektir. Bu üç şey demektir: register haritasını bilmek, ham register'ları mühendislik değerlerine çözmek ve bir değerin ne zaman yayımlanmaya değer olduğuna karar vermek.

Modbus Veri Modeline Kısa Bir Bakış

İki ayrıntı neredeyse her ilk uygulamayı tökezletir:

• Adresleme ofsetleri. Dokümantasyon holding register'ları çoğu zaman 40001, 40002 vb. olarak listeler. Hat üzerinde 40001, 0 adresidir. Kütüphaneler sıfır tabanlı adresi kullanır.
• 32 bit değerler. Float'lar ve 32 bit tam sayılar iki register'a yayılır ve üreticiler word sıralamasında anlaşamaz. Bazıları önce yüksek word'ü gönderir (big-endian, çoğu zaman ABCD olarak yazılır), bazıları önce düşük word'ü (CDAB). Sıcaklığınız 1.2e-38 veya 3.5e+12 olarak okunuyorsa word'lerin yerini değiştirmeyi deneyin.

Ön Koşullar

• Python 3.10 veya daha yenisi.
• pymodbus 3.8 ve paho-mqtt 2.x. pymodbus sürümünü sabitleyin: anahtar kelime argümanları ara sürümler arasında değişti.
• Bir Modbus TCP cihazı veya aşağıdaki simülatör.

python -m venv .venv
source .venv/bin/activate
pip install "pymodbus>=3.8,<3.9" "paho-mqtt>=2.0"

1. Adım: Bir Modbus TCP Simülatörü (İsteğe Bağlı)

Donanım olmadan geliştirmek için, holding register 0 ve 1'de 32 bit float olarak bir sıcaklık, input register 0'da ölçeklenmiş tam sayı olarak bir basınç ve coil 0'da bir çalışma durumu sunan küçük bir sunucu çalıştırın.

# modbus_sim.py - değişen değerlerle 5020 portunda Modbus TCP sunucusu
import asyncio, random, struct
from pymodbus.datastore import (ModbusSequentialDataBlock, ModbusServerContext,
                                ModbusSlaveContext)
from pymodbus.server import StartAsyncTcpServer

store = ModbusSlaveContext(
    co=ModbusSequentialDataBlock(0, [0] * 10),
    di=ModbusSequentialDataBlock(0, [0] * 10),
    hr=ModbusSequentialDataBlock(0, [0] * 10),
    ir=ModbusSequentialDataBlock(0, [0] * 10),
)
context = ModbusServerContext(slaves=store, single=True)

async def update():
    while True:
        temp = 60 + random.uniform(-0.3, 0.3)
        hi, lo = struct.unpack(">HH", struct.pack(">f", temp))
        store.setValues(3, 0, [hi, lo])                         # holding register'lar
        store.setValues(4, 0, [int(random.uniform(4.9, 5.1) * 100)])  # input register, bar x100
        store.setValues(1, 0, [random.random() > 0.1])          # coil
        await asyncio.sleep(1)

async def main():
    asyncio.create_task(update())
    await StartAsyncTcpServer(context=context, address=("0.0.0.0", 5020))

asyncio.run(main())

5020 portu, 502 portu için gereken root yetkilerine ihtiyaç duymamanızı sağlar. Gerçek bir cihazda 502 portunu ve kılavuzundaki unit ID'yi kullanın.

2. Adım: Register Haritasını Tanımlayın

Bir tag eklemenin asla kod değişikliği gerektirmemesi için haritayı yapılandırmada tutun. Her tag'in bir tablosu, adresi, veri tipi, ölçeği, birimi, deadband'i ve topic yolu vardır.

{
  "modbus": { "host": "127.0.0.1", "port": 5020, "unit_id": 1, "poll_seconds": 1.0 },
  "topic_root": "connectmq/demo/b72f/plant1/utilities/boiler1",
  "heartbeat_seconds": 60,
  "tags": [
    { "name": "temperature", "table": "holding", "address": 0, "type": "float32", "word_order": "big",
      "scale": 1.0, "unit": "degC", "deadband": 0.5 },
    { "name": "pressure", "table": "input", "address": 0, "type": "uint16",
      "scale": 0.01, "unit": "bar", "deadband": 0.05 },
    { "name": "running", "table": "coil", "address": 0, "type": "bool", "unit": "" }
  ]
}

modbus_map.json olarak kaydedin ve b72f değerini kendi rastgele dizenizle değiştirin.

3. Adım: Modbus TCP'den MQTT'ye Köprü

Aşağıdaki poller report by exception uygular: analog bir değer deadband'inden fazla değiştiğinde, bir boolean ise değiştiğinde yayımlanır; her değer ayrıca bir heartbeat aralığında yeniden yayımlanır, böylece tüketiciler köprünün canlı olduğunu bilir. Bu model hakkında daha fazlasını report by exception yazımızda okuyun.

# modbus_mqtt_bridge.py - Modbus TCP'yi sorgula, değişiklikleri MQTT'ye yayımla
import json, logging, struct, time, uuid
import paho.mqtt.client as mqtt
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("bridge")
CFG = json.load(open("modbus_map.json"))

MQTT_HOST, MQTT_PORT = "broker.connectmq.com", 1883   # üretim: 8883 + TLS
MQTT_USER, MQTT_PASS = None, None                     # üretim: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"modbus-bridge-{uuid.uuid4().hex[:8]}"   # üretim: "YOUR_CLIENT_ID"
STATUS = f"{CFG['topic_root']}/bridge_status"


def mqtt_connect():
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if MQTT_USER:
        c.username_pw_set(MQTT_USER, MQTT_PASS)
        c.tls_set()
    c.will_set(STATUS, "offline", qos=1, retain=True)
    c.on_connect = lambda cl, ud, fl, rc, pr: cl.publish(STATUS, "online", qos=1, retain=True)
    c.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
    c.loop_start()
    return c


def decode(tag, regs):
    t = tag["type"]
    if t == "uint16":
        raw = regs[0]
    elif t == "int16":
        raw = struct.unpack(">h", struct.pack(">H", regs[0]))[0]
    elif t in ("float32", "int32", "uint32"):
        words = regs[:2] if tag.get("word_order", "big") == "big" else regs[1::-1]
        packed = struct.pack(">HH", *words)
        raw = struct.unpack({"float32": ">f", "int32": ">i", "uint32": ">I"}[t], packed)[0]
    else:
        raise ValueError(f"unsupported type {t}")
    return round(raw * tag.get("scale", 1.0), 4)


def read_tag(client, tag, unit):
    count = 2 if tag["type"] in ("float32", "int32", "uint32") else 1
    table = tag["table"]
    if table == "coil":
        rr = client.read_coils(tag["address"], count=1, slave=unit)
    elif table == "discrete":
        rr = client.read_discrete_inputs(tag["address"], count=1, slave=unit)
    elif table == "holding":
        rr = client.read_holding_registers(tag["address"], count=count, slave=unit)
    else:
        rr = client.read_input_registers(tag["address"], count=count, slave=unit)
    if rr.isError():
        raise ModbusException(f"{tag['name']}: {rr}")
    return bool(rr.bits[0]) if table in ("coil", "discrete") else decode(tag, rr.registers)


def should_publish(tag, value, last):
    if last is None:
        return True
    if isinstance(value, bool):
        if value != last["value"]:
            return True
    elif abs(value - last["value"]) >= tag.get("deadband", 0):
        return True
    return time.time() - last["sent"] >= CFG["heartbeat_seconds"]


def main():
    m = CFG["modbus"]
    mq = mqtt_connect()
    mb = ModbusTcpClient(m["host"], port=m["port"], timeout=2)
    last = {}
    while True:
        if not mb.connected and not mb.connect():
            log.warning("Modbus device unreachable, retrying")
            time.sleep(5)
            continue
        for tag in CFG["tags"]:
            topic = f"{CFG['topic_root']}/{tag['name']}"
            try:
                value = read_tag(mb, tag, m["unit_id"])
                quality = "good"
            except (ModbusException, ConnectionError, OSError) as exc:
                log.error("read failed: %s", exc)
                mb.close()
                value, quality = None, "bad"
            prev = last.get(tag["name"])
            if quality == "bad":
                if prev is None or prev.get("quality") != "bad":
                    mq.publish(topic, json.dumps({"value": None, "quality": "bad"}), qos=1, retain=True)
                    last[tag["name"]] = {"value": None, "quality": "bad", "sent": time.time()}
                continue
            if prev is None or prev.get("quality") == "bad" or should_publish(tag, value, prev):
                payload = {"value": value, "unit": tag.get("unit", ""), "quality": quality,
                           "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
                mq.publish(topic, json.dumps(payload), qos=1, retain=True)
                last[tag["name"]] = {"value": value, "quality": quality, "sent": time.time()}
                log.info("%s = %s", topic, value)
        time.sleep(m["poll_seconds"])


if __name__ == "__main__":
    main()

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

python modbus_sim.py            # terminal 1, gerçek cihazınız varsa atlayın
python modbus_mqtt_bridge.py    # terminal 2
mosquitto_sub -h broker.connectmq.com -p 1883 -i modbus-viewer-b72f \
  -t "connectmq/demo/b72f/#" -v   # terminal 3

Sıcaklığı yalnızca yarım derece kaydığında, basıncı yalnızca 0,05 bar değiştiğinde ve çalışma bayrağını yalnızca durumu değiştiğinde görmelisiniz. Her değer ayrıca dakikada bir heartbeat olarak yeniden görünür. Bu, saf bir "her sorgulamada yayımla" döngüsünün göndereceği mesajların küçük bir kısmıdır; her client'ın saniyede en fazla on mesaj yayımlayabildiği public MQTT test broker'da ve her ölçülü (metered) bağlantıda bu önemlidir.

Sorgulama Tasarımı İpuçları

1. Bitişik blokları okuyun. 0 ile 19 arasındaki register'lar için tek bir istek, tek register için yirmi istekten çok daha ucuzdur. Haritanız büyüdükçe tag'leri tabloya ve adres aralığına göre gruplayın ve bloktan çözün.
2. Cihaza saygı gösterin. Birçok küçük PLC ve sayaç saniyede yalnızca birkaç isteği kaldırabilir. Bir saniyelik sorgulamayla başlayın ve ölçün.
3. Doğru unit ID'yi kullanın. Bir gateway arkasındaki Modbus TCP cihazları seri cihazları adreslemek için unit ID kullanır; bağımsız cihazlar çoğu zaman onu yok sayar ya da 1 veya 255 bekler.
4. Hızlı ve yavaş tag'leri ayırın. Alarmları ve durumları hızlı, enerji sayaçlarını yavaş sorgulayın.
5. Kaliteyi açıkça yayımlayın. Başarısız bir okuma, son değeri sessizce korumak yerine quality: bad yayımlamalıdır.
6. Poller'dan asla yanlışlıkla yazmayın. Yazma desteğini köprünün dışında tutun veya ayrı, kimlik doğrulamalı bir komut topic'inin arkasına koyun.

Modbus Verisi İçin Topic ve Payload Tasarımı

Ham Modbus size anlam değil adres verir; bu yüzden köprünün bağlam eklediği yer topic ağacıdır. Topic'leri register numarasına göre değil, değerin ne olduğuna ve nerede bulunduğuna göre adlandırın: plant1/utilities/boiler1/temperature herkes için faydalıdır, device3/hr/0 ise yalnızca register haritasını yazan kişi için. Saha, alan, hat ve ekipmandan oluşan ISA-95 tarzı bir hiyerarşi yüzlerce Modbus cihazını gezilebilir tutar ve tüketicilerin bütün bir alana tek bir wildcard ile abone olmasını sağlar.

Tüketicilerin tek tek seçtiği proses verisi için topic başına bir değer tutun ve her payload'a birim, kalite ve zaman damgası ekleyin; böylece bir mesaj aylar sonra bir log'da göründüğünde de kendini açıklar. Cihazda her zaman birlikte kullanılan çok sayıda register varsa, örneğin bir sayacın tüm okumaları gibi, döngü başına cihaz başına tek bir JSON nesnesi makul bir alternatiftir ve mesaj sayısını düşürür. Hangisini seçerseniz seçin, sahadaki tüm köprülerde tutarlı şekilde uygulayın.

Üretime Geçiş

Gerçek tesis verisi için özel bir broker'a geçin: MQTT_PORT değerini 8883 yapın ve yönetilen MQTT broker'ınızdan aldığınız kimlik bilgilerini doldurun. ConnectMQ ile her broker, oluşturulmuş bir kullanıcı adı, şifre ve client ID ile gelir; client ID bu kimlik bilgilerine bağlıdır, bu yüzden rastgele bir ID yerine onu kullanın. Köprüyü Restart=always ile bir systemd servisi olarak çalıştırın ve sahanın internet bağlantısı güvenilir değilse store and forward tamponlamayı düşünün.

Cihazlarınız Ethernet yerine RS-485 üzerinden Modbus RTU konuşuyorsa aynı köprü bir seri client ile çalışır; Modbus RTU'dan MQTT'ye eğitimimiz kablolamayı ve seri ayarları ele alıyor. Gateway'lerin, broker'ların ve tüketicilerin genel mimarisi için endüstriyel IoT MQTT rehberine bakın.

Sorun Giderme

ConnectMQ ile Başlayın

Modbus verinizi sunucu çalıştırmadan tesis ağının dışına taşıyın. app.connectmq.com adresinde kaydolun, ücretsiz Serverless planı seçin (kredi kartı gerekmez), bir broker oluşturun ve oluşturulan kimlik bilgilerini köprüye yapıştırın. Register değerleriniz ConnectMQ canlı topic explorer'ında hemen görünür; eşik uyarı kuralları da bir sıcaklık veya basınç aralığının dışına çıktığında sizi uyarabilir.

Sık Sorulan Sorular

Modbus TCP register'larını ne sıklıkla sorgulamalıyım?

Proses değerleri için bir saniyeyle, sayaçlar için daha yavaş başlayın; ardından cihazın kaldırabildiğine ve tüketicilerin ihtiyacına göre ayarlayın. Her sorgulamada değil, değişiklik olduğunda yayımlayın.

Modbus float değerlerim neden yanlış görünüyor?

Çoğu zaman iki 16 bit word, kodunuzun varsaydığının tersi sıradadır. Word sıralamasını değiştirin ve register'ın aslında float değil ölçeklenmiş bir tam sayı olup olmadığını kontrol edin.

Tek bir köprü birden fazla Modbus TCP cihazını sorgulayabilir mi?

Evet. Cihaz başına bir client veya farklı unit ID'lerle gateway başına bir client çalıştırın ve her cihaza topic ağacında kendi dalını verin.

Köprü QoS 0 mı yoksa QoS 1 mi kullanmalı?

Proses değerleri için retained message'larla QoS 1 iyi bir varsayılandır; iletimi garanti eder ve yeni abonelerin güncel durumu hemen görmesini sağlar.