Práca s protokolom MQTT

pripojenie sa, publikovanie správ, prihlásenie sa na odber správ, nastavenie poslednej vôle

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ý:

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ý:

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:

Kód Názov Význam
0 Connection Accepted pripojenie úspešné
1 Unacceptable Protocol Version Klient používa nepodporovanú verziu MQTT
2 Identifier Rejected Client ID neakceptovaný (prázdny, duplicitný, neplatný)
3 Server Unavailable Broker je dostupný, ale neprijíma nové pripojenia
4 Bad Username or Password Chyba v autentifikácii
5 Not Authorized Klient nie je autorizovaný – broker ho odmietol kvôli prístupovým právam

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.

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ý:

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ý:

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

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:

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:

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:

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)

Ďalšie zdroje