Créer un simulateur d'appareils MQTT en Python pour des centaines de capteurs virtuels

Un simulateur d'appareils MQTT vous permet de construire des tableaux de bord, des pipelines de stockage et des règles d'alerte avant même l'installation du premier capteur réel. Dans ce tutoriel, vous allez écrire un simulateur Python qui modélise des centaines de capteurs virtuels avec un bruit, une dérive, des pannes et des coupures crédibles, et les publie sur le broker public broker.connectmq.com sans dépasser ses limites de débit.

Pourquoi simuler des appareils avant d'avoir le matériel

Le matériel est généralement la partie la plus lente d'un projet IoT. Les cartes sont en commande, le firmware est à moitié écrit et l'installateur n'est pas encore passé sur site, alors que les équipes qui construisent le tableau de bord et la base de données ont déjà besoin de données. Un simulateur supprime cette dépendance. Il fournit à chaque composant en aval un flux réaliste sur lequel travailler, et il permet de répéter des situations difficiles à provoquer avec de vrais équipements : un capteur qui grimpe à une valeur impossible, un appareil qui se tait pendant dix minutes ou toute une ligne qui se met en ligne d'un coup.

Les équipes découvrent souvent un second avantage plus tard. Une fois la flotte réelle déployée, le même simulateur devient votre outil de non-régression. Vous pouvez l'exécuter contre un broker de préproduction après chaque modification de votre code d'ingestion et voir immédiatement si un parseur ou un tableau de bord a cassé.

Ce qui distingue un simulateur utile d'un jouet, c'est le réalisme. Une boucle qui publie random.uniform(20, 25) chaque seconde produit des données qui ne ressemblent en rien à celles d'une vraie sonde de température. Les signaux réels évoluent lentement, suivent des cycles journaliers, comportent du bruit de mesure, dérivent au fil des semaines et tombent parfois en panne. Le simulateur ci-dessous modélise chacun de ces comportements en quelques lignes de code.

Conception du simulateur d'appareils MQTT

Avant d'écrire du code, fixez la structure des topics et le format du payload, car tout ce que vous construirez en aval en dépendra. Ce tutoriel utilise un topic par appareil et un payload JSON compact :

connectmq/demo/<run-id>/<line>/<device>/telemetry

{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}

Le <run-id> est une chaîne aléatoire générée au démarrage. Le broker public est lisible par tous : un préfixe unique sépare donc votre flux des expériences des autres et permet de s'abonner facilement à vos seules données. La structure reste dans les six niveaux de topic, bien en dessous de la limite de huit du broker public.

Le simulateur comporte trois parties :

1. Un modèle de capteur qui conserve l'état propre à chaque appareil (valeur de base, dérive, phase du cycle journalier, indicateur de panne) et renvoie une nouvelle mesure à la demande.
2. Un ordonnanceur qui décide quel appareil publie ensuite. Les appareils sont répartis uniformément sur l'intervalle au lieu de tous se déclencher au même instant.
3. Un limiteur de débit qui maintient le client de publication sous la limite de messages par client du broker.

Respecter les limites de débit du broker

Le serveur MQTT public gratuit autorise 10 messages par seconde par client, 10 abonnements par client et 64 Ko par message. Des centaines d'appareils publiant chaque seconde dépasseraient cette limite en un instant ; le simulateur doit donc d'abord faire un peu d'arithmétique. Le débit global correspond simplement au nombre d'appareils divisé par l'intervalle de publication.

Le code se plafonne à 8 messages par seconde pour garder une marge pour les nouvelles tentatives et pour les outils d'abonnement que vous exécutez en parallèle. Si vous avez besoin d'un débit supérieur, deux options honnêtes s'offrent à vous : répartir les appareils sur plusieurs connexions clientes, chacune avec son propre ID et son propre budget, ou passer à un broker privé dont les limites sont fixées par votre plan et non par le service public partagé.

Le simulateur Python complet

Installez l'unique dépendance :

