MQTT mit CircuitPython und MiniMQTT

Adafruit MiniMQTT (adafruit_minimqtt) ist der MQTT-Client für CircuitPython. Er läuft auf WLAN-Boards wie ESP32-S2, ESP32-S3 und Raspberry Pi Pico W, nutzt den eingebauten TLS-Stack des Boards und folgt einem Callback-Stil, der dir vertraut vorkommt, wenn du paho-mqtt kennst. Dieses Tutorial installiert ihn, verbindet sich sicher, publiziert alle 10 Sekunden einen Messwert, reagiert auf Befehle und übersteht WLAN-Abbrüche.

Die Beispiele zielen auf CircuitPython 9.x mit einer aktuellen MiniMQTT-Version aus dem Adafruit-Bundle. Da CircuitPython code.py bei jedem Speichern neu lädt, kannst du schnell iterieren – und du siehst auch, warum ein ausgeliefertes Gerät eine echte Fehlerbehandlungsschleife braucht.

Was du brauchst

Installiere die Bibliothek mit circup; es kopiert die passenden .mpy-Dateien samt Abhängigkeiten auf das Laufwerk CIRCUITPY:

pip install circup
circup install adafruit_minimqtt
circup list          # zeigt installierte Versionen und markiert veraltete

Du kannst den Ordner adafruit_minimqtt aus dem Bundle auch von Hand nach lib/ kopieren; circup ist weniger fehleranfällig, weil es das Bundle passend zu deiner CircuitPython-Version wählt.

1. Secrets in settings.toml

CircuitPython liest settings.toml aus dem Wurzelverzeichnis des Laufwerks. Die beiden Schlüssel CIRCUITPY_WIFI_* sorgen außerdem dafür, dass das Board beim Booten automatisch ins WLAN geht.

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"

Lies die Werte mit os.getenv(). Liegen die Zugangsdaten hier, kannst du code.py teilen oder committen, ohne etwas preiszugeben.

2. Ein vollständiger Publisher und Subscriber

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

