Analyse de données MQTT en temps réel avec Python et pandas

Combiner MQTT et pandas offre un moyen rapide d'analyser des données IoT en direct en Python sans devoir d'abord mettre en place une base de données. Dans ce tutoriel, vous allez vous abonner à un broker avec paho-mqtt, mettre les messages en tampon en toute sécurité, les transformer en DataFrames toutes les quelques secondes et calculer des statistiques glissantes, des rééchantillonnages par minute et de simples indicateurs d'anomalie.

Quand MQTT et pandas forment la bonne combinaison

pandas est une bibliothèque de traitement par lots. Elle excelle à opérer sur une table entière de lignes et se montre médiocre lorsqu'il s'agit de la mettre à jour ligne par ligne. MQTT, c'est l'inverse : les messages arrivent un par un, à intervalles irréguliers, sur un thread réseau que vous ne contrôlez pas. L'astuce consiste à placer un petit tampon entre les deux et à laisser pandas travailler sur des micro-lots.

Ce modèle convient lorsque vous voulez explorer un nouveau flux de données, prototyper un calcul avant de l'intégrer au code de production, exécuter une surveillance légère sur un ordinateur portable ou une passerelle edge, ou produire des synthèses périodiques pour un rapport. Il ne remplace pas une base de données de séries temporelles. Si vous avez besoin de mois d'historique, de plusieurs utilisateurs interrogeant les données simultanément ou d'un stockage durable, écrivez le flux dans InfluxDB ou TimescaleDB et utilisez pandas sur les résultats des requêtes.

Architecture : tamponner, vider, analyser

Le programme comporte deux threads aux rôles très différents.

1. Le thread réseau de paho exécute on_message pour chaque message entrant. Il doit rester rapide : il se contente donc de décoder le JSON, d'extraire quelques champs et d'ajouter un simple dictionnaire à une collections.deque.
2. Le thread principal se réveille toutes les quelques secondes, vide la deque, convertit les nouvelles lignes en DataFrame, les ajoute à une fenêtre glissante et lance l'analyse.

deque.append et deque.popleft sont thread-safe en CPython : aucun verrou explicite n'est nécessaire pour ce passage de relais. Construire des DataFrames dans on_message est l'erreur la plus courante dans le code MQTT pandas : chaque construction coûte bien plus cher que l'ajout d'un dict, et tant que le callback s'exécute, paho ne peut traiter ni les keep-alives ni les autres messages.

Préparer une source de données

Il vous faut un flux avec plusieurs appareils et un peu de bruit. Le plus simple est d'utiliser le simulateur de notre tutoriel sur le simulateur d'appareils MQTT, qui publie du JSON comme celui-ci vers connectmq/demo/<run-id>/<line>/<device>/telemetry sur le serveur MQTT public gratuit :

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

Démarrez-le, notez l'identifiant d'exécution (run ID) qu'il affiche et utilisez-le ci-dessous. Tout autre flux JSON fonctionne aussi, à condition d'adapter les noms de champs. Si vous débutez avec le parsing des payloads, le guide sur la gestion du JSON MQTT en Python détaille le décodage et la validation.

Installez les dépendances :

pip install "paho-mqtt>=2.0" "pandas>=2.1"

Le script MQTT pandas complet

Enregistrez ceci sous mqtt_pandas.py et définissez RUN_ID avec le run ID du simulateur.

import json
import time
import uuid
from collections import deque

import pandas as pd
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
RUN_ID = "3f9a1c2e"  # remplacer par le run ID de votre simulateur
TOPIC = f"connectmq/demo/{RUN_ID}/+/+/telemetry"
WINDOW = pd.Timedelta(minutes=15)  # historique à conserver en mémoire
ANALYSE_EVERY_S = 10
EXPECTED_INTERVAL_S = 30

inbox: deque = deque(maxlen=100_000)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"connect failed: {reason_code}")
        return
    client.subscribe(TOPIC, qos=0)
    print(f"subscribed to {TOPIC}")


def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return  # ignorer les payloads non JSON
    levels = msg.topic.split("/")
    inbox.append({
        "ts": data.get("ts", int(time.time() * 1000)),
        "line": levels[3],
        "device": levels[4],
        "temperature": data.get("temperature"),
        "humidity": data.get("humidity"),
    })


def drain() -> pd.DataFrame:
    rows = []
    while inbox:
        rows.append(inbox.popleft())
    if not rows:
        return pd.DataFrame()
    df = pd.DataFrame(rows)
    df["ts"] = pd.to_datetime(df["ts"], unit="ms", utc=True)
    df["temperature"] = pd.to_numeric(df["temperature"], errors="coerce")
    df["humidity"] = pd.to_numeric(df["humidity"], errors="coerce")
    return df


def analyse(window: pd.DataFrame) -> None:
    now = pd.Timestamp.now(tz="UTC")
    w = window.sort_values("ts")

    # 1. Moyenne glissante sur 5 minutes par appareil (fenêtre temporelle)
    rolling = (
        w.groupby("device")
        .rolling("5min", on="ts")["temperature"]
        .mean()
        .groupby(level=0)
        .last()
        .rename("temp_5min_mean")
    )

    # 2. Rééchantillonnage par ligne et par minute
    per_minute = (
        w.set_index("ts")
        .groupby("line")["temperature"]
        .resample("1min")
        .mean()
        .unstack(level=0)
        .round(2)
    )

    # 3. Indicateur d'anomalie robuste : écart à la médiane de l'appareil en unités MAD
    med = w.groupby("device")["temperature"].transform("median")
    mad = (w["temperature"] - med).abs().groupby(w["device"]).transform("median")
    w = w.assign(score=(w["temperature"] - med).abs() / (mad * 1.4826 + 1e-9))
    anomalies = w[(w["score"] > 6) & (w["ts"] > now - pd.Timedelta(seconds=ANALYSE_EVERY_S * 2))]

    # 4. Appareils muets : pas vus depuis plus de deux intervalles
    last_seen = w.groupby("device")["ts"].max()
    stale = last_seen[last_seen < now - pd.Timedelta(seconds=EXPECTED_INTERVAL_S * 2)]

    print(f"\n=== {now:%H:%M:%S} | rows {len(w)} | devices {w['device'].nunique()} ===")
    print("Per-line means, last 3 minutes:")
    print(per_minute.tail(3).to_string())
    print("Hottest devices by 5-min mean:")
    print(rolling.sort_values(ascending=False).head(5).round(2).to_string())
    if not anomalies.empty:
        print("Anomalies:")
        print(anomalies[["ts", "device", "temperature"]].to_string(index=False))
    if not stale.empty:
        print(f"Stale devices: {', '.join(stale.index[:10])}")


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"pandas-{uuid.uuid4().hex[:8]}")
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(BROKER, PORT, keepalive=30)
    client.loop_start()

    window = pd.DataFrame()
    try:
        while True:
            time.sleep(ANALYSE_EVERY_S)
            new = drain()
            if not new.empty:
                window = pd.concat([window, new], ignore_index=True)
            if window.empty:
                continue
            cutoff = pd.Timestamp.now(tz="UTC") - WINDOW
            window = window[window["ts"] >= cutoff].reset_index(drop=True)
            analyse(window)
    except KeyboardInterrupt:
        window.to_csv("mqtt_snapshot.csv", index=False)
        print("saved mqtt_snapshot.csv")
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()

Lancez-le avec python mqtt_pandas.py. Au bout des dix premières secondes, vous verrez un tableau des moyennes par ligne, les cinq appareils les plus chauds ainsi que les éventuelles anomalies ou capteurs muets. Appuyez sur Ctrl+C pour enregistrer la fenêtre courante dans un fichier CSV.

L'analyse pas à pas

Statistiques glissantes sur une fenêtre temporelle

rolling("5min", on="ts") utilise une fenêtre temporelle plutôt qu'un nombre fixe de lignes. Cette distinction est importante pour les données IoT : les appareils publient à des rythmes différents et sautent parfois des mesures, si bien que « les 10 dernières lignes » peuvent représenter dix secondes pour un appareil et dix minutes pour un autre. Une fenêtre temporelle couvre toujours la même période. Le rolling temporel exige que la colonne de temps soit triée au sein de chaque groupe, c'est pourquoi la fenêtre est d'abord triée par ts.

Rééchantillonner avec pandas

resample("1min").mean() regroupe des mesures irrégulières en intervalles réguliers d'une minute par ligne. Le résultat est une table large avec une colonne par ligne, prête à être tracée ou exportée. Utilisez des alias en minuscules comme "1min" et "10s" ; les versions récentes de pandas déprécient l'ancien alias en majuscule pour les secondes. Remplacez mean par agg(["mean", "min", "max", "count"]) si vous voulez une synthèse plus riche, et gardez count en tête comme indicateur peu coûteux de qualité des données.

Détection d'anomalies robuste

Un z-score classique utilise la moyenne et l'écart-type, deux grandeurs faussées par les pics mêmes que vous cherchez à détecter. Le script utilise plutôt la médiane et l'écart absolu médian (MAD), multiplié par 1,4826 pour être comparable à un écart-type sur des données normales. Un score supérieur à six indique fortement qu'une mesure est invraisemblable. Les appareils défaillants du simulateur, qui sautent parfois de plusieurs dizaines de degrés, ressortent immédiatement.

Détecter les appareils muets

La vérification de la dernière apparition repère les appareils silencieux depuis plus de deux intervalles attendus. En production, c'est souvent plus précieux que n'importe quel seuil sur les valeurs elles-mêmes, car un capteur mort ne rapporte plus rien du tout.

Conseils de performance et de mémoire

À quelques centaines de messages par minute, ce script consomme très peu de CPU. Si le débit atteint des milliers de messages par seconde, déplacez le gros du travail vers une base de données ou un moteur de traitement de flux et réservez pandas à l'analyse ponctuelle.

Utiliser un broker privé pour des données réelles

Le serveur MQTT public gratuit est parfait pour ce tutoriel, mais ses topics sont lisibles par tous : il est réservé aux tests et aux prototypes. Pour de vraies données d'usine ou de bâtiment, connectez-vous à un broker MQTT managé avec identifiants et TLS. Seules les lignes de connexion changent :

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # certificat Let's Encrypt, reconnu par le magasin système
client.connect("broker.connectmq.com", 8883, keepalive=30)

Avec ConnectMQ, vous obtenez le nom d'utilisateur, le mot de passe et le client ID depuis le portail après avoir créé un broker. Avant de lancer votre analyse, l'explorateur de topics en direct du portail permet de vérifier rapidement quels topics arrivent réellement et à quoi ressemblent leurs payloads.

Pour conserver un historique long terme à côté de votre vue pandas en direct, écrivez le même flux dans une base de données. Notre guide pour stocker des données MQTT dans InfluxDB utilise la même structure de topics, ce qui permet de faire tourner les deux côte à côte. Pour d'autres options en Python, le panorama des bibliothèques clientes MQTT compare paho-mqtt aux alternatives asynchrones, et le guide utilisateur de pandas explique en détail le fenêtrage et le rééchantillonnage.

Démarrez avec ConnectMQ

Vous pouvez exécuter tout ce tutoriel dès maintenant sur le serveur MQTT public gratuit broker.connectmq.com. Lorsque votre analyse passe des données de démonstration aux appareils réels, ConnectMQ vous fournit un broker privé avec TLS, MQTT 5 et un portail pour inspecter les topics en direct. Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire. Si vous voulez d'abord comprendre les compromis d'hébergement, lisez notre guide pour choisir un broker MQTT dans le cloud.

Questions fréquentes

Puis-je ajouter chaque message MQTT directement à un DataFrame pandas ?

C'est possible, mais déconseillé. L'ajout ligne par ligne est lent et bloque le thread réseau de paho : mettez plutôt de simples dictionnaires en tampon et construisez un DataFrame à partir de chaque lot.

Comment gérer des appareils qui publient à des rythmes différents ?

Utilisez des fenêtres temporelles comme rolling("5min", on="ts") et resample("1min"). Elles couvrent une période fixe, quel que soit le nombre de mesures envoyées par chaque appareil.

Quel horodatage utiliser, celui du payload ou l'heure d'arrivée ?

Préférez l'horodatage de l'appareil dans le payload quand son horloge est fiable, car il reste correct même si les messages sont retardés. Rabattez-vous sur l'heure d'arrivée lorsque le payload n'en contient pas.

Cette approche fonctionne-t-elle avec des clients asyncio ?

Oui. Avec un client asynchrone, la boucle de messages s'exécute comme une coroutine, mais le principe reste le même : collecter les lignes dans un tampon et lancer pandas sur des lots à intervalle fixe.