Pour générer des données de test MQTT réalistes, modélisez chaque capteur comme une valeur de base à laquelle s'ajoutent un cycle quotidien, du bruit de mesure, une dérive lente et un pic de temps en temps, puis encodez la mesure exactement comme vos appareils — en JSON ou en structure binaire compacte — et publiez-la sur les mêmes topics. Dashboards, règles et stockage reçoivent ainsi des données crédibles bien avant que le matériel réel n'existe.
Ce guide construit ce générateur en Python, ajoute un encodage binaire et une série de payloads volontairement cassés pour les tests négatifs, et présente une version compacte en Node.js.
Ce qui rend des données de capteur réalistes
random.uniform(15, 30) produit des valeurs qui sautent sans aucune structure. Les vrais capteurs se comportent autrement, et ces différences sont justement ce qui casse le code en aval :
| Propriété | Exemple réel | Ce que cela teste |
|---|---|---|
| Valeur de base | Une pièce tourne autour de 21 °C | Plages par défaut des graphiques, seuils cohérents |
| Cycle quotidien | Plus chaud l'après-midi, plus frais la nuit | Agrégation par heure/jour, courbes de tendance |
| Bruit | Variation de ±0,1–0,2 °C entre échantillons | Lissage, bandes mortes, détection de changement |
| Dérive | L'étalonnage se décale lentement au fil des semaines | Détection d'anomalies, logique de réétalonnage |
| Pics | Une porte s'ouvre, le capteur a un raté | Filtres d'outliers, dédoublonnage des alertes |
| Quantification | Le capteur mesure par pas de 0,1 °C ou 0,5 % | Comparaisons d'égalité, bugs d'arrondi |
| Trous | Mesures perdues sur une liaison faible | Gestion des données manquantes, contrôle de séquence |
| Décalage d'horloge | Les horodatages de l'appareil dévient de quelques centaines de ms | Tri, jointures par fenêtre temporelle |
Un modèle de capteur en Python
Le modèle ci-dessous utilise un générateur aléatoire par appareil : chaque exécution avec la même graine produit exactement les mêmes données. La dérive est une marche aléatoire avec retour à la moyenne : elle varie, mais ne part pas à l'infini pendant un test de plusieurs jours.
import math, random
from dataclasses import dataclass
@dataclass
class Sensor:
base: float # valeur typique
daily_amp: float # amplitude du cycle jour/nuit
noise: float # écart type du bruit de mesure
drift_step: float # pas de la marche aléatoire par échantillon
lo: float # limite physique basse
hi: float # limite physique haute
resolution: float # pas de quantification, ex. 0.1
rng: random.Random
spike_prob: float = 0.002
drift: float = 0.0
def read(self, t: float) -> float:
# sinusoïde avec maximum à 15:00 UTC et minimum à 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),
}L'humidité a une amplitude négative parce que l'humidité relative baisse généralement quand une pièce se réchauffe : un petit détail, mais grâce à lui les corrélations de vos analyses se comportent comme dans la réalité.
Publier de la télémétrie JSON
Le publisher crée quelques appareils, produit une mesure par appareil à chaque intervalle, supprime environ 1 % des mesures pour simuler une liaison avec pertes et ajoute un léger décalage d'horloge aux horodatages. SPEEDUP compresse le temps pour qu'un cycle jour/nuit complet passe en quelques minutes pendant le développement.
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 : une journée simulée toutes les 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: # mesure perdue
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)Observez le flux avec mosquitto_sub -h broker.connectmq.com -t 'testmqtt/datagen/+/telemetry' -v, ou ouvrez le client MQTT en ligne et abonnez-vous à testmqtt/datagen/#. Les noms de champs et les unités méritent d'être fixés avant le déploiement des appareils ; les bonnes pratiques de conception de payload MQTT détaillent les compromis.
Encodage binaire avec struct
Beaucoup d'appareils contraints envoient du binaire compact plutôt que du JSON. La même mesure tient en 14 octets au lieu d'environ 90 :
| Champ | Type | Encodage |
|---|---|---|
| version | uint8 | Version du format, commence à 1 |
| ts | uint32 | Temps Unix en secondes |
| seq | uint16 | Repart à zéro après 65 535 |
| temperature | int16 | °C × 100 |
| humidity | uint16 | % × 100 |
| battery_mv | uint16 | Millivolts |
| flags | uint8 | Champ de bits (batterie faible, erreur capteur…) |
import math, struct
FMT = "<BIHhHHB" # little-endian, sans padding : 14 octets
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())Dans datagen.py, publiez encode(reading) sur un topic voisin comme testmqtt/datagen/sim-001/bin et vérifiez-le sur le réseau avec mosquitto_sub -t 'testmqtt/datagen/+/bin' -F '%t %x', qui affiche chaque payload en hexadécimal. Indiquez toujours l'ordre des octets (< ici) : un ESP32 et la plupart des puces ARM sont little-endian, mais un décodeur qui suppose l'ordre réseau produira silencieusement des absurdités. Si vous préférez un format binaire autodescriptif à une structure faite main, comparez CBOR et MessagePack.
Le calculateur de taille de paquet MQTT montre ce que coûte chaque encodage par message, en-tête fixe et topic compris.
Des payloads cassés pour les tests négatifs
Les données réalistes testent le cas nominal. Les consommateurs doivent aussi survivre aux payloads qu'une flotte réelle finit toujours par envoyer : un bug de firmware, un message tronqué, le mauvais encodage. Publiez une série de payloads défectueux connus et vérifiez que votre subscriber les journalise et continue de tourner :
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}', # le json de Python l'accepte, JSON.parse non
"huge-number": b'{"temperature": 1e999}',
"not-utf8": '{"site": "café"}'.encode("latin-1"),
"wrong-format": bytes.fromhex("01a0e4f1680a2a005c08"), # du binaire là où du JSON est attendu
"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()Chaque cas correspond à une panne réelle : le parser lève une exception, une chaîne se glisse dans une colonne numérique, NaN fait échouer une insertion en base, ou un gros payload dépasse la taille maximale de paquet du broker. Les erreurs de parsing JSON dans MQTT explique comment traiter chacun côté consommateur.
La même idée en Node.js
Avec MQTT.js 5, le générateur tient en quelques lignes. Box–Muller transforme Math.random() en bruit gaussien :
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);Données synthétiques ou données enregistrées
Les données synthétiques sont idéales au début et pour les scénarios difficiles à provoquer, comme 10 000 appareils ou un capteur qui tombe en panne à 3 heures du matin. Dès qu'une flotte pilote tourne, enregistrez son trafic réel et rejouez-le aussi — les tests de replay MQTT expliquent comment. Les meilleures suites de tests combinent les deux : données enregistrées pour le réalisme, données générées pour l'échelle et les cas limites. Pour transformer ce générateur en de nombreux appareils virtuels indépendants, chacun avec sa connexion et son Last Will, voir créer un simulateur d'appareils MQTT en Python.
Voir la télémétrie générée en direct
Un bon générateur de données de test coûte un après-midi et se rentabilise sur chaque dashboard, règle et schéma de base de données que vous construisez. Modélisez le signal honnêtement, fixez les graines, encodez exactement comme l'appareil et gardez un dossier de payloads cassés à côté.
Questions fréquentes
Pourquoi ne pas simplement publier des nombres aléatoires comme données de test ?
Des valeurs aléatoires uniformes n'ont ni tendance, ni cycle quotidien, ni corrélation entre échantillons : elles ne déclenchent jamais la logique que vous devez réellement tester, comme les seuils, le lissage, la détection d'anomalies, les bandes mortes ou l'échelle des graphiques. Un modèle avec valeur de base, cycle, bruit, dérive et pics occasionnels ressemble à un vrai capteur et sollicite cette logique.
Les données de test MQTT doivent-elles être en JSON ou en binaire ?
Utilisez le format que vos appareils envoient vraiment. Le JSON est le plus simple à lire et à déboguer ; une structure binaire compacte est plusieurs fois plus petite et courante sur les appareils contraints ou cellulaires. Un bon générateur produit les deux à partir de la même mesure pour tester les deux décodeurs.
Comment rendre les données de test générées reproductibles ?
Initialisez le générateur aléatoire de chaque appareil avec une graine, par exemple son index. La même graine produit la même suite de mesures : un test en échec peut être rejoué à l'identique et deux exécutions d'un benchmark comparent ce qui est comparable.
