9.19. MQTT Pythonban¶

A minden hálózatra kötött OpenMV kamerán előre telepített mqtt modul egyetlen osztályba, a mqtt.MQTTClient osztályba burkolja az MQTT vezetékprotokollt. Az osztály megnyitja a TCP socketet, végrehajtja a CONNECT kézfogást, becsomagolja és kicsomagolja a bájtszintű csomagokat, kezeli a PINGREQ keepalive jelzéseket, és a beérkező PUBLISH üzeneteket egy visszahíváshoz irányítja. Az alkalmazáskód a connect(), publish(), subscribe() és wait_msg() / check_msg() metódusokat hívja.

9.19.1. Egy közzétevő tizenöt sorban¶

A legkisebb hasznos program egyetlen közzététel. Csatlakozás, egy üzenet közzététele, leválasztás:

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

A test.mosquitto.org az Eclipse Mosquitto projekt által üzemeltetett nyilvános tesztbróker. Sima TCP-kapcsolatokat fogad az 1883-as porton, hitelesítő adatok nélkül. Ne használd komoly célokra; nem ad semmilyen adatvédelmi garanciát, és a topic-névteret az internet összes többi tesztelőjével megosztja.

A client_id-nak egyedinek kell lennie brókerkapcsolatonként – a bróker ennek alapján követi nyomon a munkameneteket. A topicok és az üzenetek hasznos adatai bájtok; adj át str értéket, ha az kényelmesebb, és a kliens UTF-8-ként kódolja azt.

9.19.2. Csatlakozás TLS-en keresztül¶

Bármi máshoz a gyors kísérleteken túl az MQTT TLS-en keresztül mindössze egy plusz argumentumot jelent. Az ssl_params szótár az ssl.wrap_socket() függvénynek kerül továbbításra, így bármi, ami ott működik, itt is működik:

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

A 8883-as port az IANA által fenntartott TLS-MQTT port. A server_hostname bekapcsolja az SNI-t, így a megosztott IP mögötti brókerek a megfelelő tanúsítványhoz tudnak irányítani – ugyanaz a mechanizmus, amelyet a HTTPS használ. A user / password a CONNECT csomag felhasználónév/jelszó mezőire képződik le; a bróker dönti el, hogy ezek a hitelesítő adatok adnak-e közzétételi vagy feliratkozási jogot bizonyos topicokhoz.

9.19.3. Feliratkozás és fogadás¶

Üzenetek fogadásához a kliens megad egy visszahívást, és meghívja a subscribe() metódust. A visszahívás két bájt argumentumot kap, a topicot és a hasznos adatot:

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

A wait_msg() addig blokkol, amíg egy MQTT-csomag meg nem érkezik, elemzi azt, meghívja a visszahívást, ha az egy feliratkozott topicon érkezett PUBLISH volt, majd visszatér. A feliratkozott visszahívások e híváson belülről indulnak el – nincs háttérszál.

Egy interaktív kameraciklushoz, amelynek más munkát is folyamatosan végeznie kell, a check_msg() ugyanezt a logikát nyújtja nem blokkoló formában. Az select.select() függvényt használja 50 ms-os időtúllépéssel, és azonnal visszatér, ha semmi sincs függőben:

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

9.19.4. Tiszta újracsatlakozás¶

Bármely hosszú ideig futó MQTT-kliensnek kezelnie kell a megszakadt kapcsolatokat. A Wi-Fi leválás, a bróker újraindulása, a NAT-időtúllépések vagy egyszerűen a keepalive túllépése forgalom nélkül mind megszakítja a socketet. Az előre telepített kliens OSError kivételt vált ki (vagy egy csupasz kivételt a bróker visszatérési kódjával) abból a hívásból, amely észlelte a megszakadást, és a szokásos minta egy újrapróbálkozási ciklus:

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)

A feliratkozások nem maradnak meg az újracsatlakozások között, hacsak a kliens nem adta át a clean_session=False paramétert csatlakozáskor, ezért a belső connect hívásnak újra ki kell adnia minden subscribe() hívást is, mielőtt belépne a közzétételi ciklusba.

9.19.5. A végakarat-kampó¶

Egy státuszt jelentő kamerának meg kell mondania a brókernek, milyen üzenetet küldjön a kamera nevében, ha a kapcsolat váratlanul megszakad. Állítsd be a végakaratot a connect() előtt

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)

Ettől kezdve bármely, a yard-cam/status topicra feliratkozott irányítópult online állapotot lát abban a pillanatban, amikor a kamera csatlakozik, és offline állapotot, amikor a bróker észleli, hogy a kamera leesett. A megőrzött offline üzenet megmarad a brókeren, így egy tíz perccel később csatlakozó irányítópult is még a helyes aktuális állapotot látja.

9.19.6. Mikor válaszd az MQTT-t a HTTP helyett¶

A webszerverekről szóló fejezet azt az esetet tárgyalja, amikor a kamera HTTP-szerverként működik, a felhőbe töltésről szóló oldalon pedig azt, amikor HTTP-kliensként JPEG-eket küld egy rögzített URL-re. Mindkettőnek megvan a helye. Az MQTT választásának megfelelő pillanata:

Ugyanazoknak az adatoknak több hallgatóhoz kell eljutniuk (egy irányítópult, egy értesítési szolgáltatás, egy rögzítő) anélkül, hogy a kamera előre ismerné a listát.
A hallgatók jöhetnek és mehetnek anélkül, hogy a kamerát újra kellene indítani.
A kamera fel akar iratkozni – hogy parancsokat fogadjon egy vezérlőtől –, amit egy HTTP-kliens nem tud megtenni hosszú lekérdezés (long polling) vagy egy visszahívási URL-re küldő szerver nélkül.
A kapcsolatnak olcsón kell túlélnie a hosszú tétlen időszakokat.

A HTTP-nél maradás megfelelő pillanata: egy kamera, egy szerver, egy rögzített kérés/válasz minta olyan törzzsel, amely túl nagy egyetlen MQTT-topichoz (a JPEG-képkockák MQTT-n működnek, de udvariatlanok a brókerrel szemben; a HTTP POST a természetes választás).

Kereszthivatkozás: a webszerverekről szóló fejezet felhőbe töltésről szóló oldala a „kamera → felhőarchívum” HTTP-változatát mutatja be. Ugyanennek a problémának az MQTT-változata a kamerát függetlenül tartja az archívum URL-jétől, és lehetővé teszi, hogy egy második fogyasztó (mondjuk egy telefonos riasztó alkalmazás) ugyanarra a folyamra csatlakozzon.