Dashboard MQTT com Streamlit em Python Puro: Gráficos de Sensores ao Vivo em 100 Linhas

Um dashboard MQTT com Streamlit permite transformar mensagens de sensores ao vivo em gráficos e métricas sem escrever uma linha de HTML ou JavaScript. Neste tutorial você vai construir um em Python puro: um subscriber paho-mqtt em segundo plano conectado a broker.connectmq.com, um buffer thread-safe e uma página que se atualiza sozinha a cada dois segundos. Você também vai construir um pequeno simulador para que o dashboard tenha dados desde o primeiro minuto.

Por que Streamlit e MQTT Precisam de uma Pequena Cola

O Streamlit tem um modelo de execução incomum. Toda vez que um usuário interage com a página, ou uma atualização é disparada, o Streamlit executa seu script inteiro de novo, de cima a baixo. Isso é ótimo para apps de dados simples, mas clientes MQTT são de longa duração: mantêm uma conexão TCP, executam um loop de rede e disparam callbacks sempre que chega uma mensagem.

Se você criar ingenuamente um cliente MQTT no topo do script, cada nova execução cria outra conexão. Em um minuto você tem dezenas de clientes, mensagens duplicadas e, no servidor MQTT público, conexões sendo derrubadas por reutilizar client IDs. A solução é criar o cliente exatamente uma vez por processo do servidor e deixar o Streamlit ler de um buffer compartilhado a cada execução. O Streamlit oferece a ferramenta certa para isso: st.cache_resource.

A arquitetura fica assim:

Pré-requisitos

Você precisa do Python 3.9 ou mais recente e de três pacotes:

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "streamlit>=1.37" "paho-mqtt>=2.0" pandas

O Streamlit 1.37 introduziu o st.fragment com o parâmetro run_every, que este tutorial usa para a atualização automática. O paho-mqtt 2.x exige a versão explícita da API de callbacks mostrada abaixo; se você está vindo de código antigo, leia primeiro o guia da API de callbacks do paho-mqtt 2.0.

Escolha um prefixo de topic aleatório para que seus dados não se misturem com os de outras pessoas no servidor MQTT público (gratuito), que é voltado a testes e protótipos. Este tutorial usa connectmq/demo/k7f3q2; troque k7f3q2 por uma string aleatória sua nos dois arquivos.

Passo 1: Um Simulador de Sensores

Antes de construir o dashboard, crie algo que publique dados. Salve 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()

Ele publica três mensagens por segundo, bem abaixo do limite do servidor MQTT público de 10 mensagens por segundo por cliente. Execute-o em um terminal próprio com python simulator.py e deixe-o rodando.

Passo 2: O Dashboard MQTT com Streamlit

