Crie um Simulador de Dispositivos MQTT em Python para Centenas de Sensores Virtuais

Um simulador de dispositivos MQTT permite construir dashboards, pipelines de armazenamento e regras de alerta antes que um único sensor real seja instalado. Neste tutorial você vai escrever um simulador em Python que modela centenas de sensores virtuais com ruído, desvio, falhas e perdas de sinal convincentes, e os publica no servidor MQTT público em broker.connectmq.com sem estourar os limites de taxa.

Por que simular dispositivos antes de ter o hardware

O hardware costuma ser a parte mais lenta de um projeto de IoT. As placas estão encomendadas, o firmware está pela metade e o instalador ainda não foi ao local, enquanto as pessoas que constroem o dashboard e o banco de dados já precisam de dados. Um simulador elimina essa dependência. Ele entrega a cada componente posterior um fluxo realista para trabalhar e permite ensaiar situações difíceis de provocar com equipamentos reais: um sensor que salta para um valor impossível, um dispositivo que fica em silêncio por dez minutos ou uma linha inteira entrando em operação de uma vez.

Há um segundo benefício que as equipes descobrem mais tarde. Quando a frota real já existe, o mesmo simulador vira a sua ferramenta de regressão. Você pode rodá-lo contra um broker de staging depois de cada mudança no código de ingestão e ver na hora se um parser ou um dashboard quebrou.

O que separa um simulador útil de um brinquedo é o realismo. Um loop que publica random.uniform(20, 25) a cada segundo gera dados que não se parecem em nada com os de uma sonda de temperatura real. Sinais reais mudam devagar, seguem ciclos diários, carregam ruído de medição, sofrem desvio ao longo de semanas e, de vez em quando, falham. O simulador abaixo modela cada um desses comportamentos com poucas linhas de código.

Design do simulador de dispositivos MQTT

Antes de escrever código, defina a estrutura de topics e o formato de payload, porque tudo o que você construir depois vai depender deles. Este tutorial usa um topic por dispositivo e um payload JSON compacto:

connectmq/demo/<run-id>/<line>/<device>/telemetry

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

O <run-id> é uma string aleatória gerada na inicialização. O servidor MQTT público pode ser lido por todos, então um prefixo único mantém o seu fluxo separado dos experimentos de outras pessoas e facilita assinar exatamente os seus dados. A estrutura fica em seis níveis de topic, com folga dentro do limite de oito níveis do servidor MQTT público.

O simulador tem três partes:

1. Um modelo de sensor que guarda o estado de cada dispositivo (valor base, desvio, fase do ciclo diário, flag de falha) e devolve uma nova leitura quando solicitado.
2. Um agendador que decide qual dispositivo publica em seguida. Os dispositivos são distribuídos uniformemente ao longo do intervalo, em vez de dispararem todos no mesmo instante.
3. Um limitador de taxa que mantém o cliente publicador abaixo do limite de mensagens por cliente do broker.

Respeite os limites de taxa do broker

O servidor MQTT público permite 10 mensagens por segundo por cliente, 10 assinaturas por cliente e 64 KB por mensagem. Centenas de dispositivos publicando a cada segundo estourariam isso em instantes, então o simulador precisa fazer uma conta antes. A taxa agregada é simplesmente o número de dispositivos dividido pelo intervalo de publicação.

O código se limita a 8 mensagens por segundo para deixar folga para retentativas e para as ferramentas de assinatura que você rodar ao lado. Se precisar de uma taxa maior, há duas opções honestas: distribuir os dispositivos entre várias conexões de cliente, cada uma com o próprio ID e a própria cota, ou migrar para um broker privado, onde os limites são definidos pelo seu plano e não pelo serviço público compartilhado.

O simulador completo em Python

Instale a única dependência:

pip install "paho-mqtt>=2.0"

Salve o código a seguir como simulator.py. Ele usa a API de callbacks do paho-mqtt 2.x e, fora isso, apenas a biblioteca padrão.

import argparse
import heapq
import json
import math
import random
import time
import uuid

import paho.mqtt.client as mqtt


class VirtualSensor:
    """Um sensor simulado de temperatura e umidade com estado persistente."""

    def __init__(self, index: int, fault_rate: float):
        self.device_id = f"sensor-{index:04d}"
        self.line = f"line{index % 4 + 1}"
        self.base_temp = random.uniform(19.0, 24.0)
        self.humidity = random.uniform(35.0, 55.0)
        self.phase = random.uniform(0, 2 * math.pi)
        self.drift = 0.0
        self.faulty = random.random() < fault_rate

    def read(self, now: float) -> dict:
        # Ciclo diário lento de +/- 2,5 graus, defasado por dispositivo
        daily = 2.5 * math.sin(2 * math.pi * (now % 86400) / 86400 + self.phase)
        # Passeio aleatório limitado para imitar o desvio de calibração
        self.drift = max(-3.0, min(3.0, self.drift + random.gauss(0, 0.05)))
        # Ruído de medição gaussiano
        temp = self.base_temp + daily + self.drift + random.gauss(0, 0.15)
        self.humidity = max(10.0, min(95.0, self.humidity + random.gauss(0, 0.3)))
        # Dispositivos com falha emitem de vez em quando picos fisicamente implausíveis
        if self.faulty and random.random() < 0.05:
            temp += random.choice([-40.0, 60.0])
        return {
            "device": self.device_id,
            "ts": int(now * 1000),
            "temperature": round(temp, 2),
            "humidity": round(self.humidity, 1),
        }


