Crea un simulador de dispositivos MQTT en Python para cientos de sensores virtuales

Un simulador de dispositivos MQTT te permite construir dashboards, pipelines de almacenamiento y reglas de alerta antes de instalar un solo sensor real. En este tutorial escribirás un simulador en Python que modela cientos de sensores virtuales con ruido, deriva, fallos y cortes creíbles, y los publica en el servidor MQTT público gratuito broker.connectmq.com sin superar sus límites de tasa.

Por qué simular dispositivos antes de tener el hardware

El hardware suele ser la parte más lenta de un proyecto IoT. Las placas están pedidas, el firmware está a medio escribir y el instalador aún no ha visitado el emplazamiento, mientras que quienes construyen el dashboard y la base de datos ya necesitan datos. Un simulador elimina esa dependencia. Proporciona a cada componente posterior un flujo realista con el que trabajar y te permite ensayar situaciones difíciles de provocar con equipos reales: un sensor que se dispara a un valor imposible, un dispositivo que se queda en silencio diez minutos o una línea entera que entra en servicio de golpe.

Hay una segunda ventaja que los equipos descubren más tarde. Cuando la flota real ya existe, el mismo simulador se convierte en tu herramienta de regresión. Puedes ejecutarlo contra un broker de staging después de cada cambio en tu código de ingesta y ver al instante si se ha roto un parser o un dashboard.

Lo que separa un simulador útil de un juguete es el realismo. Un bucle que publica random.uniform(20, 25) cada segundo genera datos que no se parecen en nada a los de una sonda de temperatura real. Las señales reales cambian despacio, siguen ciclos diarios, tienen ruido de medición, derivan con las semanas y a veces fallan. El simulador de abajo modela cada uno de esos comportamientos con unas pocas líneas de código.

Diseño del simulador de dispositivos MQTT

Antes de escribir código, decide la estructura de topics y el formato del payload, porque todo lo que construyas aguas abajo dependerá de ellos. Este tutorial usa un topic por dispositivo y un payload JSON compacto:

connectmq/demo/<run-id>/<line>/<device>/telemetry

{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}

El <run-id> es una cadena aleatoria generada al arrancar. El broker público puede leerlo cualquiera, así que un prefijo único mantiene tu flujo separado de los experimentos de otras personas y facilita suscribirte exactamente a tus datos. La estructura no pasa de seis niveles de topic, holgadamente por debajo del límite de ocho del broker público.

El simulador tiene tres partes:

1. Un modelo de sensor que guarda el estado de cada dispositivo (valor base, deriva, fase del ciclo diario, indicador de fallo) y devuelve una nueva lectura cuando se le pide.
2. Un planificador que decide qué dispositivo publica a continuación. Los dispositivos se reparten de forma uniforme a lo largo del intervalo en lugar de dispararse todos a la vez.
3. Un limitador de tasa que mantiene al cliente publicador por debajo del límite de mensajes por cliente del broker.

Respeta los límites de tasa del broker

El servidor público permite 10 mensajes por segundo por cliente, 10 suscripciones por cliente y 64 KB por mensaje. Cientos de dispositivos publicando cada segundo superarían eso en un instante, así que el simulador tiene que hacer antes algunas cuentas. La tasa agregada es simplemente el número de dispositivos dividido entre el intervalo de publicación.

El código se limita a 8 mensajes por segundo para dejar margen a los reintentos y a las herramientas de suscripción que ejecutes en paralelo. Si necesitas una tasa mayor, tienes dos opciones honestas: repartir los dispositivos entre varias conexiones de cliente, cada una con su propio ID y su propio presupuesto, o pasar a un broker privado donde los límites los fija tu plan y no el servicio público compartido.

El simulador completo en Python

Instala la única dependencia:

pip install "paho-mqtt>=2.0"

Guarda lo siguiente como simulator.py. Usa la API de callbacks de paho-mqtt 2.x y, por lo demás, solo la biblioteca estándar.

import argparse
import heapq
import json
import math
import random
import time
import uuid

import paho.mqtt.client as mqtt


class VirtualSensor:
    """Un sensor simulado de temperatura y humedad con estado persistente."""

    def __init__(self, index: int, fault_rate: float):
        self.device_id = f"sensor-{index:04d}"
        self.line = f"line{index % 4 + 1}"
        self.base_temp = random.uniform(19.0, 24.0)
        self.humidity = random.uniform(35.0, 55.0)
        self.phase = random.uniform(0, 2 * math.pi)
        self.drift = 0.0
        self.faulty = random.random() < fault_rate

    def read(self, now: float) -> dict:
        # Ciclo diario lento de +/- 2,5 grados, desfasado por dispositivo
        daily = 2.5 * math.sin(2 * math.pi * (now % 86400) / 86400 + self.phase)
        # Paseo aleatorio acotado para imitar la deriva de calibración
        self.drift = max(-3.0, min(3.0, self.drift + random.gauss(0, 0.05)))
        # Ruido de medición gaussiano
        temp = self.base_temp + daily + self.drift + random.gauss(0, 0.15)
        self.humidity = max(10.0, min(95.0, self.humidity + random.gauss(0, 0.3)))
        # Los dispositivos defectuosos emiten de vez en cuando picos físicamente imposibles
        if self.faulty and random.random() < 0.05:
            temp += random.choice([-40.0, 60.0])
        return {
            "device": self.device_id,
            "ts": int(now * 1000),
            "temperature": round(temp, 2),
            "humidity": round(self.humidity, 1),
        }


