Raspberry Pi IoT Gateway: BLE, Modbus ve Seri Verileri Toplayıp MQTT Üzerinden İletin

Bir Raspberry Pi IoT gateway, internete kendi başına ulaşamayan cihazlarla onların verilerine ihtiyaç duyan sistemlerin arasında durur. Bu rehberde Bluetooth Low Energy sensörlerinden, Modbus cihazlarından ve seri cihazlardan okumalar toplayan, bunları tek bir topic ve payload modelinde normalleştiren, ağ çöktüğünde diske arabelleğe alan ve bulut MQTT broker'a güvenli biçimde ileten bir gateway tasarlayacaksınız. Kod, genişletebileceğiniz çalışan bir iskelettir; bir oyuncak değil.

Bir IoT Gateway Aslında Ne Yapar

Saha cihazlarının çoğu, hiçbir zaman internet için tasarlanmamış protokoller konuşur: bir BLE termometrenin menzili birkaç metredir, bir Modbus RTU enerji sayacı RS-485 hattı üzerinden konuşur, bir laboratuvar terazisi de seri port üzerinden satırlar basar. Bir gateway onlar için dört iş yapar:

1. Protokol uyarlama. Her cihazı kendi yerel protokolünde okur.
2. Normalleştirme. Her şeyi birimler ve zaman damgalarıyla tutarlı topic'lere ve JSON payload'lara dönüştürür.
3. Güvenilirlik. Bağlantı koptuğunda verileri tutar ve daha sonra sırasıyla teslim eder.
4. Güvenli bağlantı. Cihazları açığa çıkarmak yerine tek bir kimliği doğrulanmış, şifreli dışa dönük bağlantı kurar.

Raspberry Pi bunun için iyi bir platformdur: tam Linux ve Python çalıştırır, RS-485 ve seri adaptörler için USB'si, dahili Bluetooth'u ve günlerce arabelleğe alınmış veri için yeterli depolaması vardır. Daha zorlu ortamlar için Revolution Pi gibi Pi tabanlı endüstriyel kontrolcüler, aynı yazılım yaklaşımını DIN ray muhafazasında sunar.

Gateway Mimarisi

 BLE sensors ----> [ble adapter]    \
 Modbus RTU/TCP -> [modbus adapter]  >--> normalise --> SQLite outbox --> MQTT uplink --> cloud broker
 Serial devices -> [serial adapter] /                                         ^
                                                                              | commands (optional)

Her adaptör bağımsız çalışır ve okumaları ortak bir giden kutusuna (outbox) teslim eder. Tek bir uplink görevi giden kutusunu broker'a boşaltır ve satırları yalnızca broker onayladıktan sonra siler. Bu ayrım, tüm gateway'deki en önemli tasarım kararıdır: yavaş bir Modbus cihazı veya kararsız bir BLE sensörü uplink'i asla engellemez; ölü bir internet bağlantısı da veri toplamayı asla engellemez.

Topic ve Payload Seçimi

Adaptörleri yazmadan önce veri modeline karar verin, çünkü onu sonradan değiştirmek her tüketiciyi değiştirmek demektir. Fiziksel sahayı yansıtan bir hiyerarşi iyi çalışır ve ISA-95 gibi endüstriyel adlandırmalara ölçeklenir:

<site>/<area>/<device>/telemetry   {"ts": 1760600000, "values": {"temp_c": 21.4}, "src": "ble"}
<site>/<area>/<device>/status      "online" / "offline" (retained)
<site>/gateway/<gateway-id>/status "online" / "offline" (retained, last will)

Zaman damgasını, okuma anında gateway tarafından belirlenecek şekilde payload'a koyun. Arabelleğe alınan veriler bir saat geç teslim edildiğinde bile tüketiciler her okumanın ne zaman alındığını bilir. Büyük sahalar için hiyerarşi rehberliği açısından Sparkplug B ve Unified Namespace kuralları iyi bir referanstır.

Raspberry Pi IoT Gateway İskeleti

Bağımlılıkları bir sanal ortama kurun:

sudo apt update && sudo apt install -y python3-venv bluetooth sqlite3
python3 -m venv ~/gw-env
~/gw-env/bin/pip install "paho-mqtt>=2.0" "pymodbus>=3.10" pyserial bleak
sudo usermod -aG dialout $USER   # serial port access; log out and back in

