ESP32 ve Pico W'de MicroPython ile MQTT (umqtt)

MicroPython, ESP32, ESP8266 veya Raspberry Pi Pico W üzerinde birkaç kilobayt RAM ile çalışan umqtt adında minik bir MQTT istemcisiyle gelir. İki çeşidi vardır: minimal, senkron bir istemci olan umqtt.simple ve bunun üzerine otomatik yeniden bağlanma ekleyen umqtt.robust. Bu eğitimde ikisini de kuruyor, Wi-Fi üzerinden bağlanıyor, publish ve subscribe yapıyor, bağlantıyı canlı tutuyor, kopmalardan kurtuluyor ve TLS'te nelere dikkat etmeniz gerektiğini açıklıyoruz.

Kod, MicroPython 1.22 veya daha yenisini ve micropython-lib'deki güncel umqtt.simple sürümünü hedefler. ESP32 kartlarında ve Pico W'de değişiklik yapmadan çalışır; sensör okuyorsanız yalnızca pin numaraları farklıdır.

Bir bakışta umqtt

İkisi de senkrondur: wait_msg() bir mesaj gelene kadar bekler, check_msg() hemen döner. Arka planda çalışan bir thread yoktur; her şeyden ana döngünüz sorumludur.

1. umqtt'yi kurun

Bazı firmware derlemeleri umqtt'yi zaten içerir (ESP8266 port'u genellikle içerir). ESP32 ve Pico W'de bilgisayarınızdan mpremote ile micropython-lib üzerinden kurun:

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

Ya da kart Wi-Fi'ye bağlandıktan sonra REPL'den:

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

from umqtt.simple import MQTTClient ile kontrol edin: ImportError yoksa hazırsınız.

2. Wi-Fi'ye bağlanın

network modülü ESP32 ve Pico W'de aynıdır. Diğer dosyaların da kullanabilmesi için bunu wifi.py içine koyun:

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

Her iki çip de yalnızca 2,4 GHz ağları destekler. Durum sonsuza kadar "bağlanıyor"da kalıyorsa MQTT'den şüphelenmeden önce bandı ve şifreyi kontrol edin.

3. umqtt.simple ile publish ve subscribe

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("alındı", topic, msg)
    if msg == b"led/on":
        machine.Pin("LED", machine.Pin.OUT).on()   # Pico W; ESP32'de kendi pininizi kullanın

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)          # subscribe()'dan önce ayarlanmalı
client.connect()
client.publish(BASE + b"/status", b"online", retain=True, qos=1)
client.subscribe(BASE + b"/cmd", qos=1)
print("bağlandı:", CLIENT_ID)

last_pub = last_ping = time.ticks_ms()
while True:
    client.check_msg()                   # bloklamaz: bekleyen veri varsa on_message çalışır
    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)

Önemli noktalar:

• Topic'ler ve payload'lar bytes'tır. String'ler çoğu zaman çalışır ama bytes sürprizleri önler ve on_message her zaman bytes alır.
• subscribe()'dan önce set_callback(): aksi halde umqtt bir assertion hatası fırlatır.
• check_msg() ve wait_msg(): wait_msg() bir şey gelene kadar bloklar; salt komut dinleyen bir cihaz için uygundur ama sensör döngünüzü durdurur. Publish de yapıyorsanız check_msg() kullanın.
• Keep-alive sizin işiniz. keepalive=60 ile broker, yaklaşık 90 saniyelik sessizliğin ardından istemciyi düşürür. Yukarıdaki döngü 30 saniyede bir ping atar. Kesin zamanlama kuralları için MQTT keep alive yazısına bakın.
• machine.unique_id()'den türetilen benzersiz client ID: aynı dosyayı çalıştıran iki kart birbirini atmaz.

Test edin

Online MQTT istemcisini açın, broker.connectmq.com adresine bağlanın ve testmqtt/upy/# topic'ine abone olun. Retained status mesajını ve her 10 saniyede bir uptime değerini görmelisiniz. testmqtt/upy/<client-id>/cmd topic'ine led/on publish edin; REPL alındı ... yazdırır.

Kartınızın publish ettiklerini tarayıcıda izleyin

4. Doğru şekilde yeniden bağlanmak

Wi-Fi düşer, router'lar yeniden başlar, broker'lar yeniden başlar. umqtt.simple'da her ağ hatası, socket'e dokunan çağrıdan bir OSError olarak döner; genellikle [Errno 104] ECONNRESET, [Errno 113] ECONNABORTED veya [Errno 110] ETIMEDOUT. Döngüyü sarmalayın, bağlantıyı yeniden kurun ve tekrar abone olun:

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)  # yeni bir clean session'da abonelikler kaybolur
    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)                # yalnızca OSError fırlatarak çıkar
    except OSError as e:
        print("bağlantı koptu:", e, "yeniden deneme", delay, "sn sonra")
        try:
            client.disconnect()
        except Exception:
            pass
        time.sleep(delay)
        delay = min(delay * 2, 60)      # üst sınırlı üstel backoff

