Gateway IoT con Raspberry Pi: recopila datos BLE, Modbus y serie y reenvíalos por MQTT

Un gateway IoT con Raspberry Pi se sitúa entre los dispositivos que no pueden llegar a internet por sí solos y los sistemas que necesitan sus datos. En esta guía diseñarás un gateway que recopila lecturas de sensores Bluetooth Low Energy, dispositivos Modbus e instrumentos serie, las normaliza en un único modelo de topics y payloads, las almacena en disco cuando falla la red y las reenvía de forma segura a un broker MQTT en la nube. El código es un esqueleto funcional que puedes ampliar, no un juguete.

Qué hace realmente un gateway IoT

Muchos dispositivos de campo hablan protocolos que nunca se pensaron para internet: un termómetro BLE tiene un alcance de pocos metros, un medidor de energía Modbus RTU se comunica por un par RS-485 y una balanza de laboratorio imprime líneas por un puerto serie. Un gateway hace cuatro trabajos por ellos:

1. Adaptación de protocolos. Lee cada dispositivo en su protocolo nativo.
2. Normalización. Lo convierte todo en topics y payloads JSON coherentes, con unidades y marcas de tiempo.
3. Fiabilidad. Conserva los datos cuando el enlace de subida está caído y los entrega más tarde en orden.
4. Enlace seguro. Establece una única conexión de salida autenticada y cifrada en lugar de exponer los dispositivos.

Una Raspberry Pi es una buena plataforma para esto: ejecuta Linux completo y Python, tiene USB para adaptadores RS-485 y serie, Bluetooth integrado y almacenamiento suficiente para días de datos en búfer. Para entornos más exigentes, como plantas de minería o agroindustria, los controladores industriales basados en Pi, como el Revolution Pi, siguen el mismo enfoque de software en una carcasa para carril DIN.

Arquitectura del gateway

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

Cada adaptador funciona de forma independiente y entrega sus lecturas a una bandeja de salida (outbox) compartida. Una única tarea de subida vacía la outbox hacia el broker y solo borra las filas cuando el broker confirma su recepción. Esa separación es la decisión de diseño más importante de todo el gateway: un dispositivo Modbus lento o un sensor BLE inestable nunca bloquean la subida, y una conexión a internet caída nunca bloquea la recopilación de datos.

Elegir topics y payloads

Decide el modelo de datos antes de escribir los adaptadores, porque cambiarlo después implica cambiar todos los consumidores. Una jerarquía que refleje el sitio físico funciona bien y escala a nomenclaturas industriales como 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)

Pon la marca de tiempo en el payload, fijada por el gateway en el momento de la lectura. Cuando los datos en búfer se entregan con una hora de retraso, los consumidores siguen sabiendo cuándo se tomó cada lectura. Para orientarte sobre jerarquías en sitios más grandes, las convenciones de Sparkplug B y Unified Namespace son una buena referencia.

El esqueleto del gateway IoT con Raspberry Pi

Instala las dependencias en un entorno virtual:

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   # acceso al puerto serie; cierra sesión y vuelve a entrar

Guarda el núcleo como ~/gateway.py. Incluye la outbox, la subida y un adaptador de ejemplo por protocolo.

#!/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)
            # Esperar el PUBACK sin bloquear el bucle de eventos
            for _ in range(100):
                if info.is_published():
                    break
                await asyncio.sleep(0.05)
            if not info.is_published():
                break                      # conservar la fila y reintentar más tarde
            db.execute("DELETE FROM outbox WHERE id = ?", (row_id,))
            await asyncio.sleep(0.11)      # mantenerse por debajo de ~10 msg/s en el broker público


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

Cada adaptador es deliberadamente mínimo. Los dispositivos reales necesitan decodificación: los sensores BLE empaquetan la temperatura en manufacturer data o service data con formatos propios de cada fabricante, y los registros Modbus necesitan el escalado y el orden de palabras del manual del dispositivo. El tutorial BLE a MQTT con Raspberry Pi y el gateway Modbus a MQTT con Raspberry Pi profundizan en cada uno. Ten en cuenta que pymodbus 3.10 y posteriores llaman device_id al argumento de unidad; las versiones 3.x anteriores lo llaman slave.

Crea la carpeta de datos y prueba primero con Modbus desactivado, solo la subida, insertando una fila a mano:

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

Observa cómo llega al servidor MQTT público gratuito con mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-91b6f0 -t 'connectmq/demo/91b6f0/#' -v.

Por qué importa el store-and-forward

El patrón de outbox significa que los datos se escriben en disco en cuanto se leen y solo se borran cuando el broker confirma su recepción. Eso te da varias propiedades sin coste:

• Los reinicios no pierden nada. Las filas persisten en SQLite.
• Los cortes son invisibles para los adaptadores. Siguen encolando; la outbox crece y se vacía después.
• Se conserva el orden. Las filas se envían por ID ascendente.
• Puedes inspeccionar el atraso con un simple SELECT COUNT(*).

Dos advertencias. Primero, limita la outbox: añade un DELETE periódico de las filas más antiguas por encima de un límite para que un corte de una semana no llene el disco. Segundo, ten en cuenta el desgaste de la tarjeta SD. El modo WAL reduce las escrituras, pero un gateway que almacena mucho en búfer debería usar un SSD o una tarjeta de grado industrial. Encontrarás más detalles en la guía de almacenamiento sin conexión.

Alternativa: Mosquitto local más un bridge

Algunos equipos prefieren adaptadores que publican en un broker Mosquitto local en la Pi, con un bridge que reenvía a la nube. Funciona bien cuando las herramientas existentes (Node-RED, Telegraf, software del fabricante) ya hablan MQTT y cuando los consumidores locales necesitan los datos incluso sin internet. La contrapartida es que el búfer depende entonces de los límites de cola de Mosquitto y no de tu propia outbox. El tutorial de configuración de bridges en Mosquitto muestra ese diseño de principio a fin.

Proteger el enlace del gateway IoT con Raspberry Pi

El gateway es lo único del sitio que habla con internet, así que trátalo como una frontera de seguridad:

1. Usa TLS en el puerto 8883 con verificación de certificados. Nunca desactives la comprobación del hostname.
2. Da a cada gateway sus propias credenciales y su propio client ID, para poder revocar por separado una Pi comprometida.
3. Mantén las redes de campo (RS-485, BLE) y el enlace de subida separados lógicamente; el gateway no debe enrutar tráfico entre ellos.
4. Ejecuta el servicio con un usuario sin privilegios y mantén Pi OS actualizado.
5. Desactiva el inicio de sesión SSH con contraseña en favor de claves y elimina los servicios que no uses.

En el lado de la nube, un broker MQTT en la nube como ConnectMQ, que funciona como broker MQTT gestionado, elimina la necesidad de operar y reforzar un broker por tu cuenta. Crea un broker en el portal, copia el usuario, la contraseña y el client ID, y define MQTT_PORT=8883, MQTT_USERNAME=YOUR_USERNAME, MQTT_PASSWORD=YOUR_PASSWORD y MQTT_CLIENT_ID=YOUR_CLIENT_ID en el entorno del servicio. ConnectMQ vincula los client IDs a las credenciales, aplica cuotas por plan y protección contra inundaciones, y bloquea las IP tras repetidos inicios de sesión fallidos. Si tus topics siguen una estructura ISA-95, la vista de dispositivos del portal los agrupa automáticamente por empresa, sitio, área y línea.

Ten en cuenta los límites del plan al vaciar un atraso. El plan Serverless gratuito permite QoS hasta 1 y mensajes de 256 KB; el servidor público de pruebas permite unos 10 mensajes por segundo por cliente, por eso la subida hace una breve pausa entre publicaciones.

Ejecutarlo como servicio

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

Pon las variables MQTT_*, SITE y ENABLE_* en /etc/gateway.env con chmod 600.

Empieza con ConnectMQ

Dale a tu gateway un enlace seguro en minutos: crea un broker en app.connectmq.com en el plan Serverless gratuito, sin tarjeta de crédito, y pega las credenciales generadas en /etc/gateway.env. El tutorial de MQTT con Raspberry Pi recoge los fundamentos si los necesitas.

Preguntas frecuentes

¿Qué modelo de Raspberry Pi es mejor para un gateway IoT?

Una Pi 4 o Pi 5 con 2 GB o más va holgada para varios adaptadores de protocolo y un búfer local. Una Pi Zero 2 W puede con un único adaptador ligero, pero tiene poco margen.

¿El gateway debe publicar directamente en la nube o a través de un Mosquitto local?

Publica directamente cuando el gateway sea el único productor y quieras control total del búfer. Usa un Mosquitto local con bridge cuando varias herramientas locales ya hablen MQTT o necesiten los datos durante los cortes.

¿Cómo envío comandos desde la nube a los dispositivos de campo?

Suscribe el gateway a un topic de comandos como <site>/<area>/<device>/set y traduce los mensajes en escrituras Modbus o comandos serie. Valida cada comando antes de actuar.

¿Cuánto tiempo puede el gateway almacenar datos sin conexión?

Tanto como permitan tu disco y el límite de la outbox. Dimensiona el límite según tu tasa de mensajes y el corte más largo al que necesites sobrevivir.