MQTT avec MicroPython sur ESP32 et Pico W (umqtt)

MicroPython embarque un minuscule client MQTT appelé umqtt qui tourne sur un ESP32, un ESP8266 ou un Raspberry Pi Pico W avec quelques kilo-octets de RAM. Il existe en deux variantes : umqtt.simple, un client synchrone minimal, et umqtt.robust, qui ajoute la reconnexion automatique par-dessus. Ce tutoriel installe les deux, se connecte en Wi-Fi, publie et s'abonne, maintient la connexion, se remet des coupures et explique les points de vigilance avec TLS.

Le code vise MicroPython 1.22 ou plus récent et la version actuelle de umqtt.simple dans micropython-lib. Il tourne sans modification sur les cartes ESP32 et sur le Pico W ; seuls les numéros de broche changent si vous lisez des capteurs.

umqtt en bref

Les deux sont synchrones : wait_msg() bloque jusqu'à l'arrivée d'un message, check_msg() rend la main immédiatement. Il n'y a pas de thread d'arrière-plan : votre boucle principale gère tout.

1. Installer umqtt

Certains firmwares incluent déjà umqtt (le port ESP8266 en général). Sur ESP32 et Pico W, installez-le depuis micropython-lib avec mpremote depuis votre ordinateur :

pip install mpremote
mpremote mip install umqtt.simple
mpremote mip install umqtt.robust

Ou, une fois la carte connectée au Wi-Fi, depuis le REPL :

import mip
mip.install("umqtt.simple")
mip.install("umqtt.robust")

Vérifiez avec from umqtt.simple import MQTTClient : pas d'ImportError, vous êtes prêt.

2. Se connecter au Wi-Fi

Le module network est le même sur ESP32 et Pico W. Placez ce code dans wifi.py pour le réutiliser ailleurs :

import network, time

def connect(ssid, password, timeout=20):
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        wlan.connect(ssid, password)
        start = time.time()
        while not wlan.isconnected():
            if time.time() - start > timeout:
                raise OSError("Wi-Fi timeout, status=%s" % wlan.status())
            time.sleep(0.5)
    print("Wi-Fi OK", wlan.ifconfig()[0])
    return wlan

Les deux puces ne gèrent que les réseaux 2,4 GHz. Si le statut reste indéfiniment sur « connexion en cours », vérifiez la bande et le mot de passe avant de soupçonner MQTT.

3. Publier et s'abonner avec umqtt.simple

import time, machine, ubinascii
from umqtt.simple import MQTTClient
import wifi

wifi.connect("your-ssid", "your-wifi-password")

CLIENT_ID = b"upy-" + ubinascii.hexlify(machine.unique_id())
BASE = b"testmqtt/upy/" + CLIENT_ID
KEEPALIVE = 60

def on_message(topic, msg):
    print("reçu", topic, msg)
    if msg == b"led/on":
        machine.Pin("LED", machine.Pin.OUT).on()   # Pico W ; sur ESP32, utilisez votre broche

client = MQTTClient(CLIENT_ID, "broker.connectmq.com", port=1883,
                    user="username", password="password",
                    keepalive=KEEPALIVE)
client.set_last_will(BASE + b"/status", b"offline", retain=True, qos=1)
client.set_callback(on_message)          # à définir avant subscribe()
client.connect()
client.publish(BASE + b"/status", b"online", retain=True, qos=1)
client.subscribe(BASE + b"/cmd", qos=1)
print("connecté en tant que", CLIENT_ID)