Çekirdeği ~/gateway.py olarak kaydedin. Giden kutusunu, uplink'i ve her protokol için bir örnek adaptörü içerir.

#!/usr/bin/env python3
"""Raspberry Pi IoT gateway: BLE + Modbus + serial -> SQLite outbox -> MQTT."""
import asyncio
import json
import os
import socket
import sqlite3
import ssl
import time

import paho.mqtt.client as mqtt

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
GW_ID = socket.gethostname()
CLIENT_ID = os.getenv("MQTT_CLIENT_ID", f"gw-{GW_ID}-91b6f0")  # YOUR_CLIENT_ID
SITE = os.getenv("SITE", "connectmq/demo/91b6f0/site1")
DB_PATH = os.getenv("OUTBOX", "/var/lib/gateway/outbox.db")

# ---------- outbox ----------
db = sqlite3.connect(DB_PATH, isolation_level=None, check_same_thread=False)
db.execute("PRAGMA journal_mode=WAL")
db.execute("CREATE TABLE IF NOT EXISTS outbox (id INTEGER PRIMARY KEY, topic TEXT, payload TEXT)")


def enqueue(area, device, values, src):
    payload = json.dumps({"ts": int(time.time()), "values": values, "src": src})
    db.execute("INSERT INTO outbox (topic, payload) VALUES (?, ?)",
               (f"{SITE}/{area}/{device}/telemetry", payload))


# ---------- uplink ----------
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)
status_topic = f"{SITE}/gateway/{GW_ID}/status"
client.will_set(status_topic, "offline", qos=1, retain=True)
client.on_connect = lambda c, u, f, rc, p: (not rc.is_failure) and c.publish(
    status_topic, "online", qos=1, retain=True)
client.reconnect_delay_set(1, 60)
client.connect_async(HOST, PORT, keepalive=60)
client.loop_start()


async def uplink(batch=50):
    while True:
        if not client.is_connected():
            await asyncio.sleep(2)
            continue
        rows = db.execute("SELECT id, topic, payload FROM outbox ORDER BY id LIMIT ?",
                          (batch,)).fetchall()
        if not rows:
            await asyncio.sleep(1)
            continue
        for row_id, topic, payload in rows:
            info = client.publish(topic, payload, qos=1)
            # Wait for PUBACK without blocking the event loop
            for _ in range(100):
                if info.is_published():
                    break
                await asyncio.sleep(0.05)
            if not info.is_published():
                break                      # keep the row, retry later
            db.execute("DELETE FROM outbox WHERE id = ?", (row_id,))
            await asyncio.sleep(0.11)      # stay under ~10 msg/s on the public broker


# ---------- adapters ----------
async def modbus_adapter():
    from pymodbus.client import AsyncModbusTcpClient
    mb = AsyncModbusTcpClient(os.getenv("MODBUS_HOST", "192.168.1.60"), port=502)
    while True:
        try:
            if not mb.connected:
                await mb.connect()
            rr = await mb.read_holding_registers(0, count=2, device_id=1)
            if not rr.isError():
                enqueue("utilities", "meter1",
                        {"voltage_v": rr.registers[0] / 10, "current_a": rr.registers[1] / 100},
                        "modbus")
        except Exception as exc:
            print("modbus:", exc)
        await asyncio.sleep(10)


async def serial_adapter():
    import serial
    port = os.getenv("SERIAL_PORT", "/dev/ttyUSB1")
    while True:
        try:
            with serial.Serial(port, 9600, timeout=1) as ser:
                while True:
                    line = await asyncio.to_thread(ser.readline)
                    text = line.decode(errors="ignore").strip()
                    if text:
                        enqueue("lab", "scale1", {"reading": text}, "serial")
        except Exception as exc:
            print("serial:", exc)
            await asyncio.sleep(5)


async def ble_adapter():
    from bleak import BleakScanner
    wanted = {m.strip().upper() for m in os.getenv("BLE_MACS", "").split(",") if m.strip()}

    def on_adv(device, adv):
        if device.address.upper() in wanted:
            enqueue("rooms", device.address.replace(":", "").lower(),
                    {"rssi": adv.rssi,
                     "mfr": {str(k): v.hex() for k, v in adv.manufacturer_data.items()}},
                    "ble")

    async with BleakScanner(on_adv):
        while True:
            await asyncio.sleep(3600)


