Enregistrer les messages MQTT en CSV avec Python

Lorsque vous devez enregistrer les messages MQTT en CSV, vous voulez généralement des réponses rapides : qu'a envoyé le capteur pendant la nuit, à quelle fréquence un appareil remonte-t-il ses données, et les données semblent-elles cohérentes avant de construire quelque chose de plus ambitieux ? Ce tutoriel construit un petit subscriber Python qui écrit chaque message en CSV, fait tourner les fichiers par jour et par taille, et produit des fichiers qui s'ouvrent proprement dans Excel, pandas ou un notebook Jupyter. Vous pouvez l'exécuter sur le serveur MQTT public gratuit en cinq minutes et le pointer plus tard vers un broker privé sans modifier le code.

Pourquoi le CSV reste une bonne première étape

Une base de données de séries temporelles est la bonne destination pour les données à long terme, mais ce n'est pas toujours la bonne première étape. Le CSV possède quelques propriétés qui le rendent idéal pour une analyse préliminaire :

Tous les outils le lisent : tableurs, pandas, R, DuckDB, et même un éditeur de texte.
Il n'y a rien à installer ni à exploiter au-delà de Python.
Les fichiers se copient facilement, s'envoient par e-mail à un collègue ou se joignent à un rapport de bug.
Les écritures en ajout seul sont simples et robustes : un plantage fait perdre au plus la dernière ligne.

Les compromis sont tout aussi clairs. Le CSV n'a ni index, ni types, ni contrôle de concurrence : il devient lent et peu pratique au-delà de quelques centaines de mégaoctets, et convient mal lorsque plusieurs processus doivent écrire. Considérez-le comme un format de capture pour le débogage, la mise en service et les expériences de courte durée. Lorsque le logger devient permanent, passez à une solution comme MQTT vers SQLite en Python ou à un véritable stockage de séries temporelles.

Ce dont vous avez besoin

Python 3.9 ou plus récent
paho-mqtt 2.x (pip install "paho-mqtt>=2.0")
En option : mosquitto_pub pour envoyer des messages de test, et pandas pour l'analyse

Pour les tests, utilisez le serveur MQTT public gratuit à l'adresse broker.connectmq.com. Il accepte les connexions anonymes avec n'importe quel client ID non vide et unique. La zone publique est lisible par tout le monde : choisissez un préfixe de topic aléatoire comme connectmq/demo/csv-4821/ et n'y envoyez jamais de vraies données d'usine ou de clients.

Concevoir la structure du CSV

Avant d'écrire du code, décidez de la forme des lignes. Il existe deux structures courantes :

Structure	Colonnes	Idéale pour	Inconvénient
Longue (une ligne par message)	timestamp, topic, qos, retain, payload, value	Topics variés, formes de payload inconnues, débogage	Nécessite un pivot pour les graphiques
Large (une colonne par champ)	timestamp, device, temperature, humidity...	Un seul schéma de payload connu	Casse lorsqu'un nouveau champ apparaît

Le logger ci-dessous utilise la structure longue, car elle ne perd jamais de données : tout ce qui arrive sur le réseau se retrouve dans le fichier. Il tente aussi d'extraire une colonne numérique value, soit à partir d'un payload purement numérique comme 21.4, soit à partir d'un objet JSON doté d'une clé value, ce qui rend les graphiques rapides triviaux sans s'engager sur un schéma fixe.

Les horodatages sont écrits en UTC au format ISO 8601 avec millisecondes. L'heure locale paraît plus conviviale, mais elle crée des heures en double ou manquantes lors des changements d'heure, et tous les outils d'analyse peuvent convertir l'UTC en heure locale à l'affichage.

Le logger CSV en Python

Enregistrez ceci sous mqtt_csv_logger.py :

import csv
import json
import os
import signal
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path

import paho.mqtt.client as mqtt

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")          # None sur le broker public
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"csv-logger-{uuid.uuid4().hex[:8]}")
TOPICS = os.environ.get("MQTT_TOPICS", "connectmq/demo/csv-4821/#").split(",")
OUT_DIR = Path(os.environ.get("CSV_DIR", "mqtt_logs"))
MAX_BYTES = int(os.environ.get("CSV_MAX_BYTES", str(50 * 1024 * 1024)))
SKIP_RETAINED = os.environ.get("SKIP_RETAINED", "1") == "1"

FIELDS = ["received_at", "topic", "qos", "retain", "value", "payload"]


