Para generar datos de prueba MQTT realistas, modela cada sensor como una línea base más un ciclo diario, ruido de medida, una deriva lenta y algún pico ocasional; después codifica la lectura igual que lo hacen tus dispositivos — JSON o un struct binario empaquetado — y publícala en los mismos topics. Así los dashboards, las reglas y el almacenamiento reciben algo creíble mucho antes de que exista el hardware real.
Esta guía construye ese generador en Python, añade una codificación binaria y un conjunto de payloads deliberadamente rotos para pruebas negativas, y muestra una versión compacta en Node.js.
Qué hace que los datos de un sensor parezcan reales
random.uniform(15, 30) produce valores que saltan sin ninguna estructura. Los sensores reales se comportan de otra forma, y esas diferencias son justo lo que rompe el código que consume los datos:
| Propiedad | Ejemplo real | Qué pone a prueba |
|---|---|---|
| Línea base | Una sala ronda los 21 °C | Rangos por defecto de gráficos, umbrales razonables |
| Ciclo diario | Más calor por la tarde, más frío de noche | Agregación por hora/día, líneas de tendencia |
| Ruido | Variación de ±0,1–0,2 °C entre muestras | Suavizado, bandas muertas, detección de cambios |
| Deriva | La calibración se desplaza poco a poco durante semanas | Detección de anomalías, lógica de recalibración |
| Picos | Se abre una puerta, el sensor falla un instante | Filtros de outliers, deduplicación de alertas |
| Cuantización | El sensor informa en pasos de 0,1 °C o 0,5 % | Comparaciones de igualdad, bugs de redondeo |
| Huecos | Lecturas perdidas en un enlace débil | Gestión de datos faltantes, control de secuencia |
| Desfase de reloj | Las marcas de tiempo del dispositivo se desvían unos cientos de ms | Ordenación, uniones por ventana temporal |
Un modelo de sensor en Python
El modelo usa un generador aleatorio por dispositivo, así que cada ejecución con la misma semilla produce exactamente los mismos datos. La deriva es un paseo aleatorio con reversión a la media: se mueve, pero no se dispara hasta el infinito durante una prueba de varios días.
import math, random
from dataclasses import dataclass
@dataclass
class Sensor:
base: float # valor típico
daily_amp: float # amplitud del ciclo día/noche
noise: float # desviación estándar del ruido de medida
drift_step: float # paso del paseo aleatorio por muestra
lo: float # límite físico inferior
hi: float # límite físico superior
resolution: float # paso de cuantización, p. ej. 0.1
rng: random.Random
spike_prob: float = 0.002
drift: float = 0.0
def read(self, t: float) -> float:
# onda senoidal con máximo a las 15:00 UTC y mínimo a las 03:00 UTC
day = math.sin(2 * math.pi * ((t % 86400) / 86400 - 0.375))
self.drift = 0.999 * self.drift + self.rng.gauss(0, self.drift_step)
v = self.base + self.daily_amp * day + self.drift + self.rng.gauss(0, self.noise)
if self.rng.random() < self.spike_prob:
v += self.rng.choice((-1, 1)) * (self.hi - self.lo) * 0.2
v = min(max(v, self.lo), self.hi)
return round(round(v / self.resolution) * self.resolution, 3)
def make_device(seed: int) -> dict:
rng = random.Random(seed)
return {
"rng": rng,
"battery_mv": rng.uniform(4000, 4200),
"temperature": Sensor(rng.uniform(19, 24), 2.5, 0.15, 0.02, -40, 85, 0.1, rng),
"humidity": Sensor(rng.uniform(40, 55), -6.0, 0.8, 0.05, 0, 100, 0.5, rng),
}La humedad usa una amplitud negativa porque la humedad relativa suele bajar cuando una sala se calienta: un detalle pequeño, pero hace que las correlaciones de tu analítica se comporten como en la realidad.
Publicar telemetría JSON
El publicador crea unos cuantos dispositivos, genera una lectura por dispositivo y por intervalo, descarta alrededor del 1 % de las lecturas para simular un enlace con pérdidas y añade un pequeño desfase de reloj a las marcas de tiempo. SPEEDUP comprime el tiempo para que un ciclo completo de día y noche pase en pocos minutos durante el desarrollo.
import json, os, time
import paho.mqtt.client as mqtt
from sensors import make_device
HOST, PORT = "broker.connectmq.com", 1883
N_DEVICES, INTERVAL, SPEEDUP = 5, 2.0, 360 # 360x: un día simulado cada 4 minutos
devices = {f"sim-{i:03d}": make_device(seed=i) for i in range(N_DEVICES)}
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"datagen-{os.getpid()}")
client.username_pw_set(os.environ.get("MQTT_USER"), os.environ.get("MQTT_PASS"))
client.connect(HOST, PORT)
client.loop_start()
start, seq = time.time(), 0
while True:
sim_t = start + (time.time() - start) * SPEEDUP
for dev_id, d in devices.items():
rng = d["rng"]
if rng.random() < 0.01: # lectura perdida
continue
d["battery_mv"] -= rng.uniform(0, 0.05)
reading = {
"ts": int((sim_t + rng.uniform(-0.3, 0.3)) * 1000),
"seq": seq,
"temperature": d["temperature"].read(sim_t),
"humidity": d["humidity"].read(sim_t),
"battery_mv": int(d["battery_mv"]),
}
client.publish(f"testmqtt/datagen/{dev_id}/telemetry", json.dumps(reading), qos=1)
seq += 1
time.sleep(INTERVAL)Observa el flujo con mosquitto_sub -h broker.connectmq.com -t 'testmqtt/datagen/+/telemetry' -v, o abre el cliente MQTT online y suscríbete a testmqtt/datagen/#. Merece la pena acertar con los nombres de campo y las unidades antes de que los dispositivos salgan al mercado; buenas prácticas de diseño de payloads MQTT repasa las ventajas e inconvenientes.
Codificación binaria con struct
Muchos dispositivos limitados envían binario empaquetado en lugar de JSON. La misma lectura cabe en 14 bytes en vez de unos 90:
| Campo | Tipo | Codificación |
|---|---|---|
| version | uint8 | Versión del formato, empieza en 1 |
| ts | uint32 | Tiempo Unix en segundos |
| seq | uint16 | Vuelve a cero tras 65.535 |
| temperature | int16 | °C × 100 |
| humidity | uint16 | % × 100 |
| battery_mv | uint16 | Milivoltios |
| flags | uint8 | Campo de bits (batería baja, error de sensor…) |
import math, struct
FMT = "<BIHhHHB" # little-endian, sin relleno: 14 bytes
def encode(r: dict, flags: int = 0) -> bytes:
return struct.pack(FMT, 1, r["ts"] // 1000, r["seq"] & 0xFFFF,
round(r["temperature"] * 100), round(r["humidity"] * 100),
r["battery_mv"], flags)
def decode(b: bytes) -> dict:
v, ts, seq, t, h, bat, flags = struct.unpack(FMT, b)
return {"v": v, "ts": ts, "seq": seq, "temperature": t / 100,
"humidity": h / 100, "battery_mv": bat, "flags": flags}
r = {"ts": 1760700000123, "seq": 42, "temperature": 21.4, "humidity": 47.5, "battery_mv": 4093}
packed = encode(r)
assert len(packed) == struct.calcsize(FMT) == 14
assert math.isclose(decode(packed)["temperature"], r["temperature"])
print(packed.hex())En datagen.py, publica encode(reading) en un topic hermano como testmqtt/datagen/sim-001/bin y compruébalo en la red con mosquitto_sub -t 'testmqtt/datagen/+/bin' -F '%t %x', que muestra cada payload en hexadecimal. Indica siempre el orden de bytes (< aquí): un ESP32 y la mayoría de los chips ARM son little-endian, pero un decodificador que asuma el orden de red producirá basura sin avisar. Si prefieres un formato binario autodescriptivo en lugar de un struct hecho a mano, compara CBOR y MessagePack.
La calculadora de tamaño de paquetes MQTT muestra cuánto cuesta cada codificación por mensaje una vez incluidos la cabecera fija y el topic.
Payloads rotos para pruebas negativas
Los datos realistas prueban el camino feliz. Los consumidores también tienen que sobrevivir a los payloads que una flota real acaba enviando tarde o temprano: un bug de firmware, un mensaje truncado, la codificación equivocada. Publica un conjunto de payloads defectuosos conocidos y verifica que tu suscriptor los registra y sigue funcionando:
import json, os
import paho.mqtt.client as mqtt
BAD = {
"empty": b"",
"truncated": b'{"temperature": 21.',
"string-number": b'{"temperature": "21.4"}',
"nan": b'{"temperature": NaN}', # el json de Python lo acepta, JSON.parse no
"huge-number": b'{"temperature": 1e999}',
"not-utf8": '{"site": "café"}'.encode("latin-1"),
"wrong-format": bytes.fromhex("01a0e4f1680a2a005c08"), # binario donde se espera JSON
"oversized": json.dumps({"temperature": 21.4, "pad": "x" * 200_000}).encode(),
}
c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
c.username_pw_set(os.environ.get("MQTT_USER"), os.environ.get("MQTT_PASS"))
c.connect("broker.connectmq.com", 1883)
c.loop_start()
for name, payload in BAD.items():
c.publish("testmqtt/datagen/sim-bad/telemetry", payload, qos=1).wait_for_publish()
print("sent", name, len(payload), "bytes")
c.loop_stop()
c.disconnect()Cada caso corresponde a un fallo real: el parser lanza una excepción, un string se cuela en una columna numérica, NaN rompe una inserción en la base de datos o un payload grande supera el tamaño máximo de paquete del broker. Errores de parseo JSON en MQTT explica cómo tratar cada uno en el consumidor.
La misma idea en Node.js
Con MQTT.js 5 el generador ocupa pocas líneas. Box–Muller convierte Math.random() en ruido gaussiano:
import mqtt from "mqtt";
const client = await mqtt.connectAsync("mqtt://broker.connectmq.com:1883", {
username: process.env.MQTT_USER,
password: process.env.MQTT_PASS,
});
const gauss = () => Math.sqrt(-2 * Math.log(1 - Math.random())) * Math.cos(2 * Math.PI * Math.random());
let drift = 0, seq = 0;
setInterval(async () => {
const t = Date.now() / 1000;
drift = 0.999 * drift + gauss() * 0.02;
const day = Math.sin(2 * Math.PI * ((t % 86400) / 86400 - 0.375));
const temperature = Math.round((21 + 2.5 * day + drift + gauss() * 0.15) * 10) / 10;
const payload = JSON.stringify({ ts: Date.now(), seq: seq++, temperature });
await client.publishAsync("testmqtt/datagen/js-001/telemetry", payload, { qos: 1 });
}, 2000);Datos sintéticos frente a datos grabados
Los datos sintéticos son ideales al principio y para escenarios difíciles de provocar, como 10.000 dispositivos o un sensor que falla a las 3 de la madrugada. En cuanto tengas una flota piloto funcionando, graba su tráfico real y reprodúcelo también; pruebas de replay MQTT muestra cómo. Las mejores baterías de pruebas usan ambos: datos grabados para el realismo y datos generados para la escala y los casos límite. Para convertir este generador en muchos dispositivos virtuales independientes, cada uno con su conexión y su Last Will, consulta crear un simulador de dispositivos MQTT en Python.
Mira la telemetría generada en directo
Un buen generador de datos de prueba cuesta una tarde y se amortiza en cada dashboard, regla y esquema de base de datos que construyas. Modela la señal con honestidad, usa semillas, codifícala exactamente como el dispositivo y guarda una carpeta de payloads rotos a su lado.
Preguntas frecuentes
¿Por qué no publicar simplemente números aleatorios como datos de prueba?
Los valores aleatorios uniformes no tienen tendencia, ni ciclo diario, ni correlación entre muestras, así que nunca activan la lógica que de verdad necesitas probar: umbrales, suavizado, detección de anomalías, bandas muertas o escalado de gráficos. Un modelo con línea base, ciclo, ruido, deriva y picos ocasionales se parece a un sensor real y ejercita esa lógica.
¿Los datos de prueba MQTT deben ser JSON o binarios?
Usa el formato que envían realmente tus dispositivos. JSON es más fácil de leer y depurar; un struct binario empaquetado es varias veces más pequeño y es habitual en dispositivos limitados o con conexión celular. Un buen generador produce ambos a partir de la misma lectura para que puedas probar los dos decodificadores.
¿Cómo hago que los datos de prueba generados sean reproducibles?
Usa una semilla para el generador aleatorio de cada dispositivo, por ejemplo su índice. La misma semilla produce la misma secuencia de lecturas, de modo que una prueba fallida se puede repetir exactamente y dos ejecuciones de un benchmark se comparan en igualdad de condiciones.