pip install "paho-mqtt>=2.0"

Enregistrez le code suivant sous simulator.py. Il utilise l'API de callbacks de paho-mqtt 2.x et, pour le reste, uniquement la bibliothèque standard.

import argparse
import heapq
import json
import math
import random
import time
import uuid

import paho.mqtt.client as mqtt


class VirtualSensor:
    """Un capteur simulé de température et d'humidité avec un état persistant."""

    def __init__(self, index: int, fault_rate: float):
        self.device_id = f"sensor-{index:04d}"
        self.line = f"line{index % 4 + 1}"
        self.base_temp = random.uniform(19.0, 24.0)
        self.humidity = random.uniform(35.0, 55.0)
        self.phase = random.uniform(0, 2 * math.pi)
        self.drift = 0.0
        self.faulty = random.random() < fault_rate

    def read(self, now: float) -> dict:
        # Cycle journalier lent de +/- 2,5 degrés, décalé par appareil
        daily = 2.5 * math.sin(2 * math.pi * (now % 86400) / 86400 + self.phase)
        # Marche aléatoire bornée pour imiter la dérive d'étalonnage
        self.drift = max(-3.0, min(3.0, self.drift + random.gauss(0, 0.05)))
        # Bruit de mesure gaussien
        temp = self.base_temp + daily + self.drift + random.gauss(0, 0.15)
        self.humidity = max(10.0, min(95.0, self.humidity + random.gauss(0, 0.3)))
        # Les appareils défaillants émettent parfois des pics physiquement impossibles
        if self.faulty and random.random() < 0.05:
            temp += random.choice([-40.0, 60.0])
        return {
            "device": self.device_id,
            "ts": int(now * 1000),
            "temperature": round(temp, 2),
            "humidity": round(self.humidity, 1),
        }


def build_client(args, run_id: str) -> mqtt.Client:
    client_id = args.client_id or f"sim-{run_id}"
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    if args.username:
        client.username_pw_set(args.username, args.password)
    if args.tls:
        client.tls_set()  # magasin de CA système ; le broker utilise un certificat Let's Encrypt

    def on_connect(client, userdata, flags, reason_code, properties):
        print(f"connected: {reason_code}")

    def on_disconnect(client, userdata, flags, reason_code, properties):
        print(f"disconnected: {reason_code}")

    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    return client


def main():
    parser = argparse.ArgumentParser(description="MQTT device simulator")
    parser.add_argument("--host", default="broker.connectmq.com")
    parser.add_argument("--port", type=int, default=1883)
    parser.add_argument("--tls", action="store_true")
    parser.add_argument("--username")
    parser.add_argument("--password")
    parser.add_argument("--client-id")
    parser.add_argument("--prefix", help="topic prefix, default connectmq/demo/<run-id>")
    parser.add_argument("--devices", type=int, default=200)
    parser.add_argument("--interval", type=float, default=30.0)
    parser.add_argument("--max-rate", type=float, default=8.0)
    parser.add_argument("--fault-rate", type=float, default=0.02)
    parser.add_argument("--dropout", type=float, default=0.01)
    args = parser.parse_args()

    run_id = uuid.uuid4().hex[:8]
    prefix = args.prefix or f"connectmq/demo/{run_id}"
    expected = args.devices / args.interval
    if expected > args.max_rate:
        raise SystemExit(f"{expected:.1f} msg/s exceeds --max-rate {args.max_rate}")
    print(f"publishing to {prefix}/# at about {expected:.1f} msg/s")

    sensors = [VirtualSensor(i, args.fault_rate) for i in range(args.devices)]
    client = build_client(args, run_id)
    client.connect_async(args.host, args.port, keepalive=30)
    client.loop_start()

    start = time.time()
    # Répartir uniformément les premières mesures sur un intervalle
    schedule = [(start + i * args.interval / args.devices, i) for i in range(args.devices)]
    heapq.heapify(schedule)
    min_gap = 1.0 / args.max_rate
    last_send = 0.0

    try:
        while True:
            due, i = heapq.heappop(schedule)
            now = time.time()
            wait = max(due - now, last_send + min_gap - now)
            if wait > 0:
                time.sleep(wait)
            sensor = sensors[i]
            now = time.time()
            if random.random() >= args.dropout and client.is_connected():
                topic = f"{prefix}/{sensor.line}/{sensor.device_id}/telemetry"
                client.publish(topic, json.dumps(sensor.read(now)), qos=0)
                last_send = time.time()
            heapq.heappush(schedule, (due + args.interval, i))
    except KeyboardInterrupt:
        print("stopping")
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()

