Notifikačná služba
Upozornenie
Predtým, ako prídete na cvičenie, tak:
nainštalovať vývojové prostredie PyCharm Professional
vytvoriť si účet v službe Pushbullet
nainštalovať si klientsku aplikáciu Pushbullet do telefónu alebo do prehliadača
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.
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
.
Task
Do prostredia svojho projektu nainštalujte potrebné balíky pre vývoj služby.
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
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
.
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):
f"Connected with result code {reason_code}")
logger.debug(
def on_message(client: mqtt.Client, userdata, msg: mqtt.MQTTMessage):
f'{msg.topic}: {msg.payload}')
logger.debug(
def main():
= mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client "maker", "mother.mqtt.password")
client.username_pw_set(
= on_connect
client.on_connect = on_message
client.on_message
connect("147.232.205.176", 1883, 60)
client.
'Waiting for messages.')
logger.info(
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)
asys.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ôlepayload
- samotná správa, ktorá bude reprezentovať poslednú vôľuqos
- úroveň QoS, predvolene je nastavená hodnota0
retain
- príznakretain
, predvolene je nastavený na hodnotuFalse
properties
- ďalšie nastavenia, ktorú sú predvolene prázdne
Vytvorenie poslednej vôle môže vyzerať nasledovne:
'services/notifier/ab123cd/status', {'status': 'offline'}, retain=True) client.will_set(
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.
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):
| list[AnyUrl]
urls: AnyUrl str = 'Alert'
title: str
body: list | None = None attachments:
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
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.
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.
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
- HiveMQ: How to Use the Paho MQTT Client in Python - MQTT Client Library Encyclopedia
- Beginner’s Guide To Using Paho-MQTT, A Python MQTT Client
- Apprise- Push Notifications that work with just about every platform!
- Pydantic - The most widely used data validation library for Python.
- Eclipse
Paho™ MQTT Python Client - Kompletná dokumentácia knižnice
paho
.