async def main():
    tasks = [uplink()]
    if os.getenv("ENABLE_MODBUS") == "1":
        tasks.append(modbus_adapter())
    if os.getenv("ENABLE_SERIAL") == "1":
        tasks.append(serial_adapter())
    if os.getenv("ENABLE_BLE") == "1":
        tasks.append(ble_adapter())
    await asyncio.gather(*tasks)


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

Her adaptör bilerek minimal tutulmuştur. Gerçek cihazlar kod çözme gerektirir: BLE sensörleri sıcaklığı üreticiye özgü biçimlerde manufacturer veya service data içine paketler; Modbus register'ları ise cihaz kılavuzundaki ölçekleme ve kelime sırasına ihtiyaç duyar. Raspberry Pi BLE'den MQTT'ye eğitimi ve Raspberry Pi Modbus'tan MQTT'ye gateway yazısı her birini derinlemesine ele alır. pymodbus 3.10 ve sonrasının birim argümanını device_id olarak adlandırdığını, önceki 3.x sürümlerinin ise slave dediğini unutmayın.

Veri klasörünü oluşturun ve önce Modbus devre dışıyken yalnızca uplink'i, elle bir satır ekleyerek test edin:

sudo mkdir -p /var/lib/gateway && sudo chown $USER /var/lib/gateway
ENABLE_MODBUS=0 ~/gw-env/bin/python ~/gateway.py &
sqlite3 /var/lib/gateway/outbox.db \
  "INSERT INTO outbox(topic,payload) VALUES('connectmq/demo/91b6f0/site1/test/dev/telemetry','{\"ok\":true}')"

Ücretsiz genel MQTT broker üzerinde mesajın geldiğini mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-91b6f0 -t 'connectmq/demo/91b6f0/#' -v ile izleyin.

Depola-ve-İlet Neden Önemli

Giden kutusu deseni, verinin okunduğu anda diske yazılması ve yalnızca broker alındığını onayladıktan sonra silinmesi anlamına gelir. Bu size birkaç özelliği bedavaya verir:

• Yeniden başlatmalarda hiçbir şey kaybolmaz. Satırlar SQLite'ta kalır.
• Kesintiler adaptörler için görünmezdir. Kuyruğa eklemeye devam ederler; giden kutusu büyür ve sonra boşalır.
• Sıra korunur. Satırlar artan ID sırasıyla gönderilir.
• Birikmiş işleri tek bir SELECT COUNT(*) ile inceleyebilirsiniz.

İki uyarı. Birincisi, giden kutusunu sınırlayın: bir haftalık kesintinin diski doldurmaması için belirli bir sınırın üzerindeki en eski satırları düzenli olarak silen bir DELETE ekleyin. İkincisi, SD kart yıpranmasına dikkat edin. WAL modu yazmaları azaltır, ancak yoğun arabelleğe alma yapan bir gateway bir SSD veya endüstriyel sınıf bir kart kullanmalıdır. Daha fazla ayrıntı çevrimdışı arabelleğe alma rehberinde.

Alternatif: Yerel Mosquitto Artı Köprü

Bazı ekipler, adaptörlerin Pi üzerindeki yerel bir Mosquitto broker'a yayın yaptığı ve bir köprünün (bridge) buluta ilettiği yapıyı tercih eder. Bu, mevcut araçlar (Node-RED, Telegraf, üretici yazılımları) zaten MQTT konuşuyorsa ve yerel tüketicilerin internet kesikken bile verilere ihtiyacı varsa iyi çalışır. Bunun bedeli, arabelleğe almanın kendi giden kutunuz yerine Mosquitto'nun kuyruk sınırlarına bağlı hale gelmesidir. Mosquitto köprü yapılandırması eğitimi bu tasarımı baştan sona gösterir.

Raspberry Pi IoT Gateway Bağlantısının Güvenliği

Gateway, sahada internetle konuşan tek şeydir; bu yüzden onu bir güvenlik sınırı olarak ele alın:

1. Sertifika doğrulamalı olarak 8883 portunda TLS kullanın. Ana bilgisayar adı kontrollerini asla devre dışı bırakmayın.
2. Her gateway'e kendi kimlik bilgilerini ve client ID'sini verin; böylece ele geçirilen bir Pi tek başına iptal edilebilir.
3. Saha ağlarını (RS-485, BLE) ve uplink'i mantıksal olarak ayrı tutun; gateway aralarında trafik yönlendirmemelidir.
4. Servisi yetkisiz bir kullanıcı olarak çalıştırın ve Pi OS'u güncel tutun.
5. Parolalı SSH girişlerini devre dışı bırakıp anahtar kullanın ve kullanmadığınız servisleri kaldırın.

Bulut tarafında, ConnectMQ gibi bir bulut MQTT broker, bir broker'ı kendiniz çalıştırma ve sıkılaştırma ihtiyacını ortadan kaldırır. Portalda bir broker oluşturun, kullanıcı adını, parolayı ve client ID'yi kopyalayın ve servis ortamında MQTT_PORT=8883, MQTT_USERNAME=YOUR_USERNAME, MQTT_PASSWORD=YOUR_PASSWORD ve MQTT_CLIENT_ID=YOUR_CLIENT_ID değerlerini ayarlayın. ConnectMQ client ID'leri kimlik bilgilerine bağlar, plan bazlı kotalar ve flood koruması uygular, tekrarlanan başarısız girişlerden sonra IP'leri engeller. Topic'leriniz ISA-95 yapısını izliyorsa, portaldaki cihazlar görünümü onları otomatik olarak işletme, saha, alan ve hat bazında gruplar.

Birikmiş işleri boşaltırken plan sınırlarına dikkat edin. Ücretsiz Serverless plan QoS 1'e kadar ve 256 KB mesajlara izin verir; genel test broker'ı ise istemci başına saniyede yaklaşık 10 mesaja izin verir; uplink'in yayınlar arasında kısaca beklemesinin nedeni budur.

Servis Olarak Çalıştırın

sudo tee /etc/systemd/system/gateway.service >/dev/null <<'EOF'
[Unit]
Description=Raspberry Pi IoT gateway
After=network-online.target bluetooth.target
Wants=network-online.target

[Service]
User=pi
EnvironmentFile=/etc/gateway.env
ExecStart=/home/pi/gw-env/bin/python /home/pi/gateway.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF
sudo systemctl daemon-reload && sudo systemctl enable --now gateway

MQTT_*, SITE ve ENABLE_* değişkenlerini chmod 600 ile korunan /etc/gateway.env dosyasına koyun.

ConnectMQ ile Başlayın

Gateway'inize dakikalar içinde güvenli bir bağlantı kazandırın: app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz Serverless planda bir broker oluşturun ve oluşturulan kimlik bilgilerini /etc/gateway.env dosyasına yapıştırın. Temellere ihtiyacınız varsa Raspberry Pi MQTT eğitimi hepsini içerir.

Sık Sorulan Sorular

IoT gateway için en iyi Raspberry Pi modeli hangisi?

2 GB veya daha fazla belleğe sahip bir Pi 4 veya Pi 5, birkaç protokol adaptörü ve yerel bir arabellek için rahattır. Bir Pi Zero 2 W tek ve hafif bir adaptörü kaldırabilir ancak pek pay bırakmaz.

Gateway doğrudan buluta mı yoksa yerel Mosquitto üzerinden mi yayın yapmalı?

Gateway tek üretici olduğunda ve arabelleğe alma üzerinde tam kontrol istediğinizde doğrudan yayın yapın. Birden fazla yerel araç zaten MQTT konuşuyorsa veya kesintiler sırasında verilere ihtiyaç duyuyorsa köprülü yerel Mosquitto kullanın.

Buluttan saha cihazlarına nasıl komut gönderirim?

Gateway'i <site>/<area>/<device>/set gibi bir komut topic'ine abone edin ve mesajları Modbus yazmalarına veya seri komutlara çevirin. Her komutu uygulamadan önce doğrulayın.

Gateway verileri çevrimdışıyken ne kadar süre arabelleğe alabilir?

Diskinizin ve giden kutusu sınırınızın izin verdiği kadar. Sınırı mesaj hızınıza ve atlatmanız gereken en uzun kesintiye göre belirleyin.