def build_client(args, run_id: str) -> mqtt.Client:
    client_id = args.client_id or f"sim-{run_id}"
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    if args.username:
        client.username_pw_set(args.username, args.password)
    if args.tls:
        client.tls_set()  # repositório de CAs do sistema; o broker usa um certificado Let's Encrypt

    def on_connect(client, userdata, flags, reason_code, properties):
        print(f"connected: {reason_code}")

    def on_disconnect(client, userdata, flags, reason_code, properties):
        print(f"disconnected: {reason_code}")

    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    return client


def main():
    parser = argparse.ArgumentParser(description="MQTT device simulator")
    parser.add_argument("--host", default="broker.connectmq.com")
    parser.add_argument("--port", type=int, default=1883)
    parser.add_argument("--tls", action="store_true")
    parser.add_argument("--username")
    parser.add_argument("--password")
    parser.add_argument("--client-id")
    parser.add_argument("--prefix", help="topic prefix, default connectmq/demo/<run-id>")
    parser.add_argument("--devices", type=int, default=200)
    parser.add_argument("--interval", type=float, default=30.0)
    parser.add_argument("--max-rate", type=float, default=8.0)
    parser.add_argument("--fault-rate", type=float, default=0.02)
    parser.add_argument("--dropout", type=float, default=0.01)
    args = parser.parse_args()

    run_id = uuid.uuid4().hex[:8]
    prefix = args.prefix or f"connectmq/demo/{run_id}"
    expected = args.devices / args.interval
    if expected > args.max_rate:
        raise SystemExit(f"{expected:.1f} msg/s exceeds --max-rate {args.max_rate}")
    print(f"publishing to {prefix}/# at about {expected:.1f} msg/s")

    sensors = [VirtualSensor(i, args.fault_rate) for i in range(args.devices)]
    client = build_client(args, run_id)
    client.connect_async(args.host, args.port, keepalive=30)
    client.loop_start()

    start = time.time()
    # Distribui as primeiras leituras uniformemente ao longo de um intervalo
    schedule = [(start + i * args.interval / args.devices, i) for i in range(args.devices)]
    heapq.heapify(schedule)
    min_gap = 1.0 / args.max_rate
    last_send = 0.0

    try:
        while True:
            due, i = heapq.heappop(schedule)
            now = time.time()
            wait = max(due - now, last_send + min_gap - now)
            if wait > 0:
                time.sleep(wait)
            sensor = sensors[i]
            now = time.time()
            if random.random() >= args.dropout and client.is_connected():
                topic = f"{prefix}/{sensor.line}/{sensor.device_id}/telemetry"
                client.publish(topic, json.dumps(sensor.read(now)), qos=0)
                last_send = time.time()
            heapq.heappush(schedule, (due + args.interval, i))
    except KeyboardInterrupt:
        print("stopping")
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()

Execute com os valores padrão:

python simulator.py --devices 200 --interval 30

O script imprime o prefixo escolhido, por exemplo connectmq/demo/3f9a1c2e/#. Abra um segundo terminal e acompanhe o fluxo:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-3f9a1c2e -t "connectmq/demo/3f9a1c2e/#" -v

Como o realismo funciona

Cada leitura é a soma de quatro componentes, e cada um existe para exercitar uma parte específica do seu pipeline.

• Ciclo diário. Uma senoide com período de 24 horas e fase aleatória por dispositivo. Os seus gráficos ganham uma forma visível, e consultas de agregação como médias por hora retornam variações significativas em vez de uma linha reta.
• Desvio limitado. Um passeio aleatório restrito a mais ou menos três graus. Ele imita sensores perdendo calibração aos poucos, exatamente o que a lógica de detecção de desvio precisa ver.
• Ruído de medição. Ruído gaussiano com desvio padrão pequeno. Suavização, médias móveis e filtros de banda morta só fazem sentido quando a entrada tem ruído.
• Falhas e perdas de sinal. Cerca de 2 por cento dos dispositivos são marcados como defeituosos e emitem, de vez em quando, picos de mais 60 ou menos 40 graus. Separadamente, 1 por cento das leituras agendadas é simplesmente pulado. Os picos testam a validação e os limiares de alerta; as lacunas testam a detecção de dados desatualizados e como os seus gráficos exibem dados ausentes.

Como o estado fica em cada objeto VirtualSensor, os valores de um dispositivo são contínuos ao longo do tempo. Isso importa quando você testa qualquer coisa que compare o valor atual com o anterior, como alertas de taxa de variação.

Armadilhas comuns e como evitá-las

