Raspberry Pi IoT Gateway: Collect BLE, Modbus and Serial Data and Forward It over MQTT

A raspberry pi iot gateway sits between devices that cannot reach the internet on their own and the systems that need their data. In this guide you will design a gateway that collects readings from Bluetooth Low Energy sensors, Modbus devices and serial instruments, normalises them into one topic and payload model, buffers them on disk when the network fails and forwards them securely to a cloud MQTT broker. The code is a working skeleton you can extend, not a toy.

What an IoT Gateway Actually Does

Many field devices speak protocols that were never meant for the internet: a BLE thermometer has a range of a few metres, a Modbus RTU energy meter talks over an RS-485 pair, and a lab balance prints lines over a serial port. A gateway does four jobs for them:

1. Protocol adaptation. Read each device in its native protocol.
2. Normalisation. Turn everything into consistent topics and JSON payloads with units and timestamps.
3. Reliability. Keep data when the uplink is down and deliver it later in order.
4. Secure uplink. Make one authenticated, encrypted outbound connection instead of exposing devices.

A Raspberry Pi is a good platform for this: it runs full Linux and Python, has USB for RS-485 and serial adapters, built-in Bluetooth and enough storage for days of buffered data. For harsher environments, industrial Pi-based controllers such as the Revolution Pi follow the same software approach in a DIN rail housing.

Gateway Architecture

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

Each adapter runs independently and hands readings to a shared outbox. A single uplink task drains the outbox to the broker and deletes rows only after the broker acknowledges them. That separation is the most important design decision in the whole gateway: a slow Modbus device or a flaky BLE sensor never blocks the uplink, and a dead internet connection never blocks data collection.

Choosing Topics and Payloads

Decide the data model before writing adapters, because changing it later means changing every consumer. A hierarchy that mirrors the physical site works well and scales to industrial naming such as ISA-95:

<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)

Put the timestamp in the payload, set by the gateway at read time. When buffered data is delivered an hour late, consumers still know when each reading was taken. For guidance on hierarchies for larger sites, Sparkplug B and Unified Namespace conventions are a good reference.

The Raspberry Pi IoT Gateway Skeleton

Install the dependencies in a virtual environment:

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

Save the core as ~/gateway.py. It includes the outbox, the uplink and one example adapter per protocol.

#!/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())

Each adapter is deliberately minimal. Real devices need decoding: BLE sensors pack temperature into manufacturer or service data in vendor-specific formats, and Modbus registers need the scaling and word order from the device manual. The Raspberry Pi BLE to MQTT tutorial and the Raspberry Pi Modbus to MQTT gateway go deep on each. Note that pymodbus 3.10 and later name the unit argument device_id; earlier 3.x releases call it slave.

Create the data folder and test with Modbus disabled first, just the uplink, by inserting a row by hand:

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}')"

Watch it arrive on the free public MQTT broker with mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-91b6f0 -t 'connectmq/demo/91b6f0/#' -v.

Why Store-and-Forward Matters

The outbox pattern means data is written to disk the moment it is read and deleted only after the broker confirms receipt. That gives you several properties for free:

• Reboots lose nothing. Rows persist in SQLite.
• Outages are invisible to adapters. They keep enqueueing; the outbox grows and drains later.
• Order is preserved. Rows are sent by ascending ID.
• You can inspect the backlog with a single SELECT COUNT(*).

Two cautions. First, bound the outbox: add a periodic DELETE of the oldest rows above a limit so a week-long outage cannot fill the disk. Second, mind SD card wear. WAL mode reduces writes, but a gateway that buffers heavily should use an SSD or an industrial-grade card. More detail is in the offline buffering guide.

Alternative: Local Mosquitto Plus a Bridge

Some teams prefer adapters that publish to a local Mosquitto broker on the Pi, with a bridge forwarding to the cloud. That works well when existing tools (Node-RED, Telegraf, vendor software) already speak MQTT, and when local consumers need data even with the internet down. The trade-off is that buffering then depends on Mosquitto's queue limits rather than your own outbox. The Mosquitto bridge configuration tutorial shows that design end to end.

Securing the Raspberry Pi IoT Gateway Uplink

The gateway is the only thing on site that talks to the internet, so treat it as a security boundary:

1. Use TLS on port 8883 with certificate verification. Never disable hostname checks.
2. Give each gateway its own credentials and client ID, so one compromised Pi can be revoked alone.
3. Keep field networks (RS-485, BLE) and the uplink logically separate; the gateway should not route traffic between them.
4. Run the service as an unprivileged user and keep Pi OS updated.
5. Disable password SSH logins in favour of keys, and remove services you do not use.

For the cloud side, a cloud MQTT broker such as ConnectMQ removes the need to run and harden a broker yourself. Create a broker in the portal, copy the username, password and client ID, and set MQTT_PORT=8883, MQTT_USERNAME=YOUR_USERNAME, MQTT_PASSWORD=YOUR_PASSWORD and MQTT_CLIENT_ID=YOUR_CLIENT_ID in the service environment. ConnectMQ binds client IDs to credentials, applies per-plan quotas and flood protection, and blocks IPs after repeated failed logins. If your topics follow an ISA-95 structure, the portal's devices view groups them by enterprise, site, area and line automatically.

Mind the plan limits when draining a backlog. The free Serverless plan allows QoS up to 1 and 256 KB messages; the public test broker allows about 10 messages per second per client, which is why the uplink sleeps briefly between publishes.

Run It as a Service

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

Put the MQTT_*, SITE and ENABLE_* variables in /etc/gateway.env with chmod 600.

Get Started with ConnectMQ

Give your gateway a secure uplink in minutes: create a broker at app.connectmq.com on the free Serverless plan, no credit card needed, and paste the generated credentials into /etc/gateway.env. The Raspberry Pi MQTT tutorial has the fundamentals if you need them.

Frequently Asked Questions

Which Raspberry Pi model is best for an IoT gateway?

A Pi 4 or Pi 5 with 2 GB or more is comfortable for several protocol adapters and a local buffer. A Pi Zero 2 W can handle a single light adapter but has little headroom.

Should the gateway publish directly to the cloud or through local Mosquitto?

Publish directly when the gateway is the only producer and you want full control of buffering. Use local Mosquitto with a bridge when several local tools already speak MQTT or need data during outages.

How do I send commands from the cloud to field devices?

Subscribe the gateway to a command topic such as <site>/<area>/<device>/set and translate messages into Modbus writes or serial commands. Validate every command before acting on it.

How long can the gateway buffer data offline?

As long as your disk and outbox limit allow. Size the limit from your message rate and the longest outage you need to survive.