Çok sayıda cihazınız olduğunda üstel backoff önemlidir: broker yeniden başladığında her saniye aynı anda tekrar deneyen bir filo, yeniden bağlanma fırtınası yaratır. Bu kalıp MQTT yeniden bağlanma en iyi uygulamaları yazısında anlatılıyor.

Asla takılı kalmaması gereken cihazlar için bir donanım watchdog'u ekleyin: wdt = machine.WDT(timeout=30000) ve her döngü turunda bir kez wdt.feed() çağırın. Kod bir socket çağrısında takılırsa kart kendini yeniden başlatır.

Peki ya umqtt.robust?

umqtt.robust.MQTTClient aynı API'ye sahiptir. publish(), wait_msg() veya check_msg() bir OSError ile karşılaştığında reconnect() çağrılır; bu da artan bir gecikmeyle sonsuza kadar connect(False) dener. İki sınırlaması vardır:

1. Wi-Fi'ye hiç dokunmaz. Erişim noktası gittiyse, çalışması mümkün olmayan bir socket'i denemeye devam eder.
2. clean_session=False ile yeniden bağlanır ve tekrar abone olmaz. Bu, aboneliklerinizi yalnızca broker session'ı sakladıysa geri getirir: özel bir broker genellikle saklar, ancak broker.connectmq.com gibi paylaşımlı test uç noktaları kalıcı session tutmaz. Kalıcı session davranışını kendi broker instance'ınızda test edin.

umqtt.robust hızlı bir prototip için yeterlidir. Gözetimsiz çalışan her şey için yukarıdaki açık döngüyü anlamak daha kolaydır.

5. TLS tuzakları

1883 portu kullanıcı adınızı ve şifrenizi düz metin olarak gönderir. Masa testinin ötesindeki her şey için 8883 portuna TLS ile bağlanın. Güncel umqtt.simple bir ssl.SSLContext alır:

import ssl, ntptime
ntptime.settime()                        # sertifika geçerliliği gerçek bir saat gerektirir

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()

Sık karşılaşılan sorunlar:

• Eski umqtt sürümleri bir context yerine ssl=True, ssl_params={...} kullanıyordu. Constructor'da TypeError alırsanız kartta hangi sürümün olduğunu kontrol edin.
• Cihazda CA deposu yok. MicroPython yerleşik kök sertifikalarla gelmez. Broker'ınızın sertifikasını imzalayan kök CA'yı karta kopyalayın (PEM'i openssl x509 -in ca.pem -outform der -out ca.der ile DER'e çevirin, mpremote cp ca.der : ile yükleyin).
• CERT_NONE yalnızca hata ayıklama içindir. Trafiği şifreler ama bir saldırganınki dahil her sunucuyu kabul eder.
• Bellek. Bir TLS el sıkışması onlarca kilobayt boş heap ister. ESP32 ve Pico W bunu kaldırır; ESP8266'da çoğu zaman yer yetmez. connect() öncesinde gc.collect() çağırın ve uygulamanız heap'i doldurmadan erkenden bağlanın.
• Saat. ntptime.settime() olmadan kart 2000 yılında olduğunu sanır ve geçerli bir sertifikayı "henüz geçerli değil" diyerek reddedebilir.

TLS bağlantısını test etme yazısı, cihazda hata ayıklamadan önce broker'ın sertifika zincirini dizüstü bilgisayarınızdan nasıl kontrol edeceğinizi gösteriyor.

Sık görülen hatalar

MQTTException sonrasındaki sayılar MQTT 3.1.1 CONNACK dönüş kodlarıdır; reason code aracı her birini açıklar. Daha ayrıntılı bir inceleme için MicroPython MQTT hataları yazısına bakın.

umqtt bir oturuşta okunabilecek kadar küçüktür ve gücü de buradan gelir: kendi keep-alive ping'lerinizi, backoff'lu bir yeniden bağlanma döngüsünü, tekrar abone olmayı ve gerçek bir CA ile TLS'i eklediğinizde pil dostu sensörler için güvenilir bir istemciye dönüşür. Prototipi paylaşımlı uç noktada geliştirin, ardından yalnızca host'u ve kimlik bilgilerini değiştirerek aynı kodu özel bir broker'a taşıyın.

Sıkça sorulan sorular