Publicação em rajadas. Se todos os dispositivos publicarem no segundo zero de cada intervalo, o broker recebe uma rajada de centenas de mensagens seguida de silêncio. Frotas reais raramente são sincronizadas assim, e uma rajada vai esbarrar nos limites por cliente. O agendador escalona a primeira leitura de cada dispositivo ao longo do intervalo, o que mantém a taxa uniforme.

Confusão de relógios. O payload carrega o próprio timestamp em milissegundos. Ferramentas posteriores, como o Telegraf ou uma ponte em Python, podem usá-lo em vez do horário de chegada, o que mantém os gráficos corretos mesmo se as mensagens atrasarem. Garanta que todos os consumidores concordem com a unidade: milissegundos aqui, não segundos.

Reutilizar client IDs. Se você iniciar duas cópias do simulador com o mesmo --client-id, o broker desconecta a sessão mais antiga toda vez que a mais nova conecta, e as duas vão brigar. Deixe o script gerar um ID aleatório no servidor MQTT público ou use o ID emitido para as suas credenciais em um broker privado.

Enviar dados sensíveis ao servidor MQTT público. Tudo o que está no servidor MQTT público pode ser lido por qualquer pessoa que adivinhe o topic ou use curingas. Dados simulados não têm problema ali; dados reais de clientes ou da planta, sim.

Publicar enquanto desconectado. O loop verifica client.is_connected() e pula as leituras enquanto a conexão está fora, como faria um sensor real sem buffer. Se quiser testar o comportamento de buffer, leia o guia de reconexão do paho MQTT e adicione uma fila offline.

Apontando o simulador para o seu próprio broker

O servidor MQTT público é ideal para começar rápido, mas para prototipar carga contínua você vai querer um ambiente isolado. Com um broker MQTT gerenciado do ConnectMQ, você recebe credenciais próprias e um endpoint TLS:

python simulator.py \
  --host broker.connectmq.com --port 8883 --tls \
  --username YOUR_USERNAME --password YOUR_PASSWORD --client-id YOUR_CLIENT_ID \
  --prefix connectmq/factory1/production \
  --devices 500 --interval 60 --max-rate 9

A porta 8883 usa TLS com um certificado Let's Encrypt, então tls_set() com o repositório de confiança do sistema basta. Em um broker MQTT gerenciado, os client IDs são vinculados às credenciais emitidas para você, então passe o ID do portal em vez de inventar um. O cliente no navegador e o explorador de topics ao vivo do portal são uma forma rápida de confirmar que os dispositivos simulados aparecem nos topics esperados. O plano Serverless (Free) suporta até 25 conexões com QoS 0 ou 1, o suficiente para dividir uma grande frota simulada entre vários processos, desde que cada um use as próprias credenciais emitidas.

Para onde enviar os dados simulados em seguida

Um simulador só fica interessante quando algo consome os dados. Próximos passos naturais:

1. Interprete os payloads em Python como mostrado em trabalhando com JSON e MQTT em Python e calcule estatísticas com pandas.
2. Persista o fluxo com Telegraf e InfluxDB usando a stack TIG com Docker Compose e depois monte painéis no Grafana.
3. Compare este script com utilitários dedicados de benchmark na nossa visão geral de ferramentas de teste de carga MQTT quando precisar medir o throughput do broker em vez de gerar dados realistas.

Para uma visão mais ampla das opções de cliente em Python e outras linguagens, o guia de bibliotecas de cliente MQTT compara as principais bibliotecas por linguagem, e a documentação do paho-mqtt cobre todas as opções de cliente usadas acima.

Comece a usar o ConnectMQ

Comece pelo servidor MQTT público (gratuito) em broker.connectmq.com para ver a sua frota simulada em minutos. Quando estiver pronto para credenciais privadas, TLS e cotas próprias, o ConnectMQ oferece um broker MQTT gerenciado na nuvem. Crie um broker Serverless gratuito em app.connectmq.com sem cartão de crédito, copie o usuário, a senha e o client ID gerados e aponte o simulador para ele. Se ainda estiver comparando opções de hospedagem, nosso guia sobre broker MQTT na nuvem explica o que avaliar.

Perguntas frequentes

Quantos dispositivos um simulador de dispositivos MQTT consegue publicar a partir de um cliente?

Depende do intervalo e do limite por cliente do broker. No servidor MQTT público, mantenha a divisão entre dispositivos e intervalo abaixo de 10 mensagens por segundo, por exemplo 200 dispositivos a cada 30 segundos ou 500 dispositivos a cada 60 segundos.

O simulador deve usar QoS 0 ou QoS 1?

QoS 0 serve para a maioria das simulações de telemetria e reflete o que muitos sensores reais fazem. Use QoS 1 quando quiser testar especificamente o tratamento de confirmações ou a detecção de duplicatas mais adiante.

Posso simular dispositivos ficando offline?

Sim. Aumente --dropout para gerar lacunas aleatórias, ou adicione uma mensagem de Last Will por cliente para que os assinantes recebam um status offline quando um processo do simulador for encerrado à força.

É seguro rodar o simulador no servidor MQTT público?

É seguro, desde que os dados sejam sintéticos. O servidor MQTT público pode ser lido por qualquer pessoa, então nunca publique ali dados reais de clientes, da planta ou pessoais.