To generate realistic MQTT test data, model each sensor as a baseline plus a daily cycle, measurement noise, slow drift and the occasional spike, then encode the reading the same way your devices do — JSON or a packed binary struct — and publish it on the same topics. That gives dashboards, rules and storage something believable to chew on, long before real hardware exists.
This guide builds that generator in Python, adds a binary encoding and a set of deliberately broken payloads for negative testing, and shows a compact Node.js version.
What makes sensor data look real
random.uniform(15, 30) produces values that jump around without any structure. Real sensors behave differently, and the differences are exactly what breaks downstream code:
| Property | Real-world example | What it tests |
|---|---|---|
| Baseline | A room sits around 21 °C | Default chart ranges, sane thresholds |
| Daily cycle | Warmer in the afternoon, cooler at night | Aggregation per hour/day, trend lines |
| Noise | ±0.1–0.2 °C jitter between samples | Smoothing, deadbands, change detection |
| Drift | Calibration slowly wanders over weeks | Anomaly detection, recalibration logic |
| Spikes | A door opens, a sensor glitches | Outlier filters, alert de-duplication |
| Quantization | The sensor reports in 0.1 °C or 0.5 % steps | Equality checks, rounding bugs |
| Gaps | Readings lost on a weak link | Missing-data handling, sequence checks |
| Clock jitter | Device timestamps are off by a few hundred ms | Ordering, time-window joins |
A sensor model in Python
The model below keeps one random generator per device, so every run with the same seed produces exactly the same data. Drift is a mean-reverting random walk: it wanders but does not run away to infinity during a multi-day test.
import math, random
from dataclasses import dataclass
@dataclass
class Sensor:
base: float # typical value
daily_amp: float # amplitude of the day/night cycle
noise: float # standard deviation of measurement noise
drift_step: float # random-walk step per sample
lo: float # physical lower limit
hi: float # physical upper limit
resolution: float # quantization step, e.g. 0.1
rng: random.Random
spike_prob: float = 0.002
drift: float = 0.0
def read(self, t: float) -> float:
# sine wave peaking at 15:00 UTC, lowest at 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),
}Humidity uses a negative amplitude because relative humidity usually falls as a room warms up — a small detail, but it means correlations in your analytics behave like the real thing.
Publishing JSON telemetry
The publisher creates a handful of devices, produces one reading per device per interval, drops about 1 % of readings to simulate a lossy link, and adds a little clock jitter to the timestamps. SPEEDUP compresses time so a full day/night cycle passes in a few minutes during development.
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: one simulated day every 4 minutes
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: # lost reading
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)Watch the stream with mosquitto_sub -h broker.connectmq.com -t 'testmqtt/datagen/+/telemetry' -v, or open the online MQTT client and subscribe to testmqtt/datagen/#. Field naming and units are worth getting right before devices ship; MQTT payload design best practices covers the trade-offs.
Binary encoding with struct
Many constrained devices send packed binary instead of JSON. The same reading fits in 14 bytes instead of about 90:
| Field | Type | Encoding |
|---|---|---|
| version | uint8 | Format version, start at 1 |
| ts | uint32 | Unix time in seconds |
| seq | uint16 | Wraps at 65,535 |
| temperature | int16 | °C × 100 |
| humidity | uint16 | % × 100 |
| battery_mv | uint16 | Millivolts |
| flags | uint8 | Bit field (low battery, sensor error…) |
import math, struct
FMT = "<BIHhHHB" # little-endian, no 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())In datagen.py, publish encode(reading) to a sibling topic such as testmqtt/datagen/sim-001/bin and check it on the wire with mosquitto_sub -t 'testmqtt/datagen/+/bin' -F '%t %x', which prints each payload as hex. Always state the byte order (< here): an ESP32 and most ARM chips are little-endian, but a decoder that assumes network byte order will silently produce nonsense. If you want a self-describing binary format instead of a hand-made struct, compare CBOR and MessagePack.
The MQTT packet size calculator shows what the two encodings cost per message once the fixed header and topic are included.
Broken payloads for negative testing
Realistic data tests the happy path. Consumers also need to survive the payloads that real fleets eventually send — a firmware bug, a truncated message, the wrong encoding. Publish a set of known-bad payloads and verify that your subscriber logs them and keeps running:
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}', # Python's json accepts it, JSON.parse does not
"huge-number": b'{"temperature": 1e999}',
"not-utf8": '{"site": "café"}'.encode("latin-1"),
"wrong-format": bytes.fromhex("01a0e4f1680a2a005c08"), # binary where JSON is expected
"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()Each case maps to a real failure mode: the parser throws, a string sneaks into a numeric column, NaN breaks a database insert, or a large payload exceeds the broker's maximum packet size. MQTT JSON parse errors explains how to handle each one on the consumer side.
The same idea in Node.js
With MQTT.js 5 the generator is a few lines. Box–Muller turns Math.random() into Gaussian noise:
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);Synthetic data vs recorded data
Synthetic data is ideal early on and for scenarios you cannot easily produce, such as 10,000 devices or a sensor failing at 3 a.m. Once a pilot fleet is running, record its real traffic and replay it as well — MQTT replay testing shows how. The best test suites use both: recorded data for realism, generated data for scale and edge cases. To turn this generator into many independent virtual devices with their own connections and Last Will, see build an MQTT device simulator in Python.
Watch generated telemetry live
A good test-data generator costs an afternoon and pays back on every dashboard, rule and database schema you build. Model the signal honestly, seed it, encode it exactly like the device, and keep a folder of broken payloads next to it.
Frequently asked questions
Why not just publish random numbers as test data?
Uniform random values have no trend, no daily cycle and no correlation between samples, so they never trigger the logic you actually need to test: thresholds, smoothing, anomaly detection, deadbands or chart scaling. A model of baseline, cycle, noise, drift and occasional spikes looks like a real sensor and exercises that logic.
Should MQTT test data be JSON or binary?
Use the format your devices really send. JSON is easiest to read and debug; a packed binary struct is several times smaller and is common on constrained or cellular devices. A good generator can produce both from the same reading so you can test both decoders.
How do I make generated test data reproducible?
Seed the random number generator per device, for example with the device index. The same seed produces the same sequence of readings, so a failing test can be replayed exactly and two runs of a benchmark compare like with like.
