Salvar Mensagens MQTT em CSV com Python

Quando você precisa salvar mensagens MQTT em CSV, normalmente quer respostas rápidas: o que o sensor enviou durante a noite, com que frequência um dispositivo reporta e se os dados fazem sentido antes de você construir algo maior. Este tutorial cria um pequeno subscriber em Python que grava cada mensagem em CSV, rotaciona os arquivos por dia e por tamanho e produz arquivos que abrem sem problemas no Excel, no pandas ou em um notebook Jupyter. Você pode executá-lo no servidor MQTT público (gratuito) em cinco minutos e apontá-lo para um broker privado depois, sem mudar o código.

Por que o CSV Ainda É um Bom Primeiro Passo

Um banco de dados de séries temporais é o destino certo para dados de longo prazo, mas nem sempre é o primeiro passo certo. O CSV tem algumas propriedades que o tornam ideal para análises iniciais:

Toda ferramenta lê CSV: planilhas, pandas, R, DuckDB e até um editor de texto.
Não há nada para instalar ou operar além do Python.
Os arquivos são fáceis de copiar, enviar por e-mail a um colega ou anexar a um relatório de bug.
Gravações apenas de inclusão no final do arquivo são simples e robustas, então uma falha perde no máximo a última linha.

As contrapartidas são igualmente claras. O CSV não tem índices, tipos nem controle de concorrência, então fica lento e desajeitado acima de algumas centenas de megabytes e não é adequado quando vários processos precisam gravar. Trate-o como um formato de captura para depuração, comissionamento e experimentos curtos. Quando o logger se tornar permanente, evolua para algo como MQTT para SQLite em Python ou um armazenamento de séries temporais de verdade.

O que Você Precisa

Python 3.9 ou mais recente
paho-mqtt 2.x (pip install "paho-mqtt>=2.0")
Opcional: mosquitto_pub para enviar mensagens de teste e pandas para análise

Para testes, use o servidor MQTT público em broker.connectmq.com. Ele aceita conexões anônimas com qualquer client ID não vazio e único. A área pública pode ser lida por todos, então escolha um prefixo de topic aleatório, como connectmq/demo/csv-4821/, e nunca envie dados reais da fábrica ou de clientes para lá.

Definindo o Layout do CSV

Antes de escrever código, decida o formato das linhas. Há dois layouts comuns:

Layout	Colunas	Ideal para	Desvantagem
Longo (uma linha por mensagem)	timestamp, topic, qos, retain, payload, value	Topics variados, formatos de payload desconhecidos, depuração	Precisa de pivot para gráficos
Largo (uma coluna por campo)	timestamp, device, temperature, humidity...	Um único schema de payload conhecido	Quebra quando aparece um campo novo

O logger abaixo usa o layout longo porque ele nunca perde dados: tudo o que chega pela rede vai parar no arquivo. Ele também tenta extrair uma coluna numérica value, seja de um payload numérico simples como 21.4, seja de um objeto JSON com a chave value, o que torna gráficos rápidos triviais sem se prender a um schema fixo.

Os timestamps são gravados em UTC ISO 8601 com milissegundos. O horário local parece mais amigável, mas cria horas duplicadas ou faltantes nas mudanças de horário de verão, e toda ferramenta de análise consegue converter UTC para o horário local na exibição.

O Logger CSV em Python

Salve como 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 no broker público
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:
    """Grava linhas em mqtt_YYYY-MM-DD_NNN.csv, rotacionando por dia e por tamanho."""

    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 adiciona um BOM para o Excel detectar UTF-8; newline="" é exigido pelo 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):
    """Retorna um float de '21.4' ou '{"value": 21.4}'; caso contrário, string vazia."""
    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()

Algumas decisões de design merecem explicação:

Thread única. loop_forever() executa os callbacks na mesma thread, então o writer não precisa de lock. Se você mudar para loop_start(), adicione um threading.Lock em volta de write().
Flush a cada linha. Custa um pouco de throughput, mas significa que uma queda de energia perde no máximo uma linha. Para taxas de mensagem muito altas, faça flush a cada N linhas.
Retained messages ignoradas por padrão. A cada reconexão o broker reenvia as retained messages, o que, de outra forma, criaria linhas duplicadas toda vez que a rede oscilasse.
Assinatura com QoS 1. Mensagens publicadas com QoS 1 são entregues pelo menos uma vez, então uma reconexão pode gerar uma duplicata ocasional. Isso costuma ser aceitável para análise; veja níveis de QoS do MQTT explicados se precisar pensar nisso com mais cuidado.

Execute e Envie Dados de Teste

Inicie o logger:

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

Em um segundo terminal, publique algumas leituras:

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

Você deve ver um arquivo como mqtt_logs/mqtt_2026-10-16_000.csv com um cabeçalho e dez linhas. As linhas de temperatura têm um número em value; as linhas de status deixam essa coluna vazia, mas mantêm o texto bruto em payload. Se você não tem um dispositivo físico, o artigo sobre o simulador de dispositivos MQTT mostra como gerar fluxos realistas.

