Notifikačná služba

jednoduchá služba v jazyku Python na odosielanie notifikácií pomocou protokolu MQTT

Motivácia

Súčasťou riešení Internetu vecí nie sú len chytré zariadenia, ale aj softvérové služby. A práve jednu takú jednoduchú notifikačnú službu si urobíme počas tohto cvičenia. Pomocou tejto služby budeme vedieť z našich zariadení posielať notifikácie prostredníctvom protokolu MQTT. V našej infraštruktúre sa táto služba bude nachádzať v cloud-e.

Smart Department: Architektúra

Službu vytvoríme v jazyku Python. Na posielanie notifikácií budeme používať knižnicu Apprise, vďaka čomu budeme vedieť využiť na notifikácie desiatky rôznych služieb.

Ciele

  • naučiť sa základy práce s knižniciou Paho-MQTT

  • naučiť sa vytvoriť poslednú vôľuv MQTT

  • naučiť sa základy práce s knižnicou Apprise

  • naučiť sa vytvárať modely pomocou knižnice Pydantic

Postup

Bootstrap

Predtým, ako sa vrhneme do vývoja samotnej služby, si pripravíme prostredie.

Task

Vo svojom vývojovom prostredí si vytvorte nový projekt s názvom Notifier.

Upravte si štruktúru súborov projektu nasledovne:

./
├── readme.md
├── requirements.txt
└── src/
    └── main.py

Do súboru readme.md napíšte jednoduchý opis projektu vo formáte Markdown. Jeho obsah môže vyzerať nasledovne:

# Notifier

