Práca s protokolom MQTT

Komunikačný protokol MQTT patrí medzi najpopulárnejšie a najpoužívanejšie protokoly v IoT riešeniach. To je dané jeho jednoduchosťou a výrazne nižším množstvom prenášaných dát v porovnaní s inými protokolmi.

V tomto návode sa pozrieme na to, ako ho použiť v jazyku MicroPython.

Podpora protokolu v jazyku MicroPython

Modul pre prácu s MQTT sa nenachádza v každom firmvéri s jazykom MicroPython. Nájdete ho napr. vo firmvéri pre mikrokontrolér ESP32, ale nenájdete ho vo firmvéri pre mikrokontroléry z rodiny RP2 (napr. RP2350 na doske RPi Pico 2 W). To, či ho máte alebo nemáte k dispozícii, si môžete overiť vypísaním zoznamu všetkých dostupných modulov príkazom:

>>> help('modules')

Ak ho nemáte, môžete ho nainštalovať priamo z prostredia REPL jazyka MicroPython týmito príkazmi:

>>> import mip
>>> mip.install('umqtt.simple')

Pripojenie k brokeru

Aby sme sa mohli pripojiť na MQTT broker, potrebujeme vytvoriť objekt typu MQTTClient a pri vytváraní mu dodať potrebné informácie pre pripojenie. Trieda MQTTClient sa nachádza v balíku umqtt.simple:

from umqtt.simple import MQTTClient

Vytvorenie MQTT klienta vyzerá nasledovne:

client = MQTTClient(
    client_id,
    server,
    port=0,
    user=None,
    password=None,
    keepalive=0,
    ssl=None
)

Význam jednotlivých parametrov je nasledovný:

• client_id - Jednoznačný identifikátor klienta, ktorým sa odlišuje od ostatných klientov pripojených k rovnakému MQTT brokeru. Ide o povinný parameter.
• server - Doménové meno alebo IP adresa MQTT brokera (bez schémy, ako napr. mqtt://). Ide o povinný parameter.
• port - Port, na ktorom broker počúva. Ak ho nezadáme, použije sa port 1883, čo je štandardný port pre komunikáciu pomocou protokolu MQTT alebo sa použije port 8883, ak je nastavený parameter ssl.
• user a password - V prípade, že pripojenie k MQTT brokeru si vyžaduje prihlasovacie meno a heslo, zadáte ich pomocou týchto dvoch parametrov.
• keepalive - Maximálny čas v sekundách medzi dvoma správami od klienta. Ak broker v tomto intervale (presnejšie v 1,5-násobku tohto intervalu) nedostane žiadnu správu, považuje klienta za nedostupného a odošle jeho poslednú vôľu. Hodnota 0 túto kontrolu vypína. Knižnica umqtt.simple správu PING neposiela sama – ak je spojenie otvorené dlhšie ako tento interval a klient nič neodosiela, je potrebné pravidelne volať metódu .ping(), inak ho broker odpojí.
• ssl - Objekt triedy ssl.SSLContext, pomocou ktorého bude spojenie šifrované protokolom TLS (napr. ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)). Staršia podoba ssl=True (spolu s parametrom ssl_params) je v knižnici ponechaná len kvôli spätnej kompatibilite. Pri nej sa však brokeru neposiela meno servera (SNI), a preto ho niektoré brokery môžu odmietnuť.

Pre vygenerovanie jedinečného identifikátora klienta môžete využiť napr. tento fragment kódu:

import binascii
import machine

client_id = binascii.hexlify(machine.unique_id())

Pripojenie sa k brokeru

Po vytvorení klienta sa môžete pokúsiť pripojiť pomocou volania metódy .connect(). Signatúra metódy je nasledovná:

.connect(self, clean_session=True, timeout=None)

Význam jednotlivých parametrov metódy je nasledovný:

• clean_session - Ak má hodnotu True, broker po pripojení začne novú reláciu a zahodí všetko, čo si o klientovi pamätal z predchádzajúceho pripojenia (napr. odbery tém a nedoručené správy s QoS 1). Ak má hodnotu False, broker v predchádzajúcej relácii pokračuje.
• timeout - Časový limit v sekundách pre operácie so soketom, napr. pre samotné pripojenie. Predvolená hodnota None znamená, že sa bude čakať bez obmedzenia.

