pymodbus ile Raspberry Pi Modbus MQTT Gateway (RTU ve TCP)

Bir Raspberry Pi Modbus MQTT gateway, yalnızca Modbus konuşan enerji sayaçlarından, invertörlerden, VFD'lerden ve PLC'lerden okumalar alıp bunları MQTT üzerinden temiz JSON olarak yayınlamanızı sağlar. Bu eğitimde Modbus RTU için bir USB RS-485 adaptörü bağlayacak, Ethernet üzerinden Modbus TCP cihazlarına bağlanacak, her cihazın register'larını küçük bir yapılandırmada tanımlayacak, ölçekli tam sayıları ve 32 bit float değerleri doğru çözecek ve sonuçları bir broker'a yayınlayacaksınız. Son betik zaman aşımlarını ve yeniden bağlanmaları yönetir; böylece gözetimsiz çalışabilir.

İki Dakikada Modbus

Modbus bir istek-yanıt protokolüdür: Pi (istemci, tarihsel olarak master) bir cihazdan (sunucu, tarihsel olarak slave) bir register bloğu ister ve cihaz yanıt verir. Topic, abonelik veya itme (push) kavramı yoktur; her şey yoklamadır (polling). MQTT sistemlerini beslemek için bir gateway'e ihtiyaç duyulmasının nedeni tam olarak budur.

Veriler dört tabloda bulunur: coil'ler ve discrete input'lar (bitler) ile input ve holding register'lar (16 bitlik kelimeler). Çoğu sayaç ölçümleri holding veya input register olarak sunar. Enerji toplamları veya kayan noktalı voltajlar gibi 16 bite sığmayan değerler ardışık iki register'a yayılır ve kelime sırasını cihaz kılavuzu söyler. Bu sırayı yanlış almak, bir gateway'in saçma sayılar yayınlamasının en yaygın nedenidir.

Her varyantın protokol arka planı için Modbus RTU'dan MQTT'ye ve Modbus TCP'den MQTT'ye rehberlerine bakın.

Modbus RTU İçin Donanım ve Kablolama

1. Pi'ye bir USB RS-485 adaptörü takın. /dev/ttyUSB0 olarak görünür. Yeniden başlatmalarda ve çıkarıp takmalarda değişmeyen kalıcı bir ad için ls /dev/serial/by-id/ kullanın.
2. A'yı A'ya (bazen D+ olarak etiketlenir), B'yi B'ye (D-) bağlayın. Hiç yanıt göremiyorsanız ilk deneyeceğiniz şey A ile B'nin yerini değiştirmektir, çünkü üreticiler etiketlemede anlaşamaz.
3. Adaptör ve cihaz sağlıyorsa sinyal toprağını bağlayın. Ortak referans olmayan uzun hatlar aralıklı hatalara neden olur.
4. Uzun hatlarda hattın iki fiziksel ucunu 120 ohm dirençlerle sonlandırın. Birçok cihaz ve adaptörde bir sonlandırma anahtarı bulunur.
5. Cihazın baud hızını, paritesini, stop bitlerini ve unit ID'sini ekranından veya kılavuzundan not edin. Aynı hat üzerindeki tüm cihazlar aynı seri ayarları paylaşmalıdır.

Modbus TCP için yalnızca cihazın IP adresine, portuna (genellikle 502) ve unit ID'sine ihtiyacınız vardır. Bir Modbus TCP-RTU gateway'inin arkasında unit ID, karşı taraftaki RTU cihazını seçer.

pymodbus ve paho-mqtt Kurulumu

sudo apt update && sudo apt install -y python3-venv
python3 -m venv ~/modbus-env
~/modbus-env/bin/pip install "pymodbus>=3.10" pyserial "paho-mqtt>=2.0"
sudo usermod -aG dialout $USER     # serial access, then log out and in

pymodbus 3.10, birim argümanının adını slave'den device_id'ye değiştirdi. Aşağıdaki kod device_id kullanır; daha eski bir 3.x sürümünde kaldıysanız onu tekrar slave olarak değiştirin.

Daha büyük bir şey yazmadan önce bir Python kabuğundan hızlı bir sağlama yapın:

from pymodbus.client import ModbusSerialClient