class RotatingCsvWriter:
    """Écrit les lignes dans mqtt_YYYY-MM-DD_NNN.csv, avec rotation quotidienne et par taille."""

    def __init__(self, directory: Path, max_bytes: int):
        self.directory = directory
        self.max_bytes = max_bytes
        self.directory.mkdir(parents=True, exist_ok=True)
        self._file = None
        self._writer = None
        self._day = None
        self._part = 0

    def _open(self, day: str):
        if self._file:
            self._file.close()
        if day != self._day:
            self._day, self._part = day, 0
        while True:
            path = self.directory / f"mqtt_{day}_{self._part:03d}.csv"
            if not path.exists() or path.stat().st_size < self.max_bytes:
                break
            self._part += 1
        new_file = not path.exists()
        # utf-8-sig ajoute un BOM pour qu'Excel détecte l'UTF-8 ; newline="" est requis par csv
        self._file = open(path, "a", newline="", encoding="utf-8-sig" if new_file else "utf-8")
        self._writer = csv.DictWriter(self._file, fieldnames=FIELDS)
        if new_file:
            self._writer.writeheader()
        print(f"Writing to {path}")

    def write(self, row: dict):
        day = row["received_at"][:10]
        if self._file is None or day != self._day or self._file.tell() >= self.max_bytes:
            if self._file is not None and day == self._day:
                self._part += 1
            self._open(day)
        self._writer.writerow(row)
        self._file.flush()

    def close(self):
        if self._file:
            self._file.close()


def extract_value(text: str):
    """Renvoie un float à partir de '21.4' ou '{"value": 21.4}', sinon une chaîne vide."""
    try:
        return float(text)
    except ValueError:
        pass
    try:
        data = json.loads(text)
        if isinstance(data, dict) and isinstance(data.get("value"), (int, float)):
            return data["value"]
    except json.JSONDecodeError:
        pass
    return ""


writer = RotatingCsvWriter(OUT_DIR, MAX_BYTES)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Connect failed: {reason_code}")
        return
    print(f"Connected as {CLIENT_ID}")
    for topic in TOPICS:
        client.subscribe(topic.strip(), qos=1)


def on_message(client, userdata, msg):
    if SKIP_RETAINED and msg.retain:
        return
    text = msg.payload.decode("utf-8", errors="replace")
    writer.write({
        "received_at": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
        "topic": msg.topic,
        "qos": msg.qos,
        "retain": int(msg.retain),
        "value": extract_value(text),
        "payload": text,
    })


def on_disconnect(client, userdata, flags, reason_code, properties):
    print(f"Disconnected: {reason_code}, paho will reconnect")


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)
    if PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)

    def shutdown(*_):
        client.disconnect()
        writer.close()
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    client.connect(BROKER, PORT, keepalive=60)
    client.loop_forever(retry_first_connection=True)


if __name__ == "__main__":
    main()

Quelques choix de conception méritent une explication :

Un seul thread. loop_forever() exécute les callbacks sur le même thread : le writer n'a donc besoin d'aucun verrou. Si vous passez à loop_start(), ajoutez un threading.Lock autour de write().
Flush à chaque ligne. Cela coûte un peu de débit, mais une coupure de courant fait perdre au plus une ligne. Pour des débits de messages très élevés, faites plutôt un flush toutes les N lignes.
Retained messages ignorés par défaut. À chaque reconnexion, le broker rejoue les retained messages, ce qui créerait sinon des lignes en double à chaque micro-coupure réseau.
Abonnement en QoS 1. Les messages publiés en QoS 1 sont livrés au moins une fois ; une reconnexion peut donc produire un doublon occasionnel. C'est généralement acceptable pour l'analyse ; consultez les niveaux de QoS MQTT expliqués si vous devez en tenir compte.

L'exécuter et envoyer des données de test

Démarrez le logger :

pip install "paho-mqtt>=2.0" pandas
python mqtt_csv_logger.py

Dans un second terminal, publiez quelques relevés :

for i in 1 2 3 4 5; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-csv-4821 \
    -t connectmq/demo/csv-4821/line1/temp -m "{\"value\": 2$i.5, \"unit\": \"C\"}"
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-csv-4821 \
    -t connectmq/demo/csv-4821/line1/status -m "running"
  sleep 1
done

Vous devriez voir un fichier comme mqtt_logs/mqtt_2026-10-16_000.csv avec un en-tête et dix lignes. Les lignes de température ont un nombre dans value ; les lignes d'état la laissent vide mais conservent le texte brut dans payload. Si vous n'avez pas d'appareil physique, l'article sur le simulateur d'appareils MQTT montre comment générer des flux réalistes.

Analyser le CSV avec pandas

Les fichiers en structure longue se transforment en graphiques en quelques lignes :

import glob
import pandas as pd

files = sorted(glob.glob("mqtt_logs/mqtt_*.csv"))
df = pd.concat((pd.read_csv(f, encoding="utf-8-sig") for f in files), ignore_index=True)
df["received_at"] = pd.to_datetime(df["received_at"], utc=True)

numeric = df.dropna(subset=["value"])
wide = numeric.pivot_table(index="received_at", columns="topic", values="value")
print(wide.resample("1min").mean().tail())

# Débit de messages par topic, utile pour repérer les appareils trop bavards ou silencieux
print(df.groupby("topic").size().sort_values(ascending=False))

Le résumé du débit de messages justifie souvent à lui seul le logger : un appareil qui publie dix fois plus que prévu, ou un topic devenu silencieux à 03:00, apparaît immédiatement.