client = MQTTClient(CLIENT_ID, server)
client.connect()
# tu príde komunikácia

Ak sa pripojiť nepodarí, dôjde k vyvolaniu výnimky MQTTException z balíka umqtt.simple s kódom chyby (kód -1 znamená, že broker neposlal platnú odpoveď). Ak sa pripojenie podarí, metóda vráti príznak session present: hodnotu 1, ak broker pokračuje v predchádzajúcej relácii klienta, inak hodnotu 0. Pri parametri clean_session=True je to teda vždy hodnota 0.

V nasledujúcej tabuľke sa nachádzajú kódy chýb podľa MQTT špecifikácie:

OSError: [Errno 12] ENOMEM

Odpojenie sa od brokera

Odpojiť sa od brokera sa môžete zavolaním metódy .disconnect() nad objektom klienta. Táto metóda nemá žiadny parameter.

client = MQTTClient(CLIENT_ID, server)
client.connect()
# tu príde komunikácia
client.disconnect()

Ak sa klient ešte nepripojil, dôjde k vyvolaniu výnimky AttributeError, pretože klient ešte nemá vytvorený soket. Ak už bol klient odpojený, dôjde k vyvolaniu výnimky OSError.

Overenie brokera pomocou certifikátu

Šifrovanie pomocou protokolu TLS zabezpečí, že komunikáciu medzi klientom a brokerom nikto nebude môcť odpočúvať. Samo osebe však nezaručí, že sa klient naozaj pripojil k správnemu brokeru, a nie k útočníkovi, ktorý sa zaňho vydáva. Na to slúži overenie certifikátu brokera: klient overí, či certifikát brokera podpísala certifikačná autorita (z angl. Certificate Authority, CA), ktorej dôveruje, a či je certifikát vydaný na meno servera, ku ktorému sa pripája.

Na overenie potrebujete certifikát certifikačnej autority, ktorá certifikát brokera podpísala. Ak broker používa vlastnú certifikačnú autoritu, dostanete jej certifikát od správcu brokera. Nahrajte ho na mikrokontrolér, napr. do súboru /certs/ca.crt. Certifikát môže byť vo formáte PEM (textový formát, ktorý začína riadkom -----BEGIN CERTIFICATE-----) aj DER (binárny formát).

Pre šifrované spojenie s overením certifikátu vytvorte objekt triedy ssl.SSLContext, načítajte doň certifikát certifikačnej autority a zapnite overovanie. Takto pripravený objekt potom odovzdajte klientovi v parametri ssl:

import ssl
from umqtt.simple import MQTTClient

context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.load_verify_locations(cafile='/certs/ca.crt')
context.verify_mode = ssl.CERT_REQUIRED

client = MQTTClient(CLIENT_ID, 'broker.example.com', port=8883, ssl=context)
client.connect()

Meno servera, ktoré sa porovná s menom v certifikáte brokera, doplní knižnica umqtt.simple sama podľa parametra server. Ak je certifikát načítaný v pamäti ako postupnosť bajtov, odovzdáte ho parametrom cadata, napr. context.load_verify_locations(cadata=data).

Pri overovaní certifikátu dávajte pozor na tieto veci:

• Overovanie musíte zapnúť. Predvolená hodnota verify_mode je ssl.CERT_NONE, takže bez nastavenia hodnoty ssl.CERT_REQUIRED sa spojenie síce zašifruje, ale certifikát brokera sa neoverí.
• Mikrokontrolér musí mať nastavený správny čas. Certifikát má obmedzenú dobu platnosti, ktorá sa pri overení kontroluje. Po zapnutí však mikrokontrolér nepozná aktuálny čas (napr. na doske RPi Pico 2 W je to 1. 1. 2021), a overenie preto zlyhá. Pred pripojením k brokeru preto čas nastavte, napr. pomocou protokolu NTP volaním ntptime.settime().
• Pripájajte sa pomocou mena servera, nie pomocou jeho IP adresy. Certifikát je vydaný na meno servera a pri IP adrese sa mená nezhodujú.

