Análise de Dados MQTT em Tempo Real com Python e pandas

Combinar MQTT e pandas é uma forma rápida de analisar dados de IoT ao vivo em Python sem precisar montar um banco de dados primeiro. Neste tutorial você vai assinar um broker com paho-mqtt, armazenar as mensagens em buffer com segurança, transformá-las em DataFrames a cada poucos segundos e calcular estatísticas móveis, reamostragens por minuto e sinalizações simples de anomalias.

Quando MQTT e pandas formam a combinação certa

O pandas é uma biblioteca de processamento em lote. Ele é excelente para operar sobre uma tabela inteira de uma vez e ruim para ser atualizado linha por linha. O MQTT é o oposto: as mensagens chegam uma a uma, em intervalos irregulares, em uma thread de rede que você não controla. O truque é colocar um pequeno buffer entre os dois e deixar o pandas trabalhar com microlotes.

Esse padrão se encaixa bem quando você quer explorar um novo fluxo de dados, prototipar um cálculo antes de levá-lo para o código de produção, fazer um monitoramento leve em um notebook ou gateway de borda, ou gerar resumos periódicos para um relatório. Ele não substitui um banco de dados de séries temporais. Se você precisa de meses de histórico, de vários usuários consultando ao mesmo tempo ou de armazenamento durável, grave o fluxo no InfluxDB ou no TimescaleDB e use o pandas sobre os resultados das consultas.

Arquitetura: armazenar, drenar, analisar

O programa tem duas threads com tarefas bem diferentes.

1. A thread de rede do paho executa o on_message para cada mensagem recebida. Ela precisa ser rápida, então apenas decodifica o JSON, extrai alguns campos e acrescenta um dicionário simples a um collections.deque.
2. A thread principal acorda a cada poucos segundos, drena o deque, converte as novas linhas em um DataFrame, adiciona-as a uma janela deslizante e executa a análise.

deque.append e deque.popleft são thread-safe no CPython, então não é preciso um lock explícito nessa passagem. Construir DataFrames dentro do on_message é o erro mais comum em código de MQTT com pandas: cada construção custa muito mais do que acrescentar um dict, e enquanto o callback roda o paho não consegue processar keep-alives nem outras mensagens.

Prepare uma fonte de dados

Você precisa de um fluxo com vários dispositivos e algum ruído. A opção mais fácil é o simulador do nosso tutorial de simulador de dispositivos MQTT, que publica um JSON como este em connectmq/demo/<run-id>/<line>/<device>/telemetry no servidor MQTT público:

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

Inicie-o, anote o run ID que ele imprime e use esse ID abaixo. Qualquer outro fluxo JSON também serve, desde que você ajuste os nomes dos campos. Se você é novo no parse de payloads, o guia sobre JSON com MQTT em Python aborda decodificação e validação em mais profundidade.

Instale as dependências:

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

O script completo de MQTT com pandas

Salve isto como mqtt_pandas.py e defina RUN_ID com o run ID do simulador.

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"  # substitua pelo run ID do seu simulador
TOPIC = f"connectmq/demo/{RUN_ID}/+/+/telemetry"
WINDOW = pd.Timedelta(minutes=15)  # quanto histórico manter em memória
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  # ignora payloads que não são 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. Média móvel de 5 minutos por dispositivo (janela baseada em tempo)
    rolling = (
        w.groupby("device")
        .rolling("5min", on="ts")["temperature"]
        .mean()
        .groupby(level=0)
        .last()
        .rename("temp_5min_mean")
    )

    # 2. Reamostragem por linha e por minuto
    per_minute = (
        w.set_index("ts")
        .groupby("line")["temperature"]
        .resample("1min")
        .mean()
        .unstack(level=0)
        .round(2)
    )

    # 3. Sinalização robusta de anomalias: distância da mediana do dispositivo em unidades de 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. Dispositivos inativos: sem notícias há mais de dois intervalos
    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()

Execute com python mqtt_pandas.py. Depois dos primeiros dez segundos, você verá uma tabela de médias por linha, os cinco dispositivos mais quentes e eventuais anomalias ou sensores inativos. Pressione Ctrl+C para salvar a janela atual em um arquivo CSV.

Entendendo a análise

Estatísticas móveis em uma janela baseada em tempo

rolling("5min", on="ts") usa uma janela baseada em tempo, e não em um número fixo de linhas. Essa diferença importa para dados de IoT: os dispositivos publicam em taxas diferentes e às vezes pulam leituras, então "as últimas 10 linhas" podem significar dez segundos para um dispositivo e dez minutos para outro. Uma janela de tempo sempre cobre o mesmo período. A janela móvel baseada em tempo exige que a coluna de tempo esteja ordenada dentro de cada grupo, e é por isso que a janela é ordenada por ts primeiro.