def build_client(args, run_id: str) -> mqtt.Client:
    client_id = args.client_id or f"sim-{run_id}"
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    if args.username:
        client.username_pw_set(args.username, args.password)
    if args.tls:
        client.tls_set()  # almacén de CA del sistema; el broker usa un certificado de Let's Encrypt

    def on_connect(client, userdata, flags, reason_code, properties):
        print(f"connected: {reason_code}")

    def on_disconnect(client, userdata, flags, reason_code, properties):
        print(f"disconnected: {reason_code}")

    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    return client


def main():
    parser = argparse.ArgumentParser(description="MQTT device simulator")
    parser.add_argument("--host", default="broker.connectmq.com")
    parser.add_argument("--port", type=int, default=1883)
    parser.add_argument("--tls", action="store_true")
    parser.add_argument("--username")
    parser.add_argument("--password")
    parser.add_argument("--client-id")
    parser.add_argument("--prefix", help="topic prefix, default connectmq/demo/<run-id>")
    parser.add_argument("--devices", type=int, default=200)
    parser.add_argument("--interval", type=float, default=30.0)
    parser.add_argument("--max-rate", type=float, default=8.0)
    parser.add_argument("--fault-rate", type=float, default=0.02)
    parser.add_argument("--dropout", type=float, default=0.01)
    args = parser.parse_args()

    run_id = uuid.uuid4().hex[:8]
    prefix = args.prefix or f"connectmq/demo/{run_id}"
    expected = args.devices / args.interval
    if expected > args.max_rate:
        raise SystemExit(f"{expected:.1f} msg/s exceeds --max-rate {args.max_rate}")
    print(f"publishing to {prefix}/# at about {expected:.1f} msg/s")

    sensors = [VirtualSensor(i, args.fault_rate) for i in range(args.devices)]
    client = build_client(args, run_id)
    client.connect_async(args.host, args.port, keepalive=30)
    client.loop_start()

    start = time.time()
    # Repartir las primeras lecturas de forma uniforme a lo largo de un intervalo
    schedule = [(start + i * args.interval / args.devices, i) for i in range(args.devices)]
    heapq.heapify(schedule)
    min_gap = 1.0 / args.max_rate
    last_send = 0.0

    try:
        while True:
            due, i = heapq.heappop(schedule)
            now = time.time()
            wait = max(due - now, last_send + min_gap - now)
            if wait > 0:
                time.sleep(wait)
            sensor = sensors[i]
            now = time.time()
            if random.random() >= args.dropout and client.is_connected():
                topic = f"{prefix}/{sensor.line}/{sensor.device_id}/telemetry"
                client.publish(topic, json.dumps(sensor.read(now)), qos=0)
                last_send = time.time()
            heapq.heappush(schedule, (due + args.interval, i))
    except KeyboardInterrupt:
        print("stopping")
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()

Ejecútalo con los valores por defecto:

python simulator.py --devices 200 --interval 30

El script imprime el prefijo elegido, por ejemplo connectmq/demo/3f9a1c2e/#. Abre una segunda terminal y observa el flujo:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-3f9a1c2e -t "connectmq/demo/3f9a1c2e/#" -v

Cómo se consigue el realismo

Cada lectura es la suma de cuatro componentes, y cada uno existe para poner a prueba una parte concreta de tu pipeline.

• Ciclo diario. Una onda sinusoidal con un periodo de 24 horas y una fase aleatoria por dispositivo. Tus gráficas tienen una forma visible y las consultas de agregación, como las medias por hora, devuelven una variación significativa en lugar de una línea plana.
• Deriva acotada. Un paseo aleatorio limitado a más o menos tres grados. Imita sensores que pierden calibración poco a poco, que es justo lo que necesita ver la lógica de detección de deriva.
• Ruido de medición. Ruido gaussiano con una desviación típica pequeña. El suavizado, las medias móviles y los filtros de banda muerta solo tienen sentido cuando la entrada tiene ruido.
• Fallos y cortes. Alrededor del 2 por ciento de los dispositivos se marcan como defectuosos y emiten de vez en cuando picos de más 60 o menos 40 grados. Aparte, el 1 por ciento de las lecturas programadas simplemente se omite. Los picos prueban la validación y los umbrales de alerta; los huecos prueban la detección de datos obsoletos y cómo representan tus gráficas los datos que faltan.

Como el estado vive en cada objeto VirtualSensor, los valores de un dispositivo son continuos en el tiempo. Eso importa cuando pruebas cualquier cosa que compare el valor actual con el anterior, como las alertas de tasa de cambio.

Errores habituales y cómo evitarlos

