Un tableau de bord MQTT Streamlit vous permet de transformer les messages de capteurs en direct en graphiques et en indicateurs sans écrire une ligne de HTML ou de JavaScript. Dans ce tutoriel, vous allez en construire un en pur Python : un subscriber paho-mqtt en arrière-plan connecté à broker.connectmq.com, un buffer thread-safe et une page qui s'actualise toute seule toutes les deux secondes. Vous construirez aussi un petit simulateur pour que le tableau de bord dispose de données dès la première minute.
Pourquoi Streamlit et MQTT ont besoin d'un peu de colle
Streamlit possède un modèle d'exécution inhabituel. Chaque fois qu'un utilisateur interagit avec la page, ou qu'une actualisation est déclenchée, Streamlit réexécute tout votre script de haut en bas. C'est idéal pour les applications de données simples, mais les clients MQTT ont une longue durée de vie : ils maintiennent une connexion TCP, font tourner une boucle réseau et déclenchent des callbacks à chaque arrivée de message.
Si vous créez naïvement un client MQTT en haut du script, chaque réexécution crée une nouvelle connexion. En une minute, vous avez des dizaines de clients, des messages en double et, sur le serveur MQTT public gratuit, des connexions coupées parce que des client ID sont réutilisés. La solution consiste à créer le client une seule fois par processus serveur et à laisser Streamlit lire un buffer partagé à chaque réexécution. Streamlit fournit l'outil adéquat pour cela : st.cache_resource.
L'architecture se présente ainsi :
| Composant | Où il s'exécute | Responsabilité |
|---|---|---|
| Client paho-mqtt | Thread réseau en arrière-plan (loop_start) |
Maintient la connexion, reçoit les messages |
Buffer deque + verrou |
Mémoire partagée, créée une seule fois | Conserve les N mesures les plus récentes |
| Script Streamlit | Réexécuté à chaque actualisation | Copie le buffer, construit un DataFrame, dessine les graphiques |
st.fragment(run_every=...) |
Planificateur de Streamlit | Ne réexécute que la section des graphiques toutes les quelques secondes |
Prérequis
Il vous faut Python 3.9 ou plus récent et trois paquets :
python -m venv .venv
source .venv/bin/activate # Windows : .venv\Scripts\activate
pip install "streamlit>=1.37" "paho-mqtt>=2.0" pandas
Streamlit 1.37 a introduit st.fragment avec le paramètre run_every, que ce tutoriel utilise pour l'actualisation automatique. paho-mqtt 2.x exige la version explicite de l'API de callbacks montrée ci-dessous ; si vous venez d'un code plus ancien, lisez d'abord le guide de l'API de callbacks de paho-mqtt 2.0.
Choisissez un préfixe de topic aléatoire pour que vos données ne se mélangent pas à celles des autres sur le serveur MQTT public gratuit, réservé aux tests. Ce tutoriel utilise connectmq/demo/k7f3q2 ; remplacez k7f3q2 par votre propre chaîne aléatoire dans les deux fichiers.
Étape 1 : un simulateur de capteurs
Avant de construire le tableau de bord, créez quelque chose qui publie des données. Enregistrez ceci sous simulator.py :
import json
import math
import random
import time
import uuid
import paho.mqtt.client as mqtt
BROKER = "broker.connectmq.com"
PREFIX = "connectmq/demo/k7f3q2"
SENSORS = ["line1", "line2", "line3"]
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"sim-{uuid.uuid4().hex[:8]}")
client.connect(BROKER, 1883, keepalive=60)
client.loop_start()
t = 0
try:
while True:
for sensor in SENSORS:
reading = {
"sensor": sensor,
"temperature": round(22 + 3 * math.sin(t / 20) + random.uniform(-0.5, 0.5), 2),
"humidity": round(45 + 5 * math.cos(t / 30) + random.uniform(-1, 1), 2),
"ts": time.time(),
}
client.publish(f"{PREFIX}/sensors/{sensor}", json.dumps(reading), qos=0)
t += 1
time.sleep(1)
except KeyboardInterrupt:
pass
finally:
client.loop_stop()
client.disconnect()
Il publie trois messages par seconde, bien en dessous de la limite du serveur public de 10 messages par seconde et par client. Lancez-le dans son propre terminal avec python simulator.py et laissez-le tourner.
Étape 2 : le tableau de bord MQTT Streamlit
Enregistrez ceci sous dashboard.py :
import json
import threading
import time
import uuid
from collections import deque
import pandas as pd
import paho.mqtt.client as mqtt
import streamlit as st
BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/k7f3q2"
TOPIC = f"{PREFIX}/sensors/+"
MAX_POINTS = 600
@st.cache_resource
def start_mqtt():
"""Create one MQTT client per Streamlit server process."""
state = {
"buffer": deque(maxlen=MAX_POINTS),
"lock": threading.Lock(),
"connected": False,
"received": 0,
}
def on_connect(client, userdata, flags, reason_code, properties):
if reason_code.is_failure:
return
state["connected"] = True
client.subscribe(TOPIC, qos=0)
def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
state["connected"] = False
def on_message(client, userdata, msg):
try:
data = json.loads(msg.payload)
row = {
"time": pd.to_datetime(data.get("ts", time.time()), unit="s"),
"sensor": data.get("sensor", msg.topic.rsplit("/", 1)[-1]),
"temperature": float(data["temperature"]),
"humidity": float(data["humidity"]),
}
except (ValueError, KeyError, TypeError):
return # ignorer les payloads mal formés
with state["lock"]:
state["buffer"].append(row)
state["received"] += 1
client = mqtt.Client(
mqtt.CallbackAPIVersion.VERSION2,
client_id=f"streamlit-{uuid.uuid4().hex[:8]}",
)
# Pour un broker ConnectMQ privé :
# client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
# client.tls_set() et utilisez le port 8883
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.on_message = on_message
client.reconnect_delay_set(min_delay=1, max_delay=30)
client.connect_async(BROKER, PORT, keepalive=60)
client.loop_start()
return state
st.set_page_config(page_title="MQTT Sensor Dashboard", layout="wide")
st.title("Live MQTT Sensor Dashboard")
st.caption(f"Subscribed to `{TOPIC}` on {BROKER}")
state = start_mqtt()
@st.fragment(run_every=2)
def live_view():
with state["lock"]:
rows = list(state["buffer"])
received = state["received"]
status = "Connected" if state["connected"] else "Connecting..."
c1, c2, c3 = st.columns(3)
c1.metric("Broker status", status)
c2.metric("Messages received", received)
c3.metric("Points in buffer", len(rows))
if not rows:
st.info("Waiting for data. Is simulator.py running with the same prefix?")
return
df = pd.DataFrame(rows)
latest = df.groupby("sensor").last()
cols = st.columns(len(latest))
for col, (sensor, r) in zip(cols, latest.iterrows()):
col.metric(f"{sensor} temperature", f"{r['temperature']:.1f} °C")
temp = df.pivot_table(index="time", columns="sensor", values="temperature")
hum = df.pivot_table(index="time", columns="sensor", values="humidity")
left, right = st.columns(2)
left.subheader("Temperature")
left.line_chart(temp)
right.subheader("Humidity")
right.line_chart(hum)
live_view()
Lancez-le avec :
streamlit run dashboard.py
Streamlit ouvre http://localhost:8501. En quelques secondes, le statut affiche Connected, les compteurs commencent à grimper et trois courbes apparaissent sur chaque graphique.
Comment fonctionne le tableau de bord
Un client, de nombreuses réexécutions
st.cache_resource exécute start_mqtt() la première fois et renvoie le même objet à chaque appel ultérieur, d'une réexécution à l'autre et entre les sessions de navigateur sur le même serveur. C'est exactement la durée de vie dont une connexion MQTT a besoin. Comme l'objet en cache est partagé par tous les spectateurs, le tableau de bord n'ouvre qu'une seule connexion au broker, quel que soit le nombre de personnes qui le consultent, ce qui compte sur les offres avec une limite de connexions.
Sécurité des threads
loop_start() de paho exécute la boucle réseau dans un thread en arrière-plan : on_message s'exécute donc sur un thread différent de celui de votre script Streamlit. Ajouter un élément à une deque est atomique en CPython, mais la copier pendant qu'un autre thread y ajoute des éléments peut lever « deque mutated during iteration ». Le verrou rend la copie et l'ajout mutuellement exclusifs, et il n'est tenu que quelques microsecondes : il ne ralentit donc jamais le thread réseau.
Le maxlen de la deque est votre garde-fou mémoire. Avec 600 points, le tableau de bord affiche environ les 200 dernières secondes de trois capteurs, et la mémoire reste stable quelle que soit la durée d'exécution.
Réexécutions partielles avec les fragments
@st.fragment(run_every=2) indique à Streamlit de ne réexécuter que cette fonction toutes les deux secondes au lieu de tout le script. Le titre et la légende restent en place, les graphiques se mettent à jour et la page ne scintille pas. Sur les anciennes versions de Streamlit, il fallait un composant tiers d'actualisation automatique ou une boucle time.sleep() plus st.rerun(), deux solutions plus maladroites.
Les reconnexions
connect_async() combiné à loop_start() permet au tableau de bord de démarrer même si le broker est brièvement injoignable, et paho réessaie en arrière-plan avec les délais définis par reconnect_delay_set. S'abonner dans on_connect est important : après une reconnexion avec une clean session, les abonnements ont disparu, et c'est dans on_connect qu'ils sont rétablis. Le guide de reconnexion paho-mqtt approfondit ce pattern.
Problèmes courants et solutions
- Messages en double. Vous créez le client en dehors de
st.cache_resource, ou vous avez modifiéstart_mqttet Streamlit a créé un nouveau client en cache pendant que l'ancien thread continue de tourner. Redémarrez le processusstreamlit runaprès avoir modifié la configuration MQTT. - Le tableau de bord affiche « Connecting... » indéfiniment. Le port 1883 est peut-être bloqué sur votre réseau. Passez à TLS sur le port 8883 avec
client.tls_set(), ou testez avecmosquitto_sub -h broker.connectmq.com -p 1883 -i test-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -v. - Connecté mais aucune donnée. Le préfixe dans
dashboard.pyetsimulator.pydoit correspondre exactement, casse comprise. - Les graphiques font des sauts. Les horodatages proviennent du publisher. Si les appareils ont des horloges non synchronisées, utilisez plutôt l'heure de réception (
time.time()danson_message). - Page lente avec de nombreux capteurs. Pivoter des milliers de lignes toutes les deux secondes finit par coûter cher. Réduisez
MAX_POINTS, augmentezrun_everyou sous-échantillonnez avecdf.resample()avant de tracer.
Passer en production
Le serveur MQTT public gratuit est idéal pour les expériences, mais tout ce qui se trouve dans son espace public est lisible par n'importe qui. Pour un tableau de bord affichant les données réelles d'une usine ou d'un bâtiment :
- Créez un broker privé sur ConnectMQ et utilisez les identifiants générés avec TLS sur le port 8883. Le guide Python MQTT TLS montre la configuration exacte de
tls_set(). - Conservez les secrets dans
.streamlit/secrets.tomlet lisez-les avecst.secrets["mqtt"]["username"]au lieu de les coder en dur. - Si vous avez besoin d'un historique au-delà du buffer en mémoire, faites tourner un subscriber séparé qui écrit dans une base de données et laissez le tableau de bord l'interroger. Le tutoriel MQTT vers SQLite en Python est un point de départ léger, et le guide MQTT et pandas aide pour l'analyse.
Streamlit est un outil Python, mais il appartient à la même famille que les tableaux de bord dans le navigateur. Si vous préférez une approche purement front-end, le client MQTT JavaScript pour navigateur se connecte directement depuis la page en WebSocket, et la vue d'ensemble des bibliothèques clientes MQTT compare les principales options dans chaque langage. La documentation de Streamlit détaille les fragments, la mise en cache et le déploiement.
Faites tourner vos tableaux de bord Streamlit sur ConnectMQ
ConnectMQ est un broker MQTT managé complet : vous disposez de TLS sur le port 8883, d'identifiants générés et d'un portail doté d'un explorateur de topics en direct et d'un client intégré au navigateur pour vérifier ce que publient vos capteurs avant de les mettre en graphique. Commencez avec broker.connectmq.com, puis passez à votre propre broker MQTT cloud. Inscrivez-vous à l'offre Serverless gratuite, sans carte bancaire.
Questions fréquentes
Streamlit peut-il s'abonner directement à MQTT ?
Streamlit ne prend pas en charge MQTT nativement, mais n'importe quel client MQTT Python convient. Le pattern fiable consiste à créer un client paho-mqtt une seule fois avec st.cache_resource et à faire tourner sa boucle réseau dans un thread en arrière-plan.
À quelle fréquence le tableau de bord doit-il s'actualiser ?
Une à cinq secondes conviennent à la plupart des tableaux de bord de capteurs. Des actualisations plus rapides consomment du CPU sur le serveur Streamlit sans rendre plus utiles des valeurs qui évoluent lentement.
Pourquoi est-ce que je reçois chaque message plusieurs fois ?
Plusieurs clients MQTT sont en cours d'exécution, généralement parce que le client a été créé en dehors de st.cache_resource ou qu'un ancien client en cache a survécu à une modification du code. Redémarrez le processus Streamlit pour qu'un seul client existe.
Plusieurs utilisateurs peuvent-ils consulter le même tableau de bord MQTT Streamlit ?
Oui. Le client en cache est partagé par toutes les sessions du serveur : dix spectateurs utilisent donc toujours une seule connexion au broker et lisent le même buffer.
