MQTT avec CircuitPython et MiniMQTT

Adafruit MiniMQTT (adafruit_minimqtt) est le client MQTT de CircuitPython. Il tourne sur des cartes Wi-Fi comme l'ESP32-S2, l'ESP32-S3 et le Raspberry Pi Pico W, utilise la pile TLS intégrée de la carte et suit un style à callbacks qui vous semblera familier si vous avez déjà utilisé paho-mqtt. Ce tutoriel l'installe, se connecte de façon sécurisée, publie une mesure toutes les 10 secondes, réagit à des commandes et survit aux coupures Wi-Fi.

Les exemples ciblent CircuitPython 9.x avec une version actuelle de MiniMQTT issue du bundle Adafruit. Comme CircuitPython recharge code.py à chaque sauvegarde, vous pouvez itérer rapidement — et vous verrez aussi pourquoi un appareil déployé a besoin d'une vraie boucle de gestion des erreurs.

Ce qu'il vous faut

Installez la bibliothèque avec circup, qui copie les bons fichiers .mpy et leurs dépendances sur le lecteur CIRCUITPY :

pip install circup
circup install adafruit_minimqtt
circup list          # affiche les versions installées et signale celles à mettre à jour

Vous pouvez aussi copier à la main le dossier adafruit_minimqtt du bundle dans lib/ ; circup fait moins d'erreurs, car il choisit le bundle correspondant à votre version de CircuitPython.

1. Les secrets dans settings.toml

CircuitPython lit settings.toml à la racine du lecteur. Les deux clés CIRCUITPY_WIFI_* font en plus rejoindre automatiquement le Wi-Fi à la carte au démarrage.

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"

Lisez-les avec os.getenv(). Garder les identifiants ici permet de partager ou de versionner code.py sans rien divulguer.

2. Un publisher et subscriber complet

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("connecté, rc =", rc)
    client.subscribe(f"{BASE}/cmd", qos=1)

def on_disconnect(client, userdata, rc):
    print("déconnecté, 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)                  # lit les paquets entrants, envoie un PINGREQ quand il le faut
    if time.monotonic() - last >= 10:
        temp = microcontroller.cpu.temperature
        mqtt.publish(f"{BASE}/temperature", f"{temp:.1f}")
        last = time.monotonic()

Le rôle des lignes importantes :

• socket_pool et ssl_context relient MiniMQTT à la pile réseau de la carte. ssl.create_default_context() utilise les certificats racine intégrés au firmware CircuitPython : un broker dont le certificat vient d'une CA publique est vérifié sans fichier supplémentaire.
• loop(timeout=1) doit tourner souvent. Il traite les messages entrants, appelle on_message et envoie les pings keep-alive à temps. Le timeout ne doit pas être inférieur au socket_timeout du client (1 seconde par défaut), sinon MiniMQTT lève une erreur.
• S'abonner dans on_connect garantit que chaque connexion réussie — la première comme chaque reconnexion — restaure l'abonnement. C'est important, car les endpoints de test partagés comme broker.connectmq.com ne conservent pas les sessions d'une connexion à l'autre.
• microcontroller.cpu.uid fournit un client ID stable et unique. Si vous préférez des ID aléatoires pour des tests jetables, le générateur de client ID en crée des valides.
• microcontroller.cpu.temperature est une valeur pratique pour la démo ; sur certaines cartes, elle renvoie None. Remplacez-la par un vrai pilote de capteur (BME280, SHT4x…) pour de vraies mesures.

Routage par topic

Quand une carte s'abonne à plusieurs topics, des callbacks par topic gardent le code lisible :

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

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

Les messages de ce topic vont à on_led ; tout le reste continue d'arriver dans le on_message général.

3. Tester depuis le navigateur

Enregistrez code.py et ouvrez la console série (Mu, Thonny ou screen /dev/tty.usbmodem* 115200). Vous devriez voir l'adresse IP, puis connecté, rc = 0.

Ouvrez maintenant le client MQTT en ligne, connectez-vous au même broker et abonnez-vous à testmqtt/cpy/#. Vous verrez le statut online retenu et une température toutes les 10 secondes. Publiez ping sur testmqtt/cpy/<device>/cmd : un pong revient en moins d'une seconde.

Voir votre carte CircuitPython dans le navigateur

Pour tester le Last Will, débranchez la carte : après environ une période et demie de keep-alive, le broker publie offline sur le topic de statut.

4. Survivre aux déconnexions

En développement, une exception arrête simplement code.py et vous corrigez. Sur un appareil rangé dans un placard, une exception non capturée signifie qu'il reste muet jusqu'à ce que quelqu'un le débranche et le rebranche. Enveloppez la boucle :

def ensure_wifi():
    if not wifi.radio.connected:
        print("reconnexion 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("erreur :", e, "- nouvel essai dans", delay, "s")
        time.sleep(delay)
        delay = min(delay * 2, 60)
        try:
            ensure_wifi()
            mqtt.reconnect()              # on_connect s'exécute de nouveau et se réabonne
            mqtt.publish(f"{BASE}/status", "online", retain=True, qos=1)
        except Exception as e2:
            print("échec de la reconnexion :", e2)
            if delay >= 60:
                microcontroller.reset()   # dernier recours après des échecs répétés

Le principe est le même sur toutes les plateformes : réparer d'abord la couche réseau, puis MQTT, appliquer un backoff exponentiel pour qu'une flotte ne submerge pas un broker qui redémarre, et garder un reset matériel comme ultime filet de sécurité. Bonnes pratiques de reconnexion MQTT détaille le raisonnement.

5. Notes sur TLS et les certificats

• Les certificats de CA publiques fonctionnent directement. CircuitPython embarque un bundle de certificats racine, et ssl.create_default_context() l'utilise.
• Certificat auto-signé ou CA privée ? Chargez-la explicitement : ctx = ssl.create_default_context(); ctx.load_verify_locations(cadata=CA_PEM), où CA_PEM est le texte PEM de votre certificat racine.
• La mémoire est limitée. TLS a besoin d'un buffer conséquent. Sur les cartes sans PSRAM, connectez-vous à MQTT tôt dans code.py et évitez les grosses allocations avant le handshake.
• TCP en clair sur le 1883 fonctionne avec is_ssl=False et port=1883, mais transmet le mot de passe en clair. À réserver à un réseau de test sur l'établi.

Si le handshake échoue, écartez d'abord le côté broker avec un test de connexion TLS depuis votre ordinateur.

Les erreurs que vous verrez

Les codes CONNACK sont les codes de retour standard de MQTT 3.1.1 ; l'outil des reason codes les répertorie tous. Pour un montage de capteurs plus complet — plusieurs capteurs, payloads JSON, un dashboard — consultez la station météo MQTT. Et si vous hésitez entre plateformes, le tutoriel umqtt pour MicroPython montre les mêmes tâches avec un client de plus bas niveau.

MiniMQTT offre aux cartes CircuitPython un client MQTT complet avec TLS en quelques dizaines de lignes. Placez les secrets dans settings.toml, abonnez-vous dans on_connect, appelez loop() en permanence et enveloppez le tout dans une boucle de reconnexion : une carte de bricolage devient un appareil que vous pouvez laisser tourner pendant des mois.

Questions fréquentes