Publicación a ráfagas. Si todos los dispositivos publican en el segundo cero de cada intervalo, el broker ve una ráfaga de cientos de mensajes seguida de silencio. Las flotas reales rara vez están sincronizadas así, y una ráfaga chocará con los límites por cliente. El planificador escalona la primera lectura de cada dispositivo a lo largo del intervalo, lo que mantiene la tasa uniforme.

Confusión con los relojes. El payload lleva su propia marca de tiempo en milisegundos. Las herramientas posteriores, como Telegraf o un puente en Python, pueden usarla en lugar de la hora de llegada, lo que mantiene correctas las gráficas aunque los mensajes se retrasen. Asegúrate de que todos los consumidores coinciden en la unidad: aquí milisegundos, no segundos.

Reutilizar client IDs. Si arrancas dos copias del simulador con el mismo --client-id, el broker desconecta la sesión antigua cada vez que se conecta la nueva y ambas se pelearán. Deja que el script genere un ID aleatorio en el broker público, o usa el ID emitido para tus credenciales en uno privado.

Enviar datos sensibles a un broker público. Todo lo que hay en el broker público puede leerlo cualquiera que adivine el topic o use comodines. Los datos simulados están bien ahí; los datos reales de clientes o de planta, no.

Publicar estando desconectado. El bucle comprueba client.is_connected() y omite las lecturas mientras la conexión está caída, igual que haría un sensor real sin búfer. Si quieres probar el comportamiento con búfer, lee la guía de reconexión de paho MQTT y añade una cola offline.

Apuntar el simulador a tu propio broker

El broker público es ideal para empezar rápido, pero para prototipar carga sostenida te conviene un entorno aislado. Con un broker MQTT gestionado de ConnectMQ obtienes tus propias credenciales y un endpoint TLS:

python simulator.py \
  --host broker.connectmq.com --port 8883 --tls \
  --username YOUR_USERNAME --password YOUR_PASSWORD --client-id YOUR_CLIENT_ID \
  --prefix connectmq/factory1/production \
  --devices 500 --interval 60 --max-rate 9

El puerto 8883 usa TLS con un certificado de Let's Encrypt, así que basta con tls_set() y el almacén de confianza del sistema. En un broker MQTT gestionado los client IDs están vinculados a las credenciales que se te emiten, así que pasa el ID del portal en lugar de inventarte uno. El cliente integrado en el navegador y el explorador de topics en vivo del portal son una forma rápida de confirmar que los dispositivos simulados aparecen bajo los topics esperados. El plan Serverless (Free) admite hasta 25 conexiones con QoS 0 o 1, suficiente para repartir una flota simulada grande entre varios procesos si cada uno usa sus propias credenciales emitidas.

Adónde enviar después los datos simulados

Un simulador solo resulta interesante cuando algo consume sus datos. Siguientes pasos naturales:

1. Interpreta los payloads en Python como se muestra en trabajar con JSON de MQTT en Python y calcula estadísticas con pandas.
2. Persiste el flujo con Telegraf e InfluxDB usando el stack TIG con Docker Compose y construye paneles de Grafana sobre él.
3. Compara este script con utilidades de benchmarking dedicadas en nuestro resumen de herramientas de pruebas de carga MQTT cuando necesites medir el rendimiento del broker en lugar de generar datos realistas.

Para una visión más amplia de las opciones de cliente en Python y otros lenguajes, la guía de bibliotecas cliente MQTT compara las principales bibliotecas por lenguaje, y la documentación de paho-mqtt cubre todas las opciones del cliente usadas arriba.

Empieza con ConnectMQ

Empieza con el servidor público en broker.connectmq.com para ver tu flota simulada en minutos. Cuando estés listo para credenciales privadas, TLS y tus propias cuotas, ConnectMQ ofrece un broker MQTT gestionado en la nube. Crea un broker Serverless gratuito en app.connectmq.com sin tarjeta de crédito, copia el usuario, la contraseña y el client ID generados y apunta el simulador hacia él. Si aún estás comparando opciones de alojamiento, nuestra guía del broker MQTT en la nube explica en qué fijarse.

Preguntas frecuentes

¿Cuántos dispositivos puede publicar un simulador de dispositivos MQTT desde un solo cliente?

Depende del intervalo y del límite por cliente del broker. En el broker público, mantén el número de dispositivos dividido entre el intervalo por debajo de 10 mensajes por segundo, por ejemplo 200 dispositivos cada 30 segundos o 500 dispositivos cada 60 segundos.

¿Debe el simulador usar QoS 0 o QoS 1?

QoS 0 sirve para la mayoría de simulaciones de telemetría y refleja lo que hacen muchos sensores reales. Usa QoS 1 cuando quieras probar específicamente el manejo de acuses de recibo o la detección de duplicados aguas abajo.

¿Puedo simular dispositivos que se desconectan?

Sí. Aumenta --dropout para obtener huecos aleatorios, o añade un mensaje Last Will por cliente para que los suscriptores reciban un estado offline cuando se mata un proceso del simulador.

¿Es seguro ejecutar el simulador en el broker público?

Es seguro siempre que los datos sean sintéticos. El broker público puede leerlo cualquiera, así que nunca publiques allí datos reales de clientes, de planta ni personales.