def on_disconnect(client, userdata, rc):
    print("getrennt, 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)                  # liest eingehende Pakete, sendet PINGREQ bei Bedarf
    if time.monotonic() - last >= 10:
        temp = microcontroller.cpu.temperature
        mqtt.publish(f"{BASE}/temperature", f"{temp:.1f}")
        last = time.monotonic()

Was die wichtigen Zeilen tun:

• socket_pool und ssl_context verbinden MiniMQTT mit dem Netzwerk-Stack des Boards. ssl.create_default_context() nutzt die in die CircuitPython-Firmware eingebauten Root-Zertifikate, sodass ein Broker mit einem Zertifikat einer öffentlichen CA ohne zusätzliche Dateien verifiziert wird.
• loop(timeout=1) muss häufig laufen. Es verarbeitet eingehende Nachrichten, ruft on_message auf und sendet Keep-Alive-Pings rechtzeitig. Der Timeout darf nicht kleiner sein als das socket_timeout des Clients (standardmäßig 1 Sekunde), sonst wirft MiniMQTT einen Fehler.
• Abonnieren in on_connect sorgt dafür, dass jede erfolgreiche Verbindung – die erste und jeder Reconnect – das Abo wiederherstellt. Das ist wichtig, weil geteilte Test-Endpunkte wie broker.connectmq.com keine Sessions zwischen Verbindungen behalten.
• microcontroller.cpu.uid liefert eine stabile, eindeutige Client-ID. Wenn du für Wegwerf-Tests lieber zufällige IDs möchtest, erzeugt der Client-ID-Generator gültige.
• microcontroller.cpu.temperature ist ein bequemer Demo-Wert; auf manchen Boards liefert es None. Für echte Messwerte setzt du einen richtigen Sensortreiber ein (BME280, SHT4x …).

Routing pro Topic

Abonniert ein Board mehrere Topics, halten Callbacks pro Topic den Code übersichtlich:

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

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

Nachrichten auf diesem Topic landen in on_led; alles andere erreicht weiterhin das allgemeine on_message.

3. Im Browser testen

Speichere code.py und öffne die serielle Konsole (Mu, Thonny oder screen /dev/tty.usbmodem* 115200). Du solltest die IP-Adresse sehen und danach verbunden, rc = 0.

Öffne jetzt den Online-MQTT-Client, verbinde dich mit demselben Broker und abonniere testmqtt/cpy/#. Du siehst den retained online-Status und alle 10 Sekunden eine Temperatur. Publiziere ping an testmqtt/cpy/<device>/cmd, und innerhalb einer Sekunde kommt ein pong zurück.

Dein CircuitPython-Board im Browser beobachten

Um den Last Will zu testen, trenn das Board vom Strom: Nach etwa anderthalb Keep-Alive-Perioden publiziert der Broker offline auf dem Status-Topic.

4. Verbindungsabbrüche überstehen

In der Entwicklung stoppt eine Exception einfach code.py, und du behebst den Fehler. Bei einem Gerät im Schrank bedeutet eine nicht abgefangene Exception, dass es stumm bleibt, bis jemand den Strom aus- und wieder einschaltet. Umschließe die Schleife:

def ensure_wifi():
    if not wifi.radio.connected:
        print("WLAN-Reconnect")
        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("Fehler:", e, "- neuer Versuch in", delay, "s")
        time.sleep(delay)
        delay = min(delay * 2, 60)
        try:
            ensure_wifi()
            mqtt.reconnect()              # on_connect läuft erneut und abonniert neu
            mqtt.publish(f"{BASE}/status", "online", retain=True, qos=1)
        except Exception as e2:
            print("Reconnect fehlgeschlagen:", e2)
            if delay >= 60:
                microcontroller.reset()   # letzter Ausweg nach wiederholten Fehlern

Das Muster ist auf jeder Plattform gleich: Zuerst die Netzwerkschicht reparieren, dann MQTT, exponentiell zurückweichen, damit eine Flotte einen neu startenden Broker nicht überrennt, und einen harten Reset als letztes Sicherheitsnetz behalten. Best Practices für MQTT-Reconnects erklärt die Hintergründe ausführlicher.

5. Hinweise zu TLS und Zertifikaten

• Zertifikate öffentlicher CAs funktionieren einfach. CircuitPython bringt ein Root-Zertifikats-Bundle mit, und ssl.create_default_context() nutzt es.
• Selbstsigniert oder private CA? Lade sie explizit: ctx = ssl.create_default_context(); ctx.load_verify_locations(cadata=CA_PEM), wobei CA_PEM der PEM-Text deines Root-Zertifikats ist.
• Der Speicher ist begrenzt. TLS braucht einen ordentlichen Puffer. Auf Boards ohne PSRAM verbindest du dich früh in code.py mit MQTT und vermeidest große Allokationen vor dem Handshake.
• Unverschlüsseltes TCP auf 1883 funktioniert mit is_ssl=False und port=1883, schickt das Passwort aber im Klartext. Nur im Testnetz am Arbeitsplatz verwenden.

Schlägt der Handshake fehl, schließ zuerst die Broker-Seite mit einem TLS-Verbindungstest von deinem Rechner aus aus.

Fehler, die dir begegnen

Die CONNACK-Codes sind die Standard-Return-Codes von MQTT 3.1.1; das Reason-Code-Tool listet sie alle auf. Für einen Sensoraufbau, der weiter geht – mehrere Sensoren, JSON-Payloads, ein Dashboard –, sieh dir die MQTT-Wetterstation an. Und wenn du noch zwischen Plattformen wählst, zeigt das MicroPython-umqtt-Tutorial dieselben Aufgaben mit einem Client auf niedrigerer Ebene.

MiniMQTT gibt CircuitPython-Boards einen vollständigen MQTT-Client mit TLS in ein paar Dutzend Zeilen. Leg die Secrets in settings.toml ab, abonniere in on_connect, ruf loop() ständig auf und pack alles in eine Reconnect-Schleife – dann wird aus einem Bastel-Board ein Gerät, das du monatelang laufen lassen kannst.

Häufig gestellte Fragen