Ak overenie zlyhá, metóda .connect() vyvolá výnimku ValueError s popisom problému, napr.:

Nastavenie poslednej vôle

Posledná vôľa (z angl. Last Will and Testament) je špeciálna správa, ktorú MQTT broker automaticky odošle za klienta, ak sa klient neočakávane odpojí (napr. kvôli výpadku napájania, signálu alebo pádu programu).

Poslednú vôľu odovzdá klient brokeru pri pripájaní – je súčasťou správy, ktorou sa klient k brokeru pripája. Nastaviť ju je možné zavolaním metódy .set_last_will() nad objektom klienta, a to ešte pred zavolaním metódy .connect(). Ak ju nastavíte až po pripojení, broker sa o nej nedozvie. Signatúra metódy je nasledovná:

.set_last_will(self, topic, msg, retain=False, qos=0)

Význam jednotlivých parametrov metódy je nasledovný:

• topic - Názov témy, do ktorej bude posledná vôľa odoslaná.
• msg - Samotná správa, ktorá reprezentuje poslednú vôľu.
• retain - Nastavenie príznaku retain pre správu poslednej vôle.
• qos - Úroveň QoS.

Správa, ktorá sa posiela ako posledná vôľa, je obyčajne veľmi jednoduchá. Obsahuje totiž len informáciu o tom, že je zariadenie vypnuté. Ak sa posiela vo formáte JSON, môže vyzerať napríklad takto:

{
  "status": "offline"
}

Odoslanie dát na MQTT broker

Pre odoslanie správy (tzv. publikovanie) použijeme metódu nad MQTT klientom s názvom .publish(). Signatúra tejto metódy vyzerá nasledovne:

.publish(self, topic, msg, retain=False, qos=0)

Význam jednotlivých parametrov metódy je nasledovný:

• topic - Názov témy, do ktorej bude správa poslaná. Musí byť serializovateľný do poľa bytov.
• msg - Samotná správa. Musí byť serializovateľná do poľa bytov.
• retain - Označenie správy príznakom retain. Predvolene správa nie je s týmto príznakom odoslaná.
• qos - Úroveň QoS, ktorá môže byť 0 alebo 1. Predvolená hodnota je 0. Úroveň 2 knižnica umqtt.simple nepodporuje.

Je dôležité, aby parametre topic, ako aj msg boli serializovateľné do poľa bytov. Ak by ste sa napríklad pokúsili odoslať priamo slovník, došlo by k vyvolaniu výnimky TypeError. Slovník preto pred odoslaním prevedte na reťazec pomocou balíka json.

Príklad použitia môže vyzerať takto:

topic = 'kpi/kronos/humidity/ab123cd'
message = {
  'dt': '2024-10-30T16:55:06Z',
  'metrics': [
    {
      'dt': '2024-10-30T16:50:00Z',
      'name': 'humidity',
      'unit': 'percent',
      'value': 77
    }
  ]
}
client.publish(topic, json.dumps(message))

client.publish(topic, b'', retain=True)

$ mosquitto_pub -h <broker> -t <téma> -r -n

Prijímanie správ z MQTT brokera

Pokiaľ chceme správy z MQTT brokera prijímať, musíme sa prihlásiť na ich odber z konkrétnej témy. Samozrejme naraz je možné byť prihlásený na odber správ z viacerých tém.

Na zabezpečenie prijímania správ je potrebné spraviť nasledovné:

1. vytvoriť tzv. callback funkciu, ktorá sa zavolá vtedy, keď zariadenie dostane správu
2. prihlásiť sa na odber správ do príslušnej témy
3. skontrolovať dostupnosť správ na brokeri

Vytvorenie callback funkcie

Callback funkcia je normálna funkcia, ktorá však musí mať dva parametre:

• topic - názov témy, v ktorej došlo k prijatiu správy, a
• message - samotná prijatá správa.

