MQTT no CircuitPython com MiniMQTT

O Adafruit MiniMQTT (adafruit_minimqtt) é o cliente MQTT do CircuitPython. Ele roda em placas Wi-Fi como ESP32-S2, ESP32-S3 e Raspberry Pi Pico W, usa a pilha TLS embutida da placa e segue um estilo de callbacks que vai parecer familiar se você já usou a paho-mqtt. Este tutorial instala a biblioteca, conecta com segurança, publica uma leitura a cada 10 segundos, reage a comandos e sobrevive a quedas do Wi-Fi.

Os exemplos miram o CircuitPython 9.x com uma versão atual do MiniMQTT do bundle da Adafruit. Como o CircuitPython recarrega o code.py sempre que você salva, dá para iterar rápido, e você também vai ver por que um dispositivo em produção precisa de um loop de tratamento de erros de verdade.

O que você precisa

Instale a biblioteca com o circup, que copia os arquivos .mpy certos e as dependências para o drive CIRCUITPY:

pip install circup
circup install adafruit_minimqtt
circup list          # mostra as versões instaladas e aponta as desatualizadas

Você também pode copiar a pasta adafruit_minimqtt do bundle para lib/ manualmente; o circup erra menos porque escolhe o bundle compatível com a sua versão do CircuitPython.

1. Segredos no settings.toml

O CircuitPython lê o settings.toml da raiz do drive. As duas chaves CIRCUITPY_WIFI_* também fazem a placa entrar no Wi-Fi automaticamente no boot.

CIRCUITPY_WIFI_SSID = "your-ssid"
CIRCUITPY_WIFI_PASSWORD = "your-wifi-password"
MQTT_BROKER = "broker.connectmq.com"
MQTT_PORT = 8883
MQTT_USER = "username"
MQTT_PASS = "password"

Leia os valores com os.getenv(). Manter as credenciais aqui permite compartilhar ou versionar o code.py sem vazar nada.

2. Publicador e assinante completos

import os, time
import microcontroller, ssl, socketpool, wifi
import adafruit_minimqtt.adafruit_minimqtt as MQTT

if not wifi.radio.connected:
    wifi.radio.connect(os.getenv("CIRCUITPY_WIFI_SSID"),
                       os.getenv("CIRCUITPY_WIFI_PASSWORD"))
print("Wi-Fi:", wifi.radio.ipv4_address)

DEVICE = "cpy-" + microcontroller.cpu.uid.hex()
BASE = f"testmqtt/cpy/{DEVICE}"

def on_connect(client, userdata, flags, rc):
    print("conectado, rc =", rc)
    client.subscribe(f"{BASE}/cmd", qos=1)

def on_disconnect(client, userdata, rc):
    print("desconectado, rc =", rc)

def on_message(client, topic, message):
    print(f"{topic}: {message}")
    if message == "ping":
        client.publish(f"{BASE}/pong", str(time.monotonic()))

pool = socketpool.SocketPool(wifi.radio)
mqtt = MQTT.MQTT(
    broker=os.getenv("MQTT_BROKER"),
    port=os.getenv("MQTT_PORT"),
    username=os.getenv("MQTT_USER"),
    password=os.getenv("MQTT_PASS"),
    client_id=DEVICE,
    is_ssl=True,
    keep_alive=60,
    socket_pool=pool,
    ssl_context=ssl.create_default_context(),
)
mqtt.on_connect = on_connect
mqtt.on_disconnect = on_disconnect
mqtt.on_message = on_message
mqtt.will_set(f"{BASE}/status", "offline", retain=True, qos=1)

mqtt.connect()
mqtt.publish(f"{BASE}/status", "online", retain=True, qos=1)

last = 0
while True:
    mqtt.loop(timeout=1)                  # lê pacotes recebidos e envia PINGREQ na hora certa
    if time.monotonic() - last >= 10:
        temp = microcontroller.cpu.temperature
        mqtt.publish(f"{BASE}/temperature", f"{temp:.1f}")
        last = time.monotonic()

O que as linhas importantes fazem:

• socket_pool e ssl_context ligam o MiniMQTT à pilha de rede da placa. ssl.create_default_context() usa os certificados raiz embutidos no firmware do CircuitPython, então um broker com certificado de uma CA pública é verificado sem arquivos extras.
• loop(timeout=1) precisa rodar com frequência. Ele processa as mensagens recebidas, chama on_message e envia os pings de keep-alive no tempo certo. O timeout não pode ser menor que o socket_timeout do cliente (1 segundo por padrão), senão o MiniMQTT lança um erro.
• Assinar dentro de on_connect faz com que toda conexão bem-sucedida, a primeira e cada reconexão, restaure a assinatura. Isso importa porque endpoints de teste compartilhados como broker.connectmq.com não mantêm sessões entre conexões.
• microcontroller.cpu.uid fornece um client ID estável e único. Se preferir IDs aleatórios para testes descartáveis, o gerador de client ID cria IDs válidos.
• microcontroller.cpu.temperature é um valor prático para a demo; em algumas placas ele retorna None. Troque por um driver de sensor real (BME280, SHT4x…) para leituras de verdade.

