9.19. MQTT v Pythonu¶

Přibalený modul mqtt na každé síťové OpenMV cam obaluje drátový protokol MQTT do jediné třídy, mqtt.MQTTClient. Tato třída otevírá TCP socket, provádí handshake CONNECT, balí a rozbaluje pakety na úrovni bajtů, obstarává keepalive PINGREQ a předává příchozí zprávy PUBLISH do callbacku. Aplikační kód volá connect(), publish(), subscribe() a wait_msg() / check_msg().

9.19.1. Publisher v patnácti řádcích¶

Nejmenším užitečným programem je jediná publikace. Připojit se, publikovat jednu zprávu, odpojit se:

from mqtt import MQTTClient

client = MQTTClient(
    client_id='yard-cam',
    server='test.mosquitto.org',
    port=1883,
)
client.connect()
client.publish(b'yard-cam/motion', b'detected at 14:02', qos=0)
client.disconnect()

test.mosquitto.org je veřejný testovací broker provozovaný projektem Eclipse Mosquitto. Přijímá nešifrovaná TCP připojení na portu 1883 bez přihlašovacích údajů. Nepoužívejte jej pro nic vážného; neposkytuje žádné záruky soukromí a jmenný prostor témat je sdílený se všemi ostatními testery na internetu.

client_id musí být jedinečné pro každé připojení k brokeru – broker jej používá ke sledování relací. Témata a užitečná data zpráv jsou bajty; předejte str, pokud je to pohodlnější, a klient jej zakóduje jako UTF-8.

9.19.2. Připojení přes TLS¶

Pro cokoli za hranicí rychlých experimentů je MQTT přes TLS jen jeden argument navíc. Slovník ssl_params se předává funkci ssl.wrap_socket(), takže cokoli, co funguje tam, funguje i zde:

import ssl

client = MQTTClient(
    client_id='yard-cam',
    server='broker.example.com',
    port=8883,                          # TLS-MQTT default port
    ssl_params={'server_hostname': 'broker.example.com'},
    user='yard-cam',
    password=load_token(),
)

Port 8883 je port pro TLS-MQTT rezervovaný organizací IANA. server_hostname zapíná SNI, aby brokeři za sdílenou IP adresou mohli směrovat na správný certifikát – jde o stejný mechanismus, jaký používá HTTPS. user / password se mapují na pole uživatelského jména a hesla v paketu CONNECT; broker rozhoduje, zda tyto přihlašovací údaje udělují právo publikovat nebo odebírat konkrétní témata.

9.19.3. Odebírání a příjem¶

Aby klient přijímal zprávy, poskytne callback a zavolá subscribe(). Callback obdrží dva argumenty typu bytes, téma a užitečná data:

def on_message(topic, msg):
    print('received on', topic.decode(), ':', msg.decode())

client = MQTTClient(
    client_id='dashboard',
    server='test.mosquitto.org',
    port=1883,
    callback=on_message,
)
client.connect()
client.subscribe(b'yard-cam/motion', qos=0)
while True:
    client.wait_msg()

wait_msg() blokuje, dokud nepřijde jeden MQTT paket, zpracuje jej, zavolá callback, pokud šlo o PUBLISH na odebíraném tématu, a vrátí se. Callbacky odběrů se spouštějí zevnitř tohoto volání – žádné vlákno na pozadí neexistuje.

Pro interaktivní smyčku kamery, která musí dál vykonávat jinou práci, je check_msg() stejnou logikou v neblokující podobě. Používá select.select() s časovým limitem 50 ms a vrací se okamžitě, pokud nic nečeká:

while True:
    client.check_msg()
    run_frame()                  # capture + processing
    check_motion_threshold()

9.19.4. Čisté znovupřipojení¶

Každý dlouho běžící MQTT klient musí zvládat přerušená připojení. Výpadky Wi-Fi, restarty brokeru, časové limity NAT nebo prostě překročení keepalive bez provozu – to vše socket ukončí. Přibalený klient vyvolá OSError (nebo holou výjimku s návratovým kódem brokeru) z volání, které výpadek zaznamenalo, a standardním vzorem je smyčka opakovaných pokusů:

import time

def keep_publishing(client, topic, get_message):
    while True:
        try:
            client.connect()
            while True:
                client.publish(topic, get_message())
                time.sleep(5)
        except OSError:
            print('connection lost, reconnecting in 5s')
            time.sleep(5)

Odběry nejsou zachovány napříč znovupřipojeními, pokud klient při připojení nepředal clean_session=False, takže vnitřní connect by měl také znovu vydat všechna volání subscribe(), než se dostane do publikační smyčky.

9.19.5. Háček poslední vůle (last will)¶

Kamera hlásící svůj stav by měla brokeru sdělit, jakou zprávu má jejím jménem odeslat, pokud připojení neočekávaně zanikne. Nastavte poslední vůli před connect()

client = MQTTClient(
    client_id='yard-cam',
    server='broker.example.com',
    port=8883,
    ssl_params={'server_hostname': 'broker.example.com'},
)
client.set_last_will(
    b'yard-cam/status',
    b'offline',
    retain=True,
    qos=0,
)
client.connect()
client.publish(b'yard-cam/status', b'online', retain=True)

Nyní každý dashboard odebírající yard-cam/status uvidí online ve chvíli, kdy se kamera připojí, a offline pokaždé, když broker zjistí, že se kamera odpojila. Uchovaná zpráva offline přetrvává na brokeru, takže dashboard, který se připojí o deset minut později, stále vidí správný aktuální stav.

9.19.6. Kdy zvolit MQTT místo HTTP¶

Kapitola o webových serverech popisuje kameru ve roli HTTP serveru a na stránce o nahrávání do cloudu ve roli HTTP klienta odesílajícího JPEGy na pevnou URL. Obojí má své místo. Správný okamžik sáhnout místo toho po MQTT:

Stejná data se musí dostat k několika posluchačům (dashboard, notifikační služba, záznamník), aniž by kamera předem znala jejich seznam.
Posluchači mohou přicházet a odcházet, aniž by se kamera restartovala.
Kamera chce odebírat – přijímat příkazy od řídicí jednotky – což HTTP klient nedokáže bez dlouhého dotazování (long polling) nebo serveru, který odesílá data na callback URL.
Připojení musí levně přečkat dlouhá období nečinnosti.

Správný okamžik zůstat u HTTP: jedna kamera, jeden server, pevný vzor požadavek/odpověď s tělem, které je příliš velké pro jediné MQTT téma (JPEG snímky přes MQTT fungují, ale jsou k brokeru neslušné; HTTP POST je přirozenější volba).

Křížový odkaz: stránka o nahrávání do cloudu v kapitole o webových serverech ukazuje HTTP verzi „kamera → cloudový archiv“. MQTT verze téhož problému ponechává kameru oddělenou od URL archivu a umožňuje druhému konzumentovi (řekněme aplikaci pro upozornění na telefonu) napojit se na tentýž proud dat.