c = ModbusSerialClient(port="/dev/ttyUSB0", baudrate=9600, parity="N",
                       stopbits=1, bytesize=8, timeout=1)
print(c.connect())
rr = c.read_holding_registers(0, count=2, device_id=1)
print(rr if rr.isError() else rr.registers)
c.close()

Bu register değerlerini yazdırıyorsa kablolama ve ayarlar doğrudur. Zaman aşımı yazdırıyorsa sırasıyla A/B'yi, baud hızını, pariteyi ve unit ID'yi kontrol edin.

Cihazları Bir Register Haritasında Tanımlayın

Adresleri Python'a sabit kodlamak hızla karmaşıklaşır. Bunun yerine her cihazın register'larını bir JSON dosyasına koyun. ~/devices.json olarak kaydedin:

{
  "devices": [
    {
      "name": "meter1",
      "area": "utilities",
      "transport": "rtu",
      "port": "/dev/ttyUSB0",
      "baudrate": 9600,
      "parity": "N",
      "unit": 1,
      "interval": 10,
      "registers": [
        {"field": "voltage_v",  "table": "holding", "address": 0, "type": "uint16",  "scale": 0.1},
        {"field": "current_a",  "table": "holding", "address": 1, "type": "uint16",  "scale": 0.01},
        {"field": "power_w",    "table": "holding", "address": 2, "type": "int32",   "word_order": "big"},
        {"field": "energy_kwh", "table": "input",   "address": 10, "type": "float32", "word_order": "little"}
      ]
    },
    {
      "name": "vfd1",
      "area": "line1",
      "transport": "tcp",
      "host": "192.168.1.70",
      "port": 502,
      "unit": 1,
      "interval": 5,
      "registers": [
        {"field": "speed_rpm",  "table": "holding", "address": 100, "type": "uint16", "scale": 1},
        {"field": "fault_code", "table": "holding", "address": 105, "type": "uint16", "scale": 1}
      ]
    }
  ]
}

Yukarıdaki adresler örnektir. Kendi cihazınızın kılavuzundaki register tablosunu kullanın ve adresleri 1'den veya 40001 ofsetiyle başlatan kılavuzlara dikkat edin: pymodbus sıfır tabanlı protokol adresini bekler; yani register 40001, adres 0'dır.

Raspberry Pi Modbus MQTT Gateway Betiği

~/modbus2mqtt.py olarak kaydedin:

#!/usr/bin/env python3
"""Poll Modbus RTU/TCP devices and publish readings over MQTT."""
import json
import os
import ssl
import struct
import time

import paho.mqtt.client as mqtt
from pymodbus.client import ModbusSerialClient, ModbusTcpClient
from pymodbus.exceptions import ModbusException

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
USERNAME = os.getenv("MQTT_USERNAME")                  # YOUR_USERNAME
PASSWORD = os.getenv("MQTT_PASSWORD")                  # YOUR_PASSWORD
CLIENT_ID = os.getenv("MQTT_CLIENT_ID", "modbus-gw-2f84ad")  # YOUR_CLIENT_ID
PREFIX = os.getenv("MQTT_PREFIX", "connectmq/demo/2f84ad/site1")
CONFIG = os.getenv("DEVICES", os.path.expanduser("~/devices.json"))

SIZES = {"uint16": 1, "int16": 1, "uint32": 2, "int32": 2, "float32": 2}


def decode(regs, rtype, word_order="big"):
    if SIZES[rtype] == 2 and word_order == "little":
        regs = [regs[1], regs[0]]
    raw = b"".join(struct.pack(">H", r) for r in regs)
    fmt = {"uint16": ">H", "int16": ">h", "uint32": ">I",
           "int32": ">i", "float32": ">f"}[rtype]
    return struct.unpack(fmt, raw)[0]


def make_client(dev):
    if dev["transport"] == "rtu":
        return ModbusSerialClient(port=dev["port"], baudrate=dev.get("baudrate", 9600),
                                  parity=dev.get("parity", "N"), stopbits=dev.get("stopbits", 1),
                                  bytesize=8, timeout=1, retries=1)
    return ModbusTcpClient(dev["host"], port=dev.get("port", 502), timeout=2, retries=1)


