Para gerar dados de teste MQTT realistas, modele cada sensor como uma linha de base mais um ciclo diário, ruído de medição, deriva lenta e um pico ocasional; depois codifique a leitura do mesmo jeito que seus dispositivos fazem — JSON ou um struct binário compactado — e publique nos mesmos topics. Assim dashboards, regras e armazenamento recebem algo crível muito antes de o hardware real existir.
Este guia monta esse gerador em Python, adiciona uma codificação binária e um conjunto de payloads propositalmente quebrados para testes negativos, e mostra uma versão compacta em Node.js.
O que faz dados de sensor parecerem reais
random.uniform(15, 30) gera valores que pulam sem nenhuma estrutura. Sensores reais se comportam de outro jeito, e essas diferenças são exatamente o que quebra o código que consome os dados:
| Propriedade | Exemplo real | O que testa |
|---|---|---|
| Linha de base | Uma sala fica em torno de 21 °C | Faixas padrão de gráficos, limites razoáveis |
| Ciclo diário | Mais quente à tarde, mais frio à noite | Agregação por hora/dia, linhas de tendência |
| Ruído | Variação de ±0,1–0,2 °C entre amostras | Suavização, zonas mortas, detecção de mudança |
| Deriva | A calibração vai se deslocando ao longo de semanas | Detecção de anomalias, lógica de recalibração |
| Picos | Uma porta se abre, o sensor falha por um instante | Filtros de outliers, deduplicação de alertas |
| Quantização | O sensor reporta em passos de 0,1 °C ou 0,5 % | Comparações de igualdade, bugs de arredondamento |
| Lacunas | Leituras perdidas num link fraco | Tratamento de dados ausentes, checagem de sequência |
| Desvio de relógio | Timestamps do dispositivo erram por algumas centenas de ms | Ordenação, joins por janela de tempo |
Um modelo de sensor em Python
O modelo abaixo usa um gerador aleatório por dispositivo, então cada execução com a mesma seed produz exatamente os mesmos dados. A deriva é um passeio aleatório com reversão à média: ela varia, mas não dispara para o infinito durante um teste de vários dias.
import math, random
from dataclasses import dataclass
@dataclass
class Sensor:
base: float # valor típico
daily_amp: float # amplitude do ciclo dia/noite
noise: float # desvio padrão do ruído de medição
drift_step: float # passo do passeio aleatório por amostra
lo: float # limite físico inferior
hi: float # limite físico superior
resolution: float # passo de quantização, ex. 0.1
rng: random.Random
spike_prob: float = 0.002
drift: float = 0.0
def read(self, t: float) -> float:
# senoide com pico às 15:00 UTC e mínimo às 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),
}A umidade usa amplitude negativa porque a umidade relativa costuma cair quando a sala esquenta — um detalhe pequeno, mas que faz as correlações da sua análise se comportarem como na vida real.
Publicando telemetria JSON
O publisher cria alguns dispositivos, gera uma leitura por dispositivo a cada intervalo, descarta cerca de 1 % das leituras para simular um link com perdas e adiciona um pouco de desvio de relógio aos timestamps. SPEEDUP comprime o tempo para que um ciclo completo de dia e noite passe em poucos minutos durante o desenvolvimento.
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: um dia simulado a 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: # leitura 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)Acompanhe o fluxo com mosquitto_sub -h broker.connectmq.com -t 'testmqtt/datagen/+/telemetry' -v, ou abra o cliente MQTT online e assine testmqtt/datagen/#. Vale acertar nomes de campos e unidades antes de os dispositivos irem para o campo; boas práticas de design de payload MQTT discute os trade-offs.
Codificação binária com struct
Muitos dispositivos limitados enviam binário compactado em vez de JSON. A mesma leitura cabe em 14 bytes em vez de cerca de 90:
| Campo | Tipo | Codificação |
|---|---|---|
| version | uint8 | Versão do formato, começa em 1 |
| ts | uint32 | Tempo Unix em segundos |
| seq | uint16 | Volta a zero depois de 65.535 |
| temperature | int16 | °C × 100 |
| humidity | uint16 | % × 100 |
| battery_mv | uint16 | Milivolts |
| flags | uint8 | Campo de bits (bateria fraca, erro de sensor…) |
import math, struct
FMT = "<BIHhHHB" # little-endian, sem padding: 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())No datagen.py, publique encode(reading) num topic irmão como testmqtt/datagen/sim-001/bin e confira no fio com mosquitto_sub -t 'testmqtt/datagen/+/bin' -F '%t %x', que mostra cada payload em hexadecimal. Sempre declare a ordem dos bytes (< aqui): um ESP32 e a maioria dos chips ARM são little-endian, mas um decodificador que assume a ordem de rede vai produzir lixo sem avisar. Se você prefere um formato binário autodescritivo em vez de um struct feito à mão, compare CBOR e MessagePack.
A calculadora de tamanho de pacote MQTT mostra quanto cada codificação custa por mensagem, já incluindo o cabeçalho fixo e o topic.
Payloads quebrados para testes negativos
Dados realistas testam o caminho feliz. Os consumidores também precisam sobreviver aos payloads que frotas reais acabam enviando — um bug de firmware, uma mensagem truncada, a codificação errada. Publique um conjunto de payloads sabidamente ruins e verifique se o seu subscriber registra o problema e continua rodando:
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}', # o json do Python aceita, JSON.parse não
"huge-number": b'{"temperature": 1e999}',
"not-utf8": '{"site": "café"}'.encode("latin-1"),
"wrong-format": bytes.fromhex("01a0e4f1680a2a005c08"), # binário onde 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 uma falha real: o parser lança exceção, uma string entra numa coluna numérica, NaN quebra um insert no banco ou um payload grande ultrapassa o tamanho máximo de pacote do broker. Erros de parse de JSON no MQTT explica como tratar cada um no consumidor.
A mesma ideia em Node.js
Com MQTT.js 5, o gerador cabe em poucas linhas. Box–Muller transforma Math.random() em ruído 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);Dados sintéticos vs dados gravados
Dados sintéticos são ideais no começo e para cenários difíceis de produzir, como 10.000 dispositivos ou um sensor falhando às 3 da manhã. Assim que uma frota piloto estiver rodando, grave o tráfego real dela e reproduza também — testes de replay MQTT mostra como. As melhores suítes de teste usam os dois: dados gravados pelo realismo, dados gerados pela escala e pelos casos extremos. Para transformar este gerador em muitos dispositivos virtuais independentes, cada um com sua própria conexão e Last Will, veja como criar um simulador de dispositivos MQTT em Python.
Veja a telemetria gerada ao vivo
Um bom gerador de dados de teste custa uma tarde e se paga em cada dashboard, regra e schema de banco que você construir. Modele o sinal com honestidade, use seeds, codifique exatamente como o dispositivo e mantenha uma pasta de payloads quebrados ao lado.
Perguntas frequentes
Por que não publicar simplesmente números aleatórios como dados de teste?
Valores aleatórios uniformes não têm tendência, ciclo diário nem correlação entre amostras, então nunca acionam a lógica que você realmente precisa testar: limites, suavização, detecção de anomalias, zonas mortas ou escala de gráficos. Um modelo com linha de base, ciclo, ruído, deriva e picos ocasionais parece um sensor real e exercita essa lógica.
Os dados de teste MQTT devem ser JSON ou binários?
Use o formato que seus dispositivos realmente enviam. JSON é mais fácil de ler e depurar; um struct binário compactado é várias vezes menor e comum em dispositivos limitados ou celulares. Um bom gerador produz os dois a partir da mesma leitura, para que você teste ambos os decodificadores.
Como deixar os dados de teste gerados reproduzíveis?
Defina uma seed para o gerador aleatório de cada dispositivo, por exemplo o índice dele. A mesma seed produz a mesma sequência de leituras, então um teste que falhou pode ser repetido exatamente e duas execuções de um benchmark comparam a mesma coisa.