Salve 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():
    """Cria um cliente MQTT por processo do servidor 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  # ignora payloads malformados
        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 um broker privado do ConnectMQ:
    # client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
    # client.tls_set()  e use a porta 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()

Inicie-o com:

streamlit run dashboard.py

O Streamlit abre http://localhost:8501. Em poucos segundos o status mostra Connected, os contadores começam a subir e três linhas aparecem em cada gráfico.

Como o Dashboard Funciona

Um cliente, muitas execuções

st.cache_resource executa start_mqtt() na primeira vez e retorna o mesmo objeto em todas as chamadas seguintes, entre execuções e entre sessões de navegador no mesmo servidor. Esse é exatamente o ciclo de vida de que uma conexão MQTT precisa. Como o objeto em cache é compartilhado por todos os visitantes, o dashboard abre uma única conexão com o broker, não importa quantas pessoas estejam olhando, o que é importante em planos com limite de conexões.

Segurança entre threads

O loop_start() do paho executa o loop de rede em uma thread em segundo plano, então on_message roda em uma thread diferente da do seu script Streamlit. Adicionar itens a um deque é atômico no CPython, mas copiá-lo enquanto outra thread adiciona itens pode gerar "deque mutated during iteration". O lock torna a cópia e a adição mutuamente exclusivas, e ele fica preso só por microssegundos, então nunca deixa a thread de rede lenta.

O maxlen do deque é sua proteção de memória. Com 600 pontos, o dashboard mostra aproximadamente os últimos 200 segundos de três sensores, e a memória fica estável por quanto tempo ele rodar.

Execuções parciais com fragments

@st.fragment(run_every=2) diz ao Streamlit para executar novamente apenas essa função a cada dois segundos, em vez do script inteiro. O título e a legenda ficam onde estão, os gráficos se atualizam e a página não pisca. Em versões mais antigas do Streamlit, você precisaria de um componente de atualização automática de terceiros ou de um loop com time.sleep() e st.rerun(), ambos mais desajeitados.

Reconexões

connect_async() junto com loop_start() significa que o dashboard inicia mesmo se o broker estiver brevemente inacessível, e o paho tenta novamente em segundo plano com os intervalos definidos por reconnect_delay_set. Assinar dentro de on_connect é importante: após uma reconexão com clean session, as assinaturas se perdem, e on_connect é onde elas são restauradas. O guia de reconexão do paho-mqtt aborda esse padrão com mais profundidade.

Problemas Comuns e Soluções

1. Mensagens duplicadas ou em dobro. Você está criando o cliente fora de st.cache_resource, ou editou start_mqtt e o Streamlit criou um novo cliente em cache enquanto a thread antiga continua rodando. Reinicie o processo streamlit run depois de alterar a configuração MQTT.
2. O dashboard fica em "Connecting..." para sempre. A porta 1883 pode estar bloqueada na sua rede. Mude para TLS na 8883 com client.tls_set(), ou teste com mosquitto_sub -h broker.connectmq.com -p 1883 -i test-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -v.
3. Conectado, mas sem dados. O prefixo em dashboard.py e em simulator.py precisa ser exatamente igual, incluindo maiúsculas e minúsculas.
4. Os gráficos ficam pulando. Os timestamps vêm do publisher. Se os dispositivos têm relógios dessincronizados, use o horário de recebimento (time.time() dentro de on_message).
5. Página lenta com muitos sensores. Fazer pivot de milhares de linhas a cada dois segundos pesa. Reduza MAX_POINTS, aumente run_every ou reduza a amostragem com df.resample() antes de desenhar os gráficos.

Levando para Produção

O servidor MQTT público é ideal para experimentos, mas tudo na sua área pública pode ser lido por qualquer pessoa, então ele não serve para dados de produção. Para um dashboard que mostra dados reais de uma fábrica em Manaus ou de um prédio comercial:

• Crie um broker privado no ConnectMQ e use as credenciais geradas com TLS na porta 8883. O guia de MQTT com TLS em Python mostra a configuração exata do tls_set().
• Guarde os segredos em .streamlit/secrets.toml e leia-os com st.secrets["mqtt"]["username"], em vez de deixá-los fixos no código.
• Se você precisa de histórico além do buffer em memória, execute um subscriber separado que grave em um banco de dados e faça o dashboard consultá-lo. O tutorial de MQTT para SQLite em Python é um ponto de partida leve, e o guia de MQTT com pandas ajuda na análise.

O Streamlit é uma ferramenta Python, mas pertence à mesma família dos dashboards no navegador. Se você prefere uma abordagem puramente de front-end, o cliente MQTT com JavaScript no navegador se conecta direto da página via WebSocket, e a visão geral de bibliotecas cliente MQTT compara as principais opções em cada linguagem. A documentação do Streamlit cobre fragments, cache e implantação em detalhes.

Execute Seus Dashboards Streamlit no ConnectMQ

O ConnectMQ é um broker MQTT gerenciado completo: você tem TLS na 8883, credenciais geradas e um portal com explorador de topics ao vivo e cliente no navegador para conferir o que seus sensores publicam antes de transformar isso em gráficos. Comece com broker.connectmq.com e depois migre para o seu próprio broker MQTT na nuvem. Cadastre-se no plano Serverless gratuito, sem cartão de crédito.

Perguntas frequentes

O Streamlit consegue assinar MQTT diretamente?

O Streamlit não tem suporte nativo a MQTT, mas qualquer cliente MQTT em Python funciona. O padrão confiável é um cliente paho-mqtt criado uma única vez com st.cache_resource e executando seu loop de rede em uma thread em segundo plano.

Com que frequência o dashboard deve se atualizar?

De um a cinco segundos serve para a maioria dos dashboards de sensores. Atualizações mais rápidas custam CPU no servidor Streamlit sem tornar mais úteis valores que mudam devagar.

Por que recebo cada mensagem várias vezes?

Há vários clientes MQTT rodando, normalmente porque o cliente foi criado fora de st.cache_resource ou porque um cliente antigo em cache sobreviveu a uma mudança de código. Reinicie o processo do Streamlit para que exista apenas um cliente.

Muitos usuários podem ver o mesmo dashboard MQTT com Streamlit?

Sim. O cliente em cache é compartilhado por todas as sessões no servidor, então dez visitantes continuam usando uma única conexão com o broker e lendo do mesmo buffer.