Un dashboard MQTT con Streamlit te permite convertir mensajes de sensores en vivo en gráficos y métricas sin escribir una sola línea de HTML ni de JavaScript. En este tutorial construirás uno en Python puro: un suscriptor paho-mqtt en segundo plano conectado a broker.connectmq.com, un búfer seguro entre hilos y una página que se refresca sola cada dos segundos. También crearás un pequeño simulador para que el dashboard tenga datos desde el primer minuto.
Por qué Streamlit y MQTT necesitan un poco de pegamento
Streamlit tiene un modelo de ejecución poco habitual. Cada vez que un usuario interactúa con la página, o se dispara un refresco, Streamlit vuelve a ejecutar todo tu script de arriba abajo. Eso es genial para apps de datos sencillas, pero los clientes MQTT son de larga duración: mantienen una conexión TCP, ejecutan un bucle de red y disparan callbacks cada vez que llega un mensaje.
Si creas ingenuamente un cliente MQTT al principio del script, cada nueva ejecución crea otra conexión. En un minuto tienes decenas de clientes, mensajes duplicados y, en el broker público, conexiones cortadas por reutilizar client IDs. La solución es crear el cliente exactamente una vez por proceso del servidor y dejar que Streamlit lea de un búfer compartido en cada ejecución. Streamlit ofrece la herramienta adecuada para ello: st.cache_resource.
La arquitectura es la siguiente:
| Componente | Dónde se ejecuta | Responsabilidad |
|---|---|---|
| Cliente paho-mqtt | Hilo de red en segundo plano (loop_start) |
Mantiene la conexión y recibe mensajes |
Búfer deque + lock |
Memoria compartida, creada una vez | Guarda las N lecturas más recientes |
| Script de Streamlit | Se vuelve a ejecutar en cada refresco | Copia el búfer, construye un DataFrame y dibuja gráficos |
st.fragment(run_every=...) |
Planificador de Streamlit | Vuelve a ejecutar solo la sección de gráficos cada pocos segundos |
Requisitos previos
Necesitas Python 3.9 o posterior y tres paquetes:
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 introdujo st.fragment con el parámetro run_every, que este tutorial usa para el refresco automático. paho-mqtt 2.x exige la versión explícita de la API de callbacks que se muestra abajo; si vienes de código antiguo, lee primero la guía de la API de callbacks de paho-mqtt 2.0.
Elige un prefijo de topic aleatorio para que tus datos no se mezclen con los de nadie en el servidor MQTT público gratuito. Este tutorial usa connectmq/demo/k7f3q2; sustituye k7f3q2 por tu propia cadena aleatoria en ambos archivos.
Paso 1: un simulador de sensores
Antes de construir el dashboard, crea algo que publique datos. Guarda esto como 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()
Publica tres mensajes por segundo, muy por debajo del límite del broker público de 10 mensajes por segundo por cliente. Ejecútalo en su propio terminal con python simulator.py y déjalo funcionando.
Paso 2: el dashboard MQTT con Streamlit
Guarda esto como 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():
"""Crea un cliente MQTT por proceso del servidor de Streamlit."""
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 # ignorar payloads mal formados
with state["lock"]:
state["buffer"].append(row)
state["received"] += 1
client = mqtt.Client(
mqtt.CallbackAPIVersion.VERSION2,
client_id=f"streamlit-{uuid.uuid4().hex[:8]}",
)
# Para un broker privado de ConnectMQ:
# client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
# client.tls_set() y usa el puerto 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()
Arráncalo con:
streamlit run dashboard.py
Streamlit abre http://localhost:8501. En un par de segundos el estado muestra Connected, los contadores empiezan a subir y aparecen tres líneas en cada gráfico.
Cómo funciona el dashboard
Un cliente, muchas ejecuciones
st.cache_resource ejecuta start_mqtt() la primera vez y devuelve el mismo objeto en cada llamada posterior, entre ejecuciones y entre sesiones de navegador del mismo servidor. Esa es exactamente la vida que necesita una conexión MQTT. Como el objeto cacheado lo comparten todos los espectadores, el dashboard abre una sola conexión con el broker sin importar cuánta gente lo esté mirando, algo importante en planes con límite de conexiones.
Seguridad entre hilos
loop_start() de paho ejecuta el bucle de red en un hilo en segundo plano, así que on_message se ejecuta en un hilo distinto al de tu script de Streamlit. Añadir a un deque es atómico en CPython, pero copiarlo mientras otro hilo añade elementos puede lanzar "deque mutated during iteration". El lock hace que la copia y la inserción sean mutuamente excluyentes, y solo se mantiene durante microsegundos, así que nunca ralentiza el hilo de red.
El maxlen del deque es tu protección de memoria. Con 600 puntos, el dashboard muestra aproximadamente los últimos 200 segundos de tres sensores, y la memoria se mantiene estable por mucho tiempo que funcione.
Ejecuciones parciales con fragments
@st.fragment(run_every=2) le dice a Streamlit que vuelva a ejecutar solo esa función cada dos segundos en lugar de todo el script. El título y el subtítulo se quedan como están, los gráficos se actualizan y la página no parpadea. En versiones antiguas de Streamlit necesitarías un componente de refresco automático de terceros o un bucle con time.sleep() y st.rerun(), ambos más torpes.
Reconexiones
connect_async() junto con loop_start() hace que el dashboard arranque aunque el broker no esté disponible durante un momento, y paho reintenta en segundo plano con las esperas definidas en reconnect_delay_set. Suscribirse dentro de on_connect es importante: tras una reconexión con sesión limpia las suscripciones desaparecen, y on_connect es donde se restauran. La guía de reconexión de paho-mqtt cubre este patrón con más detalle.
Problemas habituales y soluciones
- Mensajes duplicados. Estás creando el cliente fuera de
st.cache_resource, o has editadostart_mqtty Streamlit ha creado un nuevo cliente cacheado mientras el hilo antiguo sigue funcionando. Reinicia el procesostreamlit rundespués de cambiar la configuración de MQTT. - El dashboard muestra "Connecting..." para siempre. Puede que el puerto 1883 esté bloqueado en tu red. Cambia a TLS en el 8883 con
client.tls_set(), o prueba conmosquitto_sub -h broker.connectmq.com -p 1883 -i test-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -v. - Conectado pero sin datos. El prefijo de
dashboard.pyy desimulator.pydebe coincidir exactamente, incluidas mayúsculas y minúsculas. - Los gráficos dan saltos. Las marcas de tiempo vienen del publicador. Si los dispositivos tienen relojes sin sincronizar, usa en su lugar la hora de recepción (
time.time()dentro deon_message). - Página lenta con muchos sensores. Pivotar miles de filas cada dos segundos se nota. Reduce
MAX_POINTS, aumentarun_everyo reduce la resolución condf.resample()antes de dibujar.
Llevarlo a producción
El servidor público es ideal para experimentos y prototipos, pero todo lo que hay en su área pública lo puede leer cualquiera, así que no sirve para datos de producción. Para un dashboard que muestre datos reales de una planta o un edificio:
- Crea un broker privado en ConnectMQ y usa las credenciales generadas con TLS en el puerto 8883. La guía de TLS para MQTT en Python muestra la configuración exacta de
tls_set(). - Guarda los secretos en
.streamlit/secrets.tomly léelos const.secrets["mqtt"]["username"]en lugar de incrustarlos en el código. - Si necesitas un histórico más allá del búfer en memoria, ejecuta un suscriptor aparte que escriba en una base de datos y haz que el dashboard la consulte. El tutorial de MQTT a SQLite en Python es un punto de partida ligero, y la guía de MQTT y pandas ayuda con el análisis.
Streamlit es una herramienta de Python, pero pertenece a la misma familia que los dashboards en el navegador. Si prefieres un enfoque puramente frontend, el cliente MQTT JavaScript para el navegador se conecta directamente desde la página por WebSocket, y la visión general de librerías cliente MQTT compara las principales opciones en cada lenguaje. Y si el dashboard debe mostrar una planta de Monterrey o un edificio inteligente en Bogotá como gemelo digital 3D, ese enfoque en el navegador permite combinar Three.js con los mismos datos MQTT en vivo. La documentación de Streamlit cubre en detalle fragments, caché y despliegue.
Ejecuta tus dashboards de Streamlit en ConnectMQ
ConnectMQ es un broker MQTT gestionado: obtienes TLS en el 8883, credenciales generadas y un portal con un explorador de topics en vivo y un cliente en el navegador para comprobar lo que publican tus sensores antes de graficarlo. Empieza con broker.connectmq.com y pasa después a tu propio broker MQTT en la nube. Regístrate en el plan Serverless gratuito, sin necesidad de tarjeta de crédito.
Preguntas frecuentes
¿Puede Streamlit suscribirse a MQTT directamente?
Streamlit no tiene soporte integrado para MQTT, pero funciona cualquier cliente MQTT de Python. El patrón fiable es un cliente paho-mqtt creado una sola vez con st.cache_resource y que ejecuta su bucle de red en un hilo en segundo plano.
¿Cada cuánto debe refrescarse el dashboard?
Entre uno y cinco segundos va bien para la mayoría de los dashboards de sensores. Los refrescos más rápidos consumen CPU en el servidor de Streamlit sin hacer más útiles los valores que cambian despacio.
¿Por qué recibo cada mensaje varias veces?
Hay varios clientes MQTT funcionando, normalmente porque el cliente se creó fuera de st.cache_resource o porque un cliente cacheado antiguo sobrevivió a un cambio de código. Reinicia el proceso de Streamlit para que solo exista un cliente.
¿Pueden muchos usuarios ver el mismo dashboard MQTT de Streamlit?
Sí. El cliente cacheado lo comparten todas las sesiones del servidor, así que diez espectadores siguen usando una única conexión con el broker y leen del mismo búfer.