Roteamento por tópico

Quando uma placa assina vários topics, callbacks por tópico mantêm o código organizado:

def on_led(client, topic, message):
    led.value = message == "on"

mqtt.add_topic_callback(f"{BASE}/led", on_led)
mqtt.subscribe(f"{BASE}/led")

Mensagens nesse topic vão para on_led; todo o resto continua chegando ao on_message geral.

3. Teste pelo navegador

Salve o code.py e abra o console serial (Mu, Thonny ou screen /dev/tty.usbmodem* 115200). Você deve ver o endereço IP e depois conectado, rc = 0.

Agora abra o cliente MQTT online, conecte no mesmo broker e assine testmqtt/cpy/#. Você verá o status online retido e uma temperatura a cada 10 segundos. Publique ping em testmqtt/cpy/<device>/cmd e um pong volta em menos de um segundo.

Veja sua placa CircuitPython no navegador

Para testar o Last Will, desconecte a placa da energia: depois de aproximadamente um período e meio de keep-alive, o broker publica offline no topic de status.

4. Sobrevivendo a desconexões

Em desenvolvimento, uma exceção simplesmente para o code.py e você corrige. Num dispositivo guardado num armário, uma exceção não tratada significa que ele fica mudo até alguém desligar e religar. Envolva o loop:

def ensure_wifi():
    if not wifi.radio.connected:
        print("reconectando Wi-Fi")
        wifi.radio.connect(os.getenv("CIRCUITPY_WIFI_SSID"),
                           os.getenv("CIRCUITPY_WIFI_PASSWORD"))

delay = 1
last = 0
while True:
    try:
        mqtt.loop(timeout=1)
        if time.monotonic() - last >= 10:
            mqtt.publish(f"{BASE}/temperature", f"{microcontroller.cpu.temperature:.1f}")
            last = time.monotonic()
        delay = 1
    except (OSError, RuntimeError, ConnectionError, MQTT.MMQTTException) as e:
        print("erro:", e, "- nova tentativa em", delay, "s")
        time.sleep(delay)
        delay = min(delay * 2, 60)
        try:
            ensure_wifi()
            mqtt.reconnect()              # on_connect roda de novo e refaz a assinatura
            mqtt.publish(f"{BASE}/status", "online", retain=True, qos=1)
        except Exception as e2:
            print("falha na reconexão:", e2)
            if delay >= 60:
                microcontroller.reset()   # último recurso após falhas repetidas

O padrão é o mesmo em qualquer plataforma: resolva primeiro a camada de rede e depois o MQTT, use backoff exponencial para que uma frota não atropele um broker que está reiniciando e mantenha um reset forçado como última rede de segurança. Boas práticas de reconexão MQTT explica o raciocínio com mais profundidade.

5. Notas sobre TLS e certificados

• Certificados de CAs públicas simplesmente funcionam. O CircuitPython traz um pacote de certificados raiz, e ssl.create_default_context() o utiliza.
• Autoassinado ou CA privada? Carregue explicitamente: ctx = ssl.create_default_context(); ctx.load_verify_locations(cadata=CA_PEM), onde CA_PEM é o texto PEM do seu certificado raiz.
• A memória é limitada. O TLS precisa de um buffer considerável. Em placas sem PSRAM, conecte ao MQTT cedo no code.py e evite alocações grandes antes do handshake.
• TCP sem criptografia na 1883 funciona com is_ssl=False e port=1883, mas envia a senha em texto puro. Use só numa rede de bancada.

Se o handshake falhar, descarte primeiro o lado do broker com um teste de conexão TLS a partir do seu computador.

Erros que você vai ver

Os códigos CONNACK são os códigos de retorno padrão do MQTT 3.1.1; a ferramenta de reason codes lista todos. Para um projeto de sensores que vai além (vários sensores, payloads JSON, um dashboard), veja a estação meteorológica MQTT. E se você está escolhendo plataforma, o tutorial de umqtt no MicroPython mostra as mesmas tarefas com um cliente de mais baixo nível.

O MiniMQTT dá às placas CircuitPython um cliente MQTT completo com TLS em poucas dezenas de linhas. Coloque os segredos no settings.toml, assine em on_connect, chame loop() o tempo todo e envolva tudo num loop de reconexão: uma placa de hobby vira um dispositivo que você pode deixar rodando por meses.

Perguntas frequentes