Lancez-le avec les valeurs par défaut :

python simulator.py --devices 200 --interval 30

Le script affiche le préfixe choisi, par exemple connectmq/demo/3f9a1c2e/#. Ouvrez un second terminal et observez le flux :

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-3f9a1c2e -t "connectmq/demo/3f9a1c2e/#" -v

Comment fonctionne le réalisme

Chaque mesure est la somme de quatre composantes, et chacune sert à éprouver une partie précise de votre pipeline.

• Cycle journalier. Une sinusoïde de période 24 heures avec une phase aléatoire par appareil. Vos graphiques prennent une forme visible, et les requêtes d'agrégation comme les moyennes horaires renvoient une variation significative au lieu d'une ligne plate.
• Dérive bornée. Une marche aléatoire limitée à plus ou moins trois degrés. Elle imite des capteurs qui perdent lentement leur étalonnage, exactement ce que la logique de détection de dérive a besoin de voir.
• Bruit de mesure. Un bruit gaussien de faible écart-type. Le lissage, les moyennes glissantes et les filtres à bande morte n'ont de sens que si l'entrée est bruitée.
• Pannes et coupures. Environ 2 % des appareils sont marqués comme défaillants et émettent occasionnellement des pics de plus 60 ou moins 40 degrés. Par ailleurs, 1 % des mesures planifiées sont simplement ignorées. Les pics testent la validation et les seuils d'alerte ; les trous testent la détection de données périmées et le rendu des données manquantes dans vos graphiques.

Comme l'état réside dans chaque objet VirtualSensor, les valeurs d'un appareil sont continues dans le temps. C'est important lorsque vous testez tout ce qui compare la valeur actuelle à la précédente, comme les alertes sur le taux de variation.

Pièges courants et comment les éviter

Publication en rafales. Si chaque appareil publie à la seconde zéro de chaque intervalle, le broker voit une rafale de centaines de messages suivie d'un silence. Les flottes réelles sont rarement synchronisées ainsi, et une rafale atteindra les limites par client. L'ordonnanceur échelonne la première mesure de chaque appareil sur l'intervalle, ce qui lisse le débit.

Confusion d'horloge. Le payload contient son propre horodatage en millisecondes. Les outils en aval comme Telegraf ou une passerelle Python peuvent l'utiliser à la place de l'heure d'arrivée, ce qui garde des graphiques corrects même si les messages sont retardés. Assurez-vous que tous les consommateurs s'accordent sur l'unité : des millisecondes ici, pas des secondes.

Réutilisation des client IDs. Si vous lancez deux copies du simulateur avec le même --client-id, le broker déconnecte l'ancienne session chaque fois que la nouvelle se connecte, et les deux vont se battre. Laissez le script générer un ID aléatoire sur le broker public, ou utilisez l'ID délivré avec vos identifiants sur un broker privé.

Envoi de données sensibles vers un broker public. Tout ce qui transite sur le broker public peut être lu par quiconque devine le topic ou utilise un wildcard. Des données simulées y ont leur place ; de vraies données de clients ou d'usine, non.

Publication pendant une déconnexion. La boucle vérifie client.is_connected() et ignore les mesures tant que la connexion est coupée, exactement comme le ferait un vrai capteur sans mise en mémoire tampon. Si vous voulez tester ce comportement de mise en tampon, lisez le guide de reconnexion paho MQTT et ajoutez une file hors ligne.