Analise o CSV com pandas

Arquivos no formato longo viram gráficos com poucas linhas:

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())

# Taxa de mensagens por topic, útil para identificar dispositivos falantes demais ou silenciosos
print(df.groupby("topic").size().sort_values(ascending=False))

Só o resumo da taxa de mensagens já costuma justificar o logger: um dispositivo publicando dez vezes mais do que o esperado, ou um topic que ficou em silêncio às 03:00, aparece na hora.

Checklist de Solução de Problemas

Nada é gravado. Verifique o filtro de topic. connectmq/demo/csv-4821/# corresponde a tudo abaixo do prefixo, mas um erro de digitação em um nível faz com que nada corresponda, sem nenhum aviso. Assine com mosquitto_sub -v usando o mesmo filtro para confirmar o tráfego.
O Excel mostra tudo em uma única coluna. Algumas configurações regionais (como a brasileira) esperam ponto e vírgula. Use Dados, De Texto/CSV e escolha vírgula como delimitador, ou defina delimiter=";" no DictWriter para esse público.
Caracteres corrompidos no Excel. O BOM gravado por utf-8-sig resolve isso em arquivos novos. Arquivos criados por versões antigas do script precisam ser reimportados com UTF-8 selecionado.
Linhas em branco entre as linhas no Windows. Você abriu o arquivo sem newline="". O script acima já trata isso.
O logger fica desconectando. Dois processos estão usando o mesmo client ID e derrubando um ao outro. Deixe o script gerar um ID aleatório ou defina um MQTT_CLIENT_ID único.
Payloads binários parecem ruído. Payloads em Protobuf ou compactados não são texto. Para esses topics, armazene msg.payload.hex() em vez do texto decodificado.

Migrando para um Broker Privado

O servidor MQTT público é perfeito para experimentar o logger, mas dados reais pertencem a um broker privado com credenciais. Com o ConnectMQ, você cria um broker no portal e recebe um nome de usuário, uma senha e um client ID gerados. Depois, execute o mesmo script com variáveis de ambiente e TLS na porta 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

Como no ConnectMQ os client IDs são vinculados às credenciais emitidas, use o client ID do portal em vez de um aleatório. Um broker MQTT gerenciado também significa que o logger é apenas mais um cliente: você não precisa rodar um Mosquitto ao lado dele, e o explorador de topics ao vivo do portal permite verificar o que o logger deveria estar recebendo antes de confiar no arquivo.

Quando Deixar o CSV para Trás

O registro em CSV é um degrau. Siga em frente quando qualquer um destes pontos se tornar verdade: os arquivos passam do que o Excel abre com conforto, mais de um processo precisa gravar, você precisa de consultas sobre semanas de dados ou outras pessoas precisam de dashboards ao vivo. Os próximos passos naturais desta série são enviar dados MQTT para o Google Sheets, para relatórios leves e compartilhados, e um banco de dados de verdade para qualquer coisa de longa duração. Para uma visão geral dos clientes Python, JavaScript e outros que você pode usar nessas integrações, veja o guia de bibliotecas cliente MQTT. A documentação do paho-mqtt cobre todas as opções de cliente usadas aqui.

Comece a Usar o ConnectMQ

Aponte seu logger CSV para um broker privado e protegido por TLS em poucos minutos. Cadastre-se em app.connectmq.com, escolha o plano Serverless gratuito, sem cartão de crédito, crie um broker e cole as credenciais geradas nas variáveis de ambiente acima. No plano gratuito você tem até 25 conexões, o que é mais do que suficiente para um logger e alguns dispositivos, além de um cliente MQTT no navegador para enviar mensagens de teste.

Perguntas frequentes

Posso salvar mensagens MQTT em CSV sem escrever código?

Sim, ferramentas como Node-RED ou Telegraf conseguem gravar arquivos, mas um script Python curto dá controle total sobre colunas, rotação e tratamento de retained messages. O script deste artigo tem menos de 150 linhas e uma única dependência.

Vou perder mensagens enquanto o logger estiver offline?

Com uma clean session, sim: mensagens publicadas enquanto o logger está desconectado não são armazenadas para ele. Usar uma sessão persistente com QoS 1 permite que o broker enfileire mensagens por um tempo, mas para ter histórico garantido use um banco de dados alimentado por um serviço de execução contínua.

Qual o tamanho máximo dos arquivos CSV?

Por padrão, o script rotaciona em 50 MB e começa um arquivo novo a cada dia UTC. Ajuste CSV_MAX_BYTES de acordo com a ferramenta que você usa para analisar; as planilhas ficam lentas muito antes do pandas.

É seguro registrar dados do servidor MQTT público?

É seguro para dados de teste. Qualquer coisa na área pública de broker.connectmq.com pode ser lida por quem assinar o mesmo topic, então use um broker privado com credenciais para leituras reais.

Salvar Mensagens MQTT em CSV com Python para Análises Rápidas