Simple service for sending notifications over _MQTT_ protocol, which are 
delivered with [Apprise](https://github.com/caronc/apprise) library.

Ostatné súbory zatiaľ nechajte prázdne.

Task

Do prostredia svojho projektu nainštalujte potrebné balíky pre vývoj služby.

Pre vytvorenie služby budeme potrebovať tieto balíky:

  • apprise - Knižnica pre posielanie push notifikácií.
  • loguru - Knižnica pre jednoduché logovanie v jazyku Python.
  • paho-mqtt - Knižnica pre komunikáciu pomocou protokolu MQTT.
  • pydantic - Knižnica pre validáciu údajov.
  • pydantic-settings - Rozšírenie knižnice Pydantic pre načítanie konfigurácie (nie len) z premenných prostredia.

Vytvorte si samostatný súbor requirements.txt, do ktorého vymenujte všetky potrebné balíky, ktoré chcete nainštalovať:

apprise
loguru
paho-mqtt
pydantic
pydantic-settings

Poznámka: Aby ste sa vyhli problémom, je dobré okrem názvu balíka do tohto súboru uviesť aj verziu knižnice. Túto informáciu môžete získať napríklad príkazom pip freeze. Ten vám vypíše zoznam všetkých nainštalovaných balíkov s ich aktuálnymi verziami.

Z príkazového riadku môžete potrebné balíky nainštalovať pomocou príkazu pip takto:

$ pip install apprise paho-mqtt pydantic pydantic-settings loguru

Rovnako ich však môžete nainštalovať pomocou vytvoreného súboru requirements.txt takto:

$ pip install --requirement requirements.txt

The Service Skeleton

V tomto kroku vytvoríme kostru aplikácie a pripojíme sa s ňou k MQTT brokeru. Na komunikáciu v jazyku Python s protokolom MQTT budeme používať knižnicu paho-mqtt.

Projekt paho

Task

Pomocou nasledujúceho kódu vytvorte kostru služby Notifier.

from loguru import logger
import paho.mqtt.client as mqtt


def on_connect(client: mqtt.Client, userdata, flags, reason_code, properties):
    logger.debug(f"Connected with result code {reason_code}")


def on_message(client: mqtt.Client, userdata, msg: mqtt.MQTTMessage):
    logger.debug(f'{msg.topic}: {msg.payload}')


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    client.username_pw_set("maker", "mother.mqtt.password")

    client.on_connect = on_connect
    client.on_message = on_message

    client.connect("147.232.205.176", 1883, 60)

    logger.info('Waiting for messages.')
    client.loop_forever()


if __name__ == '__main__':
    main()

Task

Vo funkcii .on_connect() sa prihláste na odber správ tém services/notifier/ab123cd a services/notifier/ab123cd/cmd.

Význam tém je nasledujúci:

  • services/notifier/ab123cd - Pomocou tejto témy bude možné posielať vlastné notifikácie.
  • services/notifier/ab123cd/cmd - Pomocou tejto témy bude možné službe posielať príkazy na diaľku, čím ju bude možné ovládať.

Task

Overte správnosť svojej implementácie.

Na overenie použite MQTT klienta MQTT Explorer. Ak zašlete ľubovoľnú správu do ľubovoľnej z uvedených dvoch tém, v logoch aplikácie sa zobrazí prijatá správa spolu s informáciou o tom, do ktorej témy bola správa odoslaná.

Handling Commands

Službu, ktorú vytvárame, budeme môcť ovládať na diaľku. Na to bude slúžiť téma services/notifier/ab123cd/cmd. Príkaz bude odoslaný vo forme JSON dokumentu. V tomto kroku zabezpečíme ošetrenie príkazu shutdown na vypnutie služby, ale pripravíme ju na možnosť rozšíriť ju o ľubovoľný ďalší príkaz.

Task

Vytvorte kostru funkcie handle_command(), ktorá sa zavolá po prijatí správy v téme services/notifier/ab123cd/cmd.

Signatúra funkcie bude rovnaká, ako v prípade callback funkcie on_message(). Okrem logovania môžeme zabezpečiť konverziu prijatej správy na typ slovník v jazyku Python.

Task

Pomocou metódy .message_callback_add() nad objektom MQTT klienta nastavte vytvorenú funkciu handle_command() ako callback funkciu pre správy prichádzajúce do témy services/notifier/ab123cd/cmd. Implementáciu overte.

Po odoslaní správy do témy pre príkazy sa vám v logoch zobrazí príslušná správa.

Task

Zabezpečte, aby sa služba po prijatí príkazu shutdown korektne vypla.

Príkaz na vypnutie bude vyzerať nasledovne:

{
    "cmd": "shutdown"
}

Po prijatí príkazu shutdown službu ukončite. To môžete urobiť dvoma spôsobmi:

  • natvrdo zavolaním funkcií quit(0) a sys.exit(0), alebo
  • korektne odpojením klienta od brokera a ukončením aplikačnej slučky

Urobte všetko preto, aby ste službu ukončili korektne.

V prípade, že bol prijatý neznámy príkaz, túto udalosť vhodne zalogujte pomocou úrovne warning.

Task

Pomocou MQTT klienta MQTT Explorer overte, či sa služba naozaj korektne ukončí.

Službe pošlite správu na jej ukončenie. Ak ste postupovali správne, služba sa korektne ukončí. Rovnako overte, či sa v logoch zobrazí korektná správa v prípade prijatia neznámeho príkazu.

The Last Will

Keď sa služba odpojí alebo bude na diaľku vypnutá, bolo by dobré oznámiť všetkým ostatným, že už ďalej nepracuje. Vytvoríme preto samostatnú tému s postfixom /status, kde budeme informovať o jej aktuálnom stave. To znamená, že:

  • keď službu zapneme, pošleme do nej správu o tom, že sme v stave online
  • keď službu vypneme alebo z rozličných dôvodov dôjde k strate spojenia so službou, broker za nás nastaví jej nový stav na offline

Task

Po pripojení služby k brokeru pošlite do témy s postfixom /status správu {"status": "online"}.

Task

Pri inicializácii MQTT klienta nastavte pomocou metódy .will_set() poslednú vôľu služby.

Signatúra tejto metódy vyzerá nasledovne:

.will_set(topic, payload, qos, retain, properties)

Ich význam je nasledovný:

  • topic - téma pre odoslanie poslednej vôle
  • payload - samotná správa, ktorá bude reprezentovať poslednú vôľu
  • qos - úroveň QoS, predvolene je nastavená hodnota 0
  • retain - príznak retain, predvolene je nastavený na hodnotu False
  • properties - ďalšie nastavenia, ktorú sú predvolene prázdne

Vytvorenie poslednej vôle môže vyzerať nasledovne:

client.will_set('services/notifier/ab123cd/status', {'status': 'offline'}, retain=True)

Task

Overte správnosť svojej implementácie.

Overte, či správa v téme so stavom služby, sa po vypnutí služby korektne aktualizuje. Správanie overte pre všetky situácie:

  • stav online - zapnutie služby
  • stav offline - korektné vypnutie služby, ako aj nekorektné vypnutie služby (napr. jej vypnutím z vývojového prostredia)

Na overenie správania použite MQTT klienta MQTT Explorer.

The Notifications

V tomto kroku zabezpečíme samotné posielanie notifikácií a upozornení. Pre samotné správy, ktoré budu reprezentovať notifikáciu, vytvoríme model, pomocou ktorého budeme prijaté správy vedieť aj validovať. Na tento účel využijeme knižnicu Pydantic. Samotné notifikácie budeme posielať pomocou knižnice Apprise.

Knižnica Apprise

Task

V module models.py vytvorte model s názvom Notification, ktorý bude reprezentovať príslušnú notifkáciu.

Vytvorte samostatný modul s názvom models.py, do ktorého pridáme definície všetkých modelov, ktoré budeme používať. Model definujúci notifikáciu je uvedený nižšie:

from pydantic import BaseModel, AnyUrl

class Notification(BaseModel):
    urls: AnyUrl | list[AnyUrl]
    title: str = 'Alert'
    body: str
    attachments: list | None = None

Task

Vytvorte funkciu handle_notification(), ktorá bude slúžiť ako callback funkcia pre správy prichádzajúce do témy services/notifier/ab123cd.

Funkcia bude mať rovnakú signatúru, ako callback funkcia on_message().

Task

Pomocou metódy .message_callback_add() nad objektom MQTT klienta nastavte vytvorenú funkciu handle_notification() ako callback funkciu pre správy prichádzajúce do témy services/notifier/ab123cd.

Task

Overte správnosť svojej implementácie.

Správa reprezentujúca notifikáciu bude v našom prípade vyzerať nasledovne:

{
  "urls": "pbul://token-of-receiver",
  "title": "Hello world!",
  "body": "This message was sent with Apprise library, delivered by Paho MQTT Client and validated with Pydantic.",
}

Po jej úspešnom odoslaní do témy pre notifikácie sa vám správa zobrazí v telefóne alebo v prehliadači. V prípade, že odošlete nevalídnu správu, služba sa ukončí s chybou validácie.

Ďalšie úlohy

  1. Knižnica Apprise umožňuje naraz poslať notifikáciu viacerým príjemcom. Model notifikácie podporu pre viacerých príjemcov už má, ale počas cvičenia sme túto funkcionalitu neimplementovali. Upravte ju teda tak, aby bolo možné prijatú správu poslať viacerým príjemcom na základe zoznamu príjemcov v prijatej správe cez protokol MQTT.

  2. Vo vytvorenej implementácii nijako nevalidujeme prijatý príkaz. Vytvorte preto samostatný model aj pre príkazy. Pozrite sa na možnosti knižnice Pydantic, ktoré je možné využiť na validovanie platných príkazov. Zatiaľ máme síce len jeden, ale rozšíriť ich zoznam o ďalšie by tým pádom nemal byť problém.

  3. Pokiaľ kedykoľvek dôjde k výnimke (najmä napr. pri validácii správ prijatých cez protokol MQTT), aplikácia sa ukončí. Oštritre preto možné výskyty výnimiek, aby k ukončeniu aplikácie nedošlo, ale miesto toho vzniknuté chyby logujte.

Ďalšie zdroje