Reamostragem com pandas

resample("1min").mean() agrupa leituras irregulares em intervalos regulares de um minuto por linha. O resultado é uma tabela larga, com uma coluna por linha, pronta para virar gráfico ou ser exportada. Use aliases em minúsculas, como "1min" e "10s"; as versões recentes do pandas depreciam o antigo alias de segundos em maiúscula. Troque mean por agg(["mean", "min", "max", "count"]) se quiser um resumo mais rico, e lembre que count é um indicador barato de qualidade dos dados.

Detecção robusta de anomalias

Um z-score simples usa a média e o desvio padrão, e ambos são distorcidos justamente pelos picos que você está tentando encontrar. O script usa a mediana e o desvio absoluto mediano (MAD), escalado por 1,4826 para ficar comparável a um desvio padrão em dados normais. Uma pontuação acima de seis é um sinal forte de que a leitura é implausível. Os dispositivos defeituosos do simulador, que de vez em quando saltam dezenas de graus, aparecem na hora.

Detectando dispositivos inativos

A verificação de "visto pela última vez" encontra dispositivos que ficaram em silêncio por mais de dois intervalos esperados. Em produção, isso costuma valer mais do que qualquer limiar sobre os próprios valores, porque um sensor morto não reporta absolutamente nada.

Dicas de desempenho e memória

Com algumas centenas de mensagens por minuto, este script usa pouquíssima CPU. Se a taxa de mensagens chegar a milhares por segundo, leve o trabalho pesado para um banco de dados ou um processador de streams e mantenha o pandas para análises pontuais.

Usando um broker privado para dados reais

O servidor MQTT público é perfeito para este tutorial, mas os topics dele podem ser lidos por qualquer pessoa e ele serve apenas para testes. Para dados reais de uma planta em Campinas, de um frigorífico ou de um prédio, conecte-se a um broker MQTT gerenciado com credenciais e TLS. Só as linhas de conexão mudam:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # certificado Let's Encrypt, confiável pelo repositório do sistema
client.connect("broker.connectmq.com", 8883, keepalive=30)

Com o ConnectMQ, você obtém o usuário, a senha e o client ID no portal depois de criar um broker. Antes de rodar a análise, o explorador de topics ao vivo do portal é uma forma rápida de confirmar quais topics estão realmente chegando e como são os seus payloads.

Para manter um histórico de longo prazo ao lado da visão ao vivo no pandas, grave o mesmo fluxo em um banco de dados. Nosso guia sobre como armazenar dados MQTT no InfluxDB usa a mesma estrutura de topics, então você pode rodar os dois lado a lado. Para outras opções em Python, a visão geral de bibliotecas de cliente MQTT compara o paho-mqtt com alternativas assíncronas, e o guia do usuário do pandas explica janelas e reamostragem em profundidade.

Comece a usar o ConnectMQ

Você pode rodar tudo deste tutorial contra o servidor MQTT público broker.connectmq.com agora mesmo. Quando a sua análise passar de dados de demonstração para dispositivos reais, o ConnectMQ, um broker MQTT gerenciado, oferece um broker privado com TLS, MQTT 5 e um portal para inspecionar topics ao vivo. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito. Se quiser entender antes os prós e contras de hospedagem, leia nosso guia para escolher um broker MQTT na nuvem.

Perguntas frequentes

Posso acrescentar cada mensagem MQTT diretamente a um DataFrame do pandas?

Pode, mas não deve. Acrescentar linha por linha é lento e bloqueia a thread de rede do paho; armazene dicionários simples em buffer e construa um DataFrame a partir de cada lote.

Como lidar com dispositivos que publicam em taxas diferentes?

Use janelas baseadas em tempo, como rolling("5min", on="ts") e resample("1min"). Elas cobrem um período fixo, independentemente de quantas leituras cada dispositivo enviou.

Qual timestamp devo usar, o do payload ou o de chegada?

Prefira o timestamp do dispositivo no payload quando o relógio do dispositivo for confiável, porque ele continua correto mesmo que as mensagens atrasem. Recorra ao horário de chegada quando o payload não tiver timestamp.

Essa abordagem funciona com clientes asyncio?

Sim. Com um cliente assíncrono, o loop de mensagens roda como uma coroutine, mas o padrão é o mesmo: colete as linhas em um buffer e rode o pandas sobre lotes em um intervalo fixo.