Um realistische MQTT-Testdaten zu erzeugen, modellierst du jeden Sensor als Grundwert plus Tageszyklus, Messrauschen, langsame Drift und gelegentliche Ausreißer, kodierst den Messwert genau wie deine Geräte – JSON oder ein gepacktes Binär-Struct – und publizierst ihn auf denselben Topics. So bekommen Dashboards, Regeln und Speicher etwas Glaubwürdiges zu verarbeiten, lange bevor echte Hardware existiert.
Dieser Leitfaden baut den Generator in Python, ergänzt eine Binärkodierung und eine Sammlung absichtlich kaputter Payloads für Negativtests und zeigt eine kompakte Node.js-Variante.
Was Sensordaten realistisch macht
random.uniform(15, 30) liefert Werte, die ohne jede Struktur springen. Echte Sensoren verhalten sich anders – und genau diese Unterschiede bringen nachgelagerten Code zum Scheitern:
| Eigenschaft | Beispiel aus der Praxis | Was damit getestet wird |
|---|---|---|
| Grundwert | Ein Raum liegt bei etwa 21 °C | Standard-Diagrammbereiche, sinnvolle Schwellwerte |
| Tageszyklus | Nachmittags wärmer, nachts kühler | Aggregation pro Stunde/Tag, Trendlinien |
| Rauschen | ±0,1–0,2 °C Schwankung zwischen Messwerten | Glättung, Totbänder, Änderungserkennung |
| Drift | Die Kalibrierung wandert über Wochen | Anomalieerkennung, Rekalibrierungslogik |
| Ausreißer | Eine Tür geht auf, der Sensor hat einen Aussetzer | Ausreißerfilter, Deduplizierung von Alarmen |
| Quantisierung | Der Sensor meldet in Schritten von 0,1 °C oder 0,5 % | Gleichheitsvergleiche, Rundungsfehler |
| Lücken | Messwerte gehen auf einer schwachen Verbindung verloren | Umgang mit fehlenden Daten, Sequenzprüfung |
| Uhrabweichung | Geräte-Zeitstempel liegen ein paar hundert ms daneben | Sortierung, Joins über Zeitfenster |
Ein Sensormodell in Python
Das Modell unten nutzt pro Gerät einen eigenen Zufallsgenerator; jeder Lauf mit demselben Seed liefert also exakt dieselben Daten. Die Drift ist ein Random Walk mit Rückkehr zum Mittelwert: Sie wandert, läuft aber in einem mehrtägigen Test nicht ins Unendliche davon.
import math, random
from dataclasses import dataclass
@dataclass
class Sensor:
base: float # typischer Wert
daily_amp: float # Amplitude des Tag-Nacht-Zyklus
noise: float # Standardabweichung des Messrauschens
drift_step: float # Random-Walk-Schritt pro Messwert
lo: float # physikalische Untergrenze
hi: float # physikalische Obergrenze
resolution: float # Quantisierungsschritt, z. B. 0.1
rng: random.Random
spike_prob: float = 0.002
drift: float = 0.0
def read(self, t: float) -> float:
# Sinuswelle mit Maximum um 15:00 UTC und Minimum um 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),
}Die Luftfeuchtigkeit hat eine negative Amplitude, weil die relative Feuchte meist sinkt, wenn sich ein Raum erwärmt – ein kleines Detail, aber dadurch verhalten sich Korrelationen in deiner Auswertung wie in der Realität.
JSON-Telemetrie publizieren
Der Publisher legt ein paar Geräte an, erzeugt pro Gerät und Intervall einen Messwert, verwirft etwa 1 % der Werte, um eine verlustbehaftete Verbindung zu simulieren, und versieht die Zeitstempel mit etwas Uhrabweichung. SPEEDUP staucht die Zeit, sodass während der Entwicklung ein ganzer Tag-Nacht-Zyklus in wenigen Minuten vergeht.
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: ein simulierter Tag alle 4 Minuten
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: # verlorener Messwert
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)Beobachte den Datenstrom mit mosquitto_sub -h broker.connectmq.com -t 'testmqtt/datagen/+/telemetry' -v oder öffne den Online-MQTT-Client und abonniere testmqtt/datagen/#. Feldnamen und Einheiten solltest du festlegen, bevor die Geräte ausgeliefert werden; Best Practices für MQTT-Payload-Design erklärt die Abwägungen.
Binärkodierung mit struct
Viele ressourcenarme Geräte senden gepackte Binärdaten statt JSON. Derselbe Messwert passt in 14 Bytes statt in rund 90:
| Feld | Typ | Kodierung |
|---|---|---|
| version | uint8 | Formatversion, beginnt bei 1 |
| ts | uint32 | Unix-Zeit in Sekunden |
| seq | uint16 | Läuft nach 65.535 über |
| temperature | int16 | °C × 100 |
| humidity | uint16 | % × 100 |
| battery_mv | uint16 | Millivolt |
| flags | uint8 | Bitfeld (Akku schwach, Sensorfehler…) |
import math, struct
FMT = "<BIHhHHB" # Little-Endian, ohne 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())Publiziere in datagen.py zusätzlich encode(reading) auf ein Nachbar-Topic wie testmqtt/datagen/sim-001/bin und prüfe es auf der Leitung mit mosquitto_sub -t 'testmqtt/datagen/+/bin' -F '%t %x', das jedes Payload als Hex ausgibt. Gib die Byte-Reihenfolge immer explizit an (< hier): Ein ESP32 und die meisten ARM-Chips sind Little-Endian, aber ein Decoder, der Network Byte Order annimmt, produziert stillschweigend Unsinn. Wenn du lieber ein selbstbeschreibendes Binärformat statt eines handgebauten Structs willst, vergleiche CBOR und MessagePack.
Der MQTT-Paketgrößen-Rechner zeigt, was beide Kodierungen pro Nachricht inklusive Fixed Header und Topic kosten.
Kaputte Payloads für Negativtests
Realistische Daten testen den Normalfall. Consumer müssen aber auch die Payloads überstehen, die echte Flotten irgendwann senden – ein Firmware-Bug, eine abgeschnittene Nachricht, die falsche Kodierung. Publiziere eine Reihe bekannt fehlerhafter Payloads und prüfe, ob dein Subscriber sie protokolliert und weiterläuft:
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}', # Pythons json akzeptiert es, JSON.parse nicht
"huge-number": b'{"temperature": 1e999}',
"not-utf8": '{"site": "café"}'.encode("latin-1"),
"wrong-format": bytes.fromhex("01a0e4f1680a2a005c08"), # Binärdaten, wo JSON erwartet wird
"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()Jeder Fall entspricht einem echten Fehlerbild: Der Parser wirft eine Exception, ein String rutscht in eine numerische Spalte, NaN lässt einen Datenbank-Insert scheitern, oder ein großes Payload überschreitet die maximale Paketgröße des Brokers. MQTT-JSON-Parse-Fehler zeigt, wie du jeden Fall im Consumer behandelst.
Dieselbe Idee in Node.js
Mit MQTT.js 5 ist der Generator in wenigen Zeilen geschrieben. Box–Muller macht aus Math.random() gaußsches Rauschen:
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);Synthetische vs. aufgezeichnete Daten
Synthetische Daten sind ideal am Anfang und für Szenarien, die sich schwer herstellen lassen, etwa 10.000 Geräte oder ein Sensor, der um 3 Uhr nachts ausfällt. Sobald eine Pilotflotte läuft, zeichne ihren echten Traffic auf und spiele ihn ebenfalls ab – MQTT-Replay-Tests zeigt, wie. Die besten Testsuiten nutzen beides: aufgezeichnete Daten für Realismus, generierte Daten für Skalierung und Grenzfälle. Um aus diesem Generator viele unabhängige virtuelle Geräte mit eigener Verbindung und eigenem Last Will zu machen, siehe MQTT-Gerätesimulator in Python bauen.
Generierte Telemetrie live ansehen
Ein guter Testdatengenerator kostet einen Nachmittag und zahlt sich bei jedem Dashboard, jeder Regel und jedem Datenbankschema aus, das du baust. Modelliere das Signal ehrlich, setze Seeds, kodiere genau wie das Gerät und leg einen Ordner mit kaputten Payloads daneben.
Häufig gestellte Fragen
Warum nicht einfach Zufallszahlen als Testdaten publizieren?
Gleichverteilte Zufallswerte haben keinen Trend, keinen Tageszyklus und keine Korrelation zwischen Messwerten. Sie lösen deshalb nie die Logik aus, die du eigentlich testen willst: Schwellwerte, Glättung, Anomalieerkennung, Totbänder oder Diagrammskalierung. Ein Modell aus Grundwert, Zyklus, Rauschen, Drift und gelegentlichen Ausreißern sieht aus wie ein echter Sensor und testet genau diese Logik.
Sollten MQTT-Testdaten JSON oder binär sein?
Nimm das Format, das deine Geräte tatsächlich senden. JSON ist am leichtesten zu lesen und zu debuggen; ein gepacktes Binär-Struct ist um ein Vielfaches kleiner und auf ressourcenarmen oder Mobilfunk-Geräten üblich. Ein guter Generator erzeugt beides aus demselben Messwert, damit du beide Decoder testen kannst.
Wie mache ich generierte Testdaten reproduzierbar?
Gib dem Zufallsgenerator jedes Geräts einen Seed, zum Beispiel den Geräteindex. Derselbe Seed liefert dieselbe Folge von Messwerten, sodass ein fehlgeschlagener Test exakt wiederholt werden kann und zwei Benchmark-Läufe Gleiches mit Gleichem vergleichen.