Oba parametre sú reprezentované postupnosťou bytov. Preto ich netreba zabudnúť konvertovať do požadovaných údajových typov.

Funkcia nič nevracia.

Jednoduchá podoba callback funkcie, v ktorej je téma vo formáte reťazec a správa je prenášaná vo formáte JSON, môže vyzerať nasledovne:

def on_message(topic: bytes, message: bytes):
    topic = topic.decode('utf-8')
    message = json.loads( message )
    print(f'>>> {topic}: {message}')

Callback funkciu je potrebné zaregistrovať u klienta ešte predtým, ako dôjde k pripojeniu k MQTT brokeru, resp. predtým, ako dôjde k prihláseniu sa na odber správ z konkrétnej témy. To je možné zabezpečiť volaním metódy .set_callback() nad objektom klienta. Táto metóda má jediný parameter, ktorým je adresa callback funkcie.

Zaregistrovanie vytvorenej callback funkcie on_message() bude vyzerať takto:

client = MQTTClient(CLIENT_ID, server)
client.set_callback(on_message)
client.connect()

Prihlásenie sa na odber správ

Prihlásiť sa na odber tém je možné pomocou metódy .subscribe() nad objektom klienta. Táto metóda má dva parametre:

• topic - názov témy, do ktorej sa chceme prihlásiť na odber správ, a
• qos - úroveň QoS, ktorá môže byť 0 alebo 1 (úroveň 2 knižnica umqtt.simple nepodporuje).

Metóda nič nevracia.

Prihlásiť sa na odber je možné až potom, keď je klient pripojený k MQTT brokeru. Okrem toho musí mať klient zadefinovanú callback funkciu. Ak to tak nie je, prihlásenie na odber skončí s výnimkou.

Príklad prihlásenia sa na odber správ z témy kpi/hyperion/switch/cd456ef bude vyzerať takto:

client = MQTTClient(CLIENT_ID, server)
client.set_callback(on_message)
client.connect()

client.subscribe('kpi/hyperion/switch/cd456ef')

Kontrola doručených správ

Overiť dostupnosť správ je však potrebné vykonať ručne. To je možné vykonať dvoma metódami nad objektom MQTT klienta:

• .check_msg() - Neblokujúca metóda. Skontroluje, či klientovi neprišla od brokera správa. Ak neprišla, metóda hneď skončí. Ak prišla, spracuje ju pomocou callback funkcie a skončí.

• .wait_msg() - Blokujúca metóda. Zastaví vykonávanie programu a bude čakať na prijatie jednej správy. Tú následne spracuje pomocou callback funkcie a skončí.

Metódu .wait_msg() je možné s výhodou využiť v prípadoch, keď funkcionalita zariadenia je vyslovene závislá od správ, ktoré prijme zo servera. V tomto prípade môžeme hovoriť o okamžitom spracovaní prijatých správ.

Príklad jej použitia môže vyzerať nasledovne:

while True:
  print('Waiting for message...')
  client.wait_msg()  # blocking call
  print('Message was processed')

Naopak metóda .check_msg() je výhodná vtedy, ak je potrebné len rýchlo overiť a spracovať prijatú správu a následne napr. zaspať. Broker však správy nedrží, kým si ich klient nevyzdvihne – posiela ich len klientom, ktorí sú práve pripojení a prihlásení na odber. Zariadenie, ktoré väčšinu času spí, by tak správy odoslané počas spánku nedostalo. Preto je v tomto prípade veľmi dôležité, aby správa mala v danej téme príznak retain. Broker si ju potom uloží a pošle ju každému klientovi hneď, ako sa na odber témy prihlási – teda aj zariadeniu po jeho prebudení.

Príklad jej použitia môže vyzerať nasledovne:

client.check_msg()
machine.deepsleep(10 * 1000)

sleep(0.1)
client.check_msg()
machine.deepsleep(10 * 1000)

Ďalšie zdroje

• MicroPython: umqtt.simple - GitHub repozitár s implementáciou modulu umqtt.simple

• MicroPython: umqtt.robust - GitHub repozitár s implementáciou modulu umqtt.robust