last_pub = last_ping = time.ticks_ms()
while True:
    client.check_msg()                   # non bloquant : appelle on_message si des données attendent
    now = time.ticks_ms()
    if time.ticks_diff(now, last_pub) > 10_000:
        client.publish(BASE + b"/uptime", str(time.ticks_ms() // 1000))
        last_pub = now
    if time.ticks_diff(now, last_ping) > KEEPALIVE * 1000 // 2:
        client.ping()
        last_ping = now
    time.sleep_ms(50)

Points clés :

• Topics et payloads sont des bytes. Les chaînes marchent souvent, mais les bytes évitent les surprises, et on_message reçoit toujours des bytes.
• set_callback() avant subscribe() : sinon umqtt lève une erreur d'assertion.
• check_msg() ou wait_msg() : wait_msg() bloque jusqu'à ce que quelque chose arrive, ce qui convient à un simple récepteur de commandes mais fige votre boucle de capteurs. Utilisez check_msg() si vous publiez aussi.
• Le keep-alive, c'est votre affaire. Avec keepalive=60, le broker coupe le client après environ 90 secondes de silence. La boucle ci-dessus envoie un ping toutes les 30 secondes. Les règles de timing exactes sont dans MQTT keep alive.
• Client ID unique dérivé de machine.unique_id(), pour que deux cartes exécutant le même fichier ne s'éjectent pas mutuellement.

Tester

Ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com et abonnez-vous à testmqtt/upy/#. Vous devriez voir le message status retenu et une valeur uptime toutes les 10 secondes. Publiez led/on sur testmqtt/upy/<votre-client-id>/cmd et le REPL affiche reçu ....

Voir ce que publie votre carte dans le navigateur

4. Se reconnecter correctement

Le Wi-Fi tombe, les routeurs redémarrent, les brokers redémarrent. Avec umqtt.simple, chaque panne réseau remonte sous forme d'OSError — typiquement [Errno 104] ECONNRESET, [Errno 113] ECONNABORTED ou [Errno 110] ETIMEDOUT — depuis l'appel qui a touché le socket. Enveloppez la boucle, reconstruisez la connexion et réabonnez-vous :

def mqtt_connect():
    c = MQTTClient(CLIENT_ID, "broker.connectmq.com", user="username",
                   password="password", keepalive=KEEPALIVE)
    c.set_last_will(BASE + b"/status", b"offline", retain=True, qos=1)
    c.set_callback(on_message)
    c.connect()
    c.publish(BASE + b"/status", b"online", retain=True, qos=1)
    c.subscribe(BASE + b"/cmd", qos=1)  # avec une nouvelle clean session, les abonnements disparaissent
    return c

def run_loop(c):
    last_pub = last_ping = time.ticks_ms()
    while True:
        c.check_msg()
        now = time.ticks_ms()
        if time.ticks_diff(now, last_pub) > 10_000:
            c.publish(BASE + b"/uptime", str(time.ticks_ms() // 1000))
            last_pub = now
        if time.ticks_diff(now, last_ping) > KEEPALIVE * 1000 // 2:
            c.ping()
            last_ping = now
        time.sleep_ms(50)

client = None
delay = 1
while True:
    try:
        wifi.connect("your-ssid", "your-wifi-password")
        client = mqtt_connect()
        delay = 1
        run_loop(client)                # ne se termine qu'en levant OSError
    except OSError as e:
        print("connexion perdue :", e, "nouvel essai dans", delay, "s")
        try:
            client.disconnect()
        except Exception:
            pass
        time.sleep(delay)
        delay = min(delay * 2, 60)      # backoff exponentiel plafonné

Le backoff exponentiel compte dès que vous avez beaucoup d'appareils : si le broker redémarre, une flotte qui réessaie toutes les secondes au même moment provoque une tempête de reconnexions. Le principe est détaillé dans bonnes pratiques de reconnexion MQTT.

Pour les appareils qui ne doivent jamais rester bloqués, ajoutez un watchdog matériel : wdt = machine.WDT(timeout=30000) et appelez wdt.feed() une fois par itération de boucle. Si le code se bloque dans un appel socket, la carte se réinitialise d'elle-même.

Et umqtt.robust ?

umqtt.robust.MQTTClient a la même API. Quand publish(), wait_msg() ou check_msg() rencontre une OSError, il appelle reconnect(), qui retente connect(False) indéfiniment avec un délai croissant. Deux limites :

1. Il ne touche jamais au Wi-Fi. Si le point d'accès a disparu, il s'acharne sur un socket qui ne peut pas fonctionner.
2. Il se reconnecte avec clean_session=False et ne se réabonne pas. Vos abonnements ne reviennent donc que si le broker a conservé la session : un broker privé le fait en général, mais les endpoints de test partagés comme broker.connectmq.com ne conservent pas de sessions persistantes. Testez le comportement des sessions persistantes sur votre propre instance de broker.

umqtt.robust convient à un prototype rapide. Pour tout ce qui tourne sans surveillance, la boucle explicite ci-dessus est plus facile à comprendre.

5. Les pièges de TLS

Le port 1883 transmet votre nom d'utilisateur et votre mot de passe en clair. Dès que vous dépassez le test sur l'établi, connectez-vous sur le 8883 en TLS. La version actuelle de umqtt.simple accepte un ssl.SSLContext :

import ssl, ntptime
ntptime.settime()                        # la validité du certificat exige une horloge réelle

ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.verify_mode = ssl.CERT_REQUIRED
with open("ca.der", "rb") as f:
    ctx.load_verify_locations(cadata=f.read())

client = MQTTClient(CLIENT_ID, "broker.connectmq.com", port=8883,
                    user="username", password="password",
                    keepalive=KEEPALIVE, ssl=ctx)
client.connect()

Ce qui piège souvent :

• Les anciennes versions d'umqtt utilisaient ssl=True, ssl_params={...} au lieu d'un contexte. Si vous obtenez un TypeError sur le constructeur, vérifiez la version présente sur la carte.
• Pas de magasin de CA sur l'appareil. MicroPython n'embarque aucun certificat racine. Copiez sur la carte la CA racine qui a signé le certificat de votre broker (conversion PEM vers DER avec openssl x509 -in ca.pem -outform der -out ca.der, envoi avec mpremote cp ca.der :).
• CERT_NONE, uniquement pour déboguer. Le trafic est chiffré, mais n'importe quel serveur est accepté, y compris celui d'un attaquant.
• Mémoire. Un handshake TLS demande plusieurs dizaines de kilo-octets de heap libre. ESP32 et Pico W s'en sortent ; l'ESP8266 est souvent trop juste. Appelez gc.collect() avant connect() et connectez-vous tôt, avant que votre application ne remplisse le heap.
• Horloge. Sans ntptime.settime(), la carte se croit en l'an 2000 et peut refuser un certificat valide comme « pas encore valide ».

Tester une connexion TLS montre comment vérifier la chaîne de certificats du broker depuis votre portable avant de déboguer sur l'appareil.

Erreurs courantes

Les nombres après MQTTException sont des codes de retour CONNACK de MQTT 3.1.1 ; l'outil des reason codes explique chacun d'eux. Pour aller plus loin, consultez erreurs MQTT sous MicroPython.

umqtt est assez petit pour être lu d'une traite, et c'est sa force : dès que vous ajoutez vos propres pings keep-alive, une boucle de reconnexion avec backoff, le réabonnement et TLS avec une vraie CA, il devient un client fiable pour des capteurs économes en batterie. Prototypez sur l'endpoint partagé, puis passez le même code sur un broker privé en changeant simplement l'hôte et les identifiants.

Questions fréquentes