Liste de dépannage

Rien n'est écrit. Vérifiez le filtre de topic. connectmq/demo/csv-4821/# correspond à tout ce qui se trouve sous le préfixe, mais une faute de frappe dans un niveau ne correspond silencieusement à rien. Abonnez-vous avec mosquitto_sub -v et le même filtre pour confirmer le trafic.
Excel affiche tout dans une seule colonne. Certains paramètres régionaux attendent des points-virgules, notamment en France. Utilisez Données, À partir d'un fichier texte/CSV et choisissez la virgule comme délimiteur, ou définissez delimiter=";" dans DictWriter pour ce public.
Caractères illisibles dans Excel. Le BOM écrit par utf-8-sig règle ce problème pour les nouveaux fichiers. Les fichiers créés par d'anciennes versions du script doivent être réimportés en sélectionnant UTF-8.
Lignes vides entre les lignes sous Windows. Vous avez ouvert le fichier sans newline="". Le script ci-dessus le gère déjà.
Le logger se déconnecte sans cesse. Deux processus utilisent le même client ID et s'éjectent mutuellement. Laissez le script générer un ID aléatoire ou définissez un MQTT_CLIENT_ID unique.
Les payloads binaires ressemblent à du bruit. Les payloads Protobuf ou compressés ne sont pas du texte. Stockez msg.payload.hex() au lieu du texte décodé pour ces topics.

Passer à un broker privé

Le serveur MQTT public gratuit est parfait pour essayer le logger, mais les vraies données ont leur place sur un broker privé avec identifiants. Avec ConnectMQ, vous créez un broker dans le portail et recevez un username, un mot de passe et un client ID générés. Exécutez ensuite le même script avec des variables d'environnement et TLS sur le port 8883 :

export MQTT_HOST=YOUR_BROKER_HOST
export MQTT_PORT=8883
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_TOPICS="factory1/#"
python mqtt_csv_logger.py

Comme les client IDs sur ConnectMQ sont liés aux identifiants émis, utilisez le client ID fourni par le portail plutôt qu'un ID aléatoire. Un broker MQTT managé signifie aussi que le logger n'est qu'un client parmi d'autres : vous n'avez pas à exécuter Mosquitto à côté, et l'explorateur de topics en direct du portail vous permet de vérifier ce que le logger devrait recevoir avant de vous fier au fichier.

Quand dépasser le CSV

La journalisation en CSV est une étape de transition. Passez à autre chose dès que l'un de ces points devient vrai : les fichiers dépassent ce qu'Excel ouvre confortablement, plus d'un processus doit écrire, vous avez besoin de requêtes sur plusieurs semaines de données, ou d'autres personnes ont besoin de tableaux de bord en direct. Les étapes suivantes naturelles de cette série sont l'envoi de données MQTT vers Google Sheets pour un reporting léger et partagé, et une vraie base de données pour tout ce qui doit durer. Pour un panorama des clients Python, JavaScript et autres utilisables pour ces intégrations, consultez le guide des bibliothèques clientes MQTT. La documentation de paho-mqtt couvre chaque option client utilisée ici.

Démarrer avec ConnectMQ

Pointez votre logger CSV vers un broker privé sécurisé par TLS en quelques minutes. Inscrivez-vous sur app.connectmq.com, choisissez le plan Serverless gratuit sans carte bancaire, créez un broker et collez les identifiants générés dans les variables d'environnement ci-dessus. Le plan gratuit offre jusqu'à 25 connexions, largement assez pour un logger et quelques appareils, ainsi qu'un client MQTT intégré au navigateur pour envoyer des messages de test.

Questions fréquentes

Puis-je enregistrer les messages MQTT en CSV sans écrire de code ?

Oui, des outils comme Node-RED ou Telegraf peuvent écrire des fichiers, mais un court script Python vous donne un contrôle total sur les colonnes, la rotation et le traitement des retained messages. Le script de cet article fait moins de 150 lignes et n'a qu'une seule dépendance.

Vais-je perdre des messages pendant que le logger est hors ligne ?

Avec une clean session, oui : les messages publiés pendant que le logger est déconnecté ne sont pas conservés pour lui. Une session persistante avec le QoS 1 permet au broker de mettre les messages en file pendant un certain temps, mais pour un historique garanti, utilisez une base de données alimentée par un service permanent.

Quelle taille peuvent atteindre les fichiers CSV ?

Le script effectue une rotation à 50 Mo par défaut et démarre un nouveau fichier chaque jour UTC. Ajustez CSV_MAX_BYTES selon l'outil d'analyse ; les tableurs ralentissent bien avant pandas.

Est-il sûr d'enregistrer des données issues du serveur MQTT public gratuit ?

C'est sûr pour des données de test. Tout ce qui se trouve dans la zone publique de broker.connectmq.com peut être lu par quiconque s'abonne au même topic : utilisez un broker privé avec identifiants pour les vrais relevés.

Enregistrer les messages MQTT en CSV avec Python pour une analyse rapide