Pointer le simulateur vers votre propre broker

Le broker public est idéal pour démarrer rapidement, mais pour du prototypage de charge soutenu, il vous faut un environnement isolé. Avec un broker MQTT managé ConnectMQ, vous obtenez vos propres identifiants et un point de terminaison TLS :

python simulator.py \
  --host broker.connectmq.com --port 8883 --tls \
  --username YOUR_USERNAME --password YOUR_PASSWORD --client-id YOUR_CLIENT_ID \
  --prefix connectmq/factory1/production \
  --devices 500 --interval 60 --max-rate 9

Le port 8883 utilise TLS avec un certificat Let's Encrypt, donc tls_set() avec le magasin de confiance du système suffit. Sur un broker MQTT managé, les client IDs sont liés aux identifiants qui vous sont délivrés : passez l'ID fourni par le portail plutôt que d'en inventer un. Le client intégré au navigateur et l'explorateur de topics en direct du portail permettent de vérifier rapidement que les appareils simulés apparaissent sous les topics attendus. Le plan Serverless (Free) prend en charge jusqu'à 25 connexions en QoS 0 ou 1, ce qui suffit pour répartir une grande flotte simulée sur plusieurs processus si chacun utilise ses propres identifiants.

Où envoyer ensuite les données simulées

Un simulateur ne devient intéressant que lorsque quelque chose consomme ses données. Prochaines étapes naturelles :

1. Parser les payloads en Python comme indiqué dans manipuler du JSON MQTT en Python et calculer des statistiques avec pandas.
2. Persister le flux avec Telegraf et InfluxDB grâce à la stack TIG avec Docker Compose, puis construire des panneaux Grafana dessus.
3. Comparer ce script avec des outils de benchmark dédiés dans notre panorama des outils de test de charge MQTT lorsque vous devez mesurer le débit du broker plutôt que générer des données réalistes.

Pour une vue d'ensemble des options clientes Python et autres, le guide des bibliothèques clientes MQTT compare les principales bibliothèques par langage, et la documentation paho-mqtt décrit chaque option du client utilisée ci-dessus.

Démarrez avec ConnectMQ

Commencez par le serveur MQTT public gratuit broker.connectmq.com pour voir votre flotte simulée en quelques minutes. Quand vous êtes prêt pour des identifiants privés, TLS et vos propres quotas, ConnectMQ exploite un broker MQTT managé dans le cloud. Créez un broker Serverless gratuit sur app.connectmq.com sans carte bancaire, copiez le nom d'utilisateur, le mot de passe et le client ID générés, puis pointez le simulateur dessus. Si vous comparez encore les options d'hébergement, notre guide du broker MQTT cloud explique les critères à examiner.

Questions fréquentes

Combien d'appareils un simulateur d'appareils MQTT peut-il publier depuis un seul client ?

Cela dépend de l'intervalle et de la limite par client du broker. Sur le broker public, gardez le rapport appareils/intervalle sous 10 messages par seconde, par exemple 200 appareils toutes les 30 secondes ou 500 appareils toutes les 60 secondes.

Le simulateur doit-il utiliser la QoS 0 ou la QoS 1 ?

La QoS 0 convient à la plupart des simulations de télémétrie et reflète ce que font de nombreux capteurs réels. Utilisez la QoS 1 lorsque vous voulez spécifiquement tester la gestion des accusés de réception ou la détection de doublons en aval.

Puis-je simuler des appareils qui se déconnectent ?

Oui. Augmentez --dropout pour obtenir des trous aléatoires, ou ajoutez un message Last Will par client afin que les subscribers reçoivent un statut hors ligne quand un processus de simulation est tué.

Est-il sûr d'exécuter le simulateur sur le broker public ?

Oui, tant que les données sont synthétiques. Le broker public est lisible par tous : n'y publiez jamais de vraies données de clients, d'usine ou personnelles.