def read_device(mb, dev):
    values = {}
    for reg in dev["registers"]:
        count = SIZES[reg["type"]]
        fn = mb.read_holding_registers if reg["table"] == "holding" else mb.read_input_registers
        rr = fn(reg["address"], count=count, device_id=dev["unit"])
        if rr.isError():
            raise ModbusException(f"{dev['name']} {reg['field']}: {rr}")
        v = decode(rr.registers, reg["type"], reg.get("word_order", "big"))
        v = v * reg.get("scale", 1)
        values[reg["field"]] = round(v, 4) if isinstance(v, float) else v
    return values


# ---------- MQTT ----------
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                     protocol=mqtt.MQTTv5)
if USERNAME:
    client.username_pw_set(USERNAME, PASSWORD)
if PORT == 8883:
    client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
gw_status = f"{PREFIX}/gateway/modbus/status"
client.will_set(gw_status, "offline", qos=1, retain=True)


def on_connect(c, userdata, flags, reason_code, properties):
    if not reason_code.is_failure:
        c.publish(gw_status, "online", qos=1, retain=True)


client.on_connect = on_connect
client.reconnect_delay_set(1, 60)
client.connect_async(HOST, PORT, keepalive=60)
client.loop_start()

# ---------- polling loop ----------
with open(CONFIG) as f:
    devices = json.load(f)["devices"]

clients = {}
next_poll = {d["name"]: 0.0 for d in devices}
last_ok = {}

while True:
    now = time.monotonic()
    for dev in devices:
        name = dev["name"]
        if now < next_poll[name]:
            continue
        next_poll[name] = now + dev.get("interval", 10)
        base = f"{PREFIX}/{dev['area']}/{name}"
        # RTU devices on the same port share one client
        key = dev.get("port") if dev["transport"] == "rtu" else f"{dev['host']}:{dev.get('port', 502)}"
        mb = clients.get(key) or clients.setdefault(key, make_client(dev))
        try:
            if not mb.connected:
                mb.connect()
            values = read_device(mb, dev)
            payload = {"ts": int(time.time()), "values": values}
            client.publish(f"{base}/telemetry", json.dumps(payload), qos=1)
            if last_ok.get(name) is not True:
                client.publish(f"{base}/status", "online", qos=1, retain=True)
            last_ok[name] = True
        except (ModbusException, OSError) as exc:
            print(f"{name}: {exc}")
            if last_ok.get(name) is not False:
                client.publish(f"{base}/status", "offline", qos=1, retain=True)
            last_ok[name] = False
            mb.close()
    time.sleep(0.2)

Anlaşılmaya değer tasarım notları:

• Her fiziksel bağlantı için bir istemci. Aynı RS-485 portundaki RTU cihazları tek bir seri istemciyi paylaşır, çünkü aynı portu açan iki süreç veya nesne birbirinin çerçevelerini bozar.
• Cihaz başına durum topic'leri. Her cihaz, yalnızca durum değiştiğinde yayınlanan retained bir online veya offline alır. Paneller, gateway'in kendi durumundan ayrı olarak hangi sayacın yanıt vermeyi bıraktığını hemen görür.
• Ham register'lar tek bir yerde çözülür. decode fonksiyonu 16 ve 32 bitlik tipleri ve kelime sırasını struct ile ele alır; böylece pymodbus sürümleri arasında değişmiş yardımcı API'lere bağımlı olmaz.
• Cihaz başına yoklama aralıkları. Her 5 saniyede bir VFD hızı ve her 60 saniyede bir enerji toplamı mantıklı bir karışımdır; tüketicilerinizin ihtiyacından daha hızlı yoklama yapmayın.

Raspberry Pi Modbus MQTT Çıktısını Test Edin

Ön planda çalıştırın:

~/modbus-env/bin/python ~/modbus2mqtt.py

Başka bir terminalden ücretsiz genel MQTT broker üzerindeki çıktıyı izleyin:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-2f84ad \
  -t 'connectmq/demo/2f84ad/#' -v

Beklenen çıktı:

connectmq/demo/2f84ad/site1/gateway/modbus/status online
connectmq/demo/2f84ad/site1/utilities/meter1/status online
connectmq/demo/2f84ad/site1/utilities/meter1/telemetry {"ts": 1760601234, "values": {"voltage_v": 231.4, "current_a": 1.27, "power_w": 287, "energy_kwh": 1532.6875}}

En az bir değeri cihazın kendi ekranıyla karşılaştırın. Bir float değer aşırı derecede yanlışsa (çok büyük üsler veya çok küçük kesirler), word_order değerini tersine çevirin. Ölçekli bir değer on kat sapıyorsa scale değerini düzeltin.

Henüz Modbus donanımınız yok mu? pymodbus bir simülatör içerir ve Modbus TCP'den MQTT'ye rehberi, test için bir dizüstü bilgisayarda nasıl çalıştırılacağını gösterir.

Üretim: Özel Broker, TLS ve Servis

Sayaç ve makine verileri operasyonel bilgidir; bu yüzden üretimdeki gateway'ler şifrelemeli özel bir broker'a yayın yapmalıdır. Tamamen yönetilen bir MQTT hizmeti olan ConnectMQ ile portalda bir broker oluşturur ve bir kullanıcı adı, parola ve client ID alırsınız. Bunları /etc/modbus2mqtt.env dosyasına koyun:

MQTT_HOST=broker.connectmq.com
MQTT_PORT=8883
MQTT_USERNAME=YOUR_USERNAME
MQTT_PASSWORD=YOUR_PASSWORD
MQTT_CLIENT_ID=YOUR_CLIENT_ID
MQTT_PREFIX=acme/plant1
DEVICES=/home/pi/devices.json

Ardından bu serideki diğer Pi eğitimlerinde olduğu gibi EnvironmentFile=/etc/modbus2mqtt.env, Restart=always ve After=network-online.target içeren bir systemd birimi oluşturun. Öneki işletme, saha, alan ve hat şeklinde yapılandırırsanız ConnectMQ portalı ISA-95 hiyerarşisini tanır ve sayaçları cihazlar görünümünde gruplar; eşik tabanlı uyarı kuralları da sıfır olmayan bir VFD fault_code gibi değerleri işaretleyebilir.

İki üretim konusu daha var. Birincisi, uplink uzun süre kesik kalabilecekse yayın çağrısının önüne bir disk arabelleği ekleyin; Raspberry Pi IoT gateway rehberi yeniden kullanabileceğiniz bir SQLite giden kutusu içerir. İkincisi, buluttan cihazlara yazmayı planlıyorsanız katı doğrulama ve yazılabilir register'lar için bir izin listesi ekleyin. Yanlış bir Modbus yazması bir makineyi durdurabilir.

Sorun Giderme

ConnectMQ ile Başlayın

Modbus verilerinizi işe yarar bir yere gönderin: app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz Serverless planda bir broker oluşturun ve sayaç okumalarınızın portalın canlı topic gezginine gelişini izleyin. Pi serisinin geri kalanı için Raspberry Pi MQTT eğitimi ile başlayın.

Sık Sorulan Sorular

Tek bir Raspberry Pi birden fazla Modbus RTU cihazını okuyabilir mi?

Evet. Onları benzersiz unit ID'ler ve aynı seri ayarlarla aynı RS-485 hattına koyun; gateway tek bir seri istemci üzerinden onları sırayla yoklar.

Holding ve input register'lar arasındaki fark nedir?

Input register'lar salt okunur ölçümlerdir; holding register'lar ise okunabilir ve çoğu zaman ayarlar ve set değerleri için yazılabilir. Her değerin hangi tabloda bulunduğunu cihaz kılavuzunuz söyler.

Gateway Modbus RTU'yu ne kadar hızlı yoklayabilir?

Bu; baud hızına, register sayısına ve cihazın yanıt süresine bağlıdır. Mümkün olduğunda ardışık register'ları tek bir okumada gruplayın ve yalnızca tüketicilerinizin ihtiyaç duyduğu sıklıkta yoklayın.

Gateway register başına mı yoksa cihaz başına mı bir mesaj yayınlamalı?

Genellikle en iyisi her cihaz ve yoklama için tek bir JSON mesajıdır. İlgili değerleri tek bir zaman damgasıyla bir arada tutar ve çok daha az mesaj kullanır.