MQTT Client (Message Queue Telemetry Transport)

Podporované typy a verzie zariadení 
O protokole MQTT Sparkplug
Konfigurácia komunikačnej linky
Parametre protokolu linky
Konfigurácia komunikačnej stanice
Konfigurácia meraných bodov
Literatúra
Revízie dokumentu

Podporované typy a verzie zariadení

Protokol je implementáciou štandardu MQTT 3.1.1 (október 2014). MQTT protokol je klient/server protokol typu subscribe/publish. Je jednoduchý, má malú réžiu a je ľahko implementovateľný. Používa sa na komunikáciu M2M (Machine to Machine) a v kontexte IoT (Internet of Things). MQTT server sa nazýva aj MQTT broker.
D2000 KOM implementuje klientskú časť protokolu. Protokol je implementovaný na TCP/IP linke.
Pre prenos LoRaWAN dát v rámci MQTT protokolu pozri popis protokolu LoRaWan.

Čiastočne je podporený aj Sparkplug 3.0.0 štandard (Sparkplug B MQTT Payload). Podpora zahŕňa:

• čítanie jednoduchých typov hodnôt
• čítanie polí (s podporou Cieľového stĺpca štruktúry)
• zápis jednoduchých typov hodnôt
• čítanie položiek šablón (UDT)
• čítanie položiek datasetov

Komunikácia bola testovaná/nasadená voči:

• cloudu www.TheThings.network
• cloudu Loriot.io
• cloudu Slovanet
• Pixii PowerShaper (energy storage solution)
• cloudu liveobjects.orange-business.com
• test.mosquitto.org (Sparkplug B)

Pozn: bola otestovaná aj komunikácia voči cloudu liveobjects.orange-business.com cez websockety (wss://liveobjects.orange-business.com:443/mqtt). Ako WSS wrapper bol použitý program https://github.com/jimparis/unwebsockify.git. Tento bol spustený s parametrami:
./unwebsockify.py --port 1883 --listen 172.16.0.1 wss://liveobjects.orange-business.com:443/mqtt
D2000 KOM proces sa pripájal k adrese 172.16.0.1 na port 1883. WSS wrapper sa pripojil na definovanú URL a zabalil dáta MQTT komunikácie do websocket obálky.

Každá PUBLISH správa obsahuje tému (Topic), samotné dáta (Payload) a úroveň potvrdzovania (QoS). PUBLISH správy môže odosielať klient aj server. Klient na začiatku komunikácie použije správu SUBSCRIBE na oznámenie, o aké témy (parameter protokolu Topic Filter) má záujem.
Protokol definuje tieto úrovne potvrdzovania PUBLISH správ - QoS (Quality of Service):

• QoS_0 - správa PUBLISH nie je potvrdzovaná, môže dôjsť ku strate
• QoS_1 - správa PUBLISH je druhou stranou potvrdzovaná PUBACK, môže dôjsť k duplicite
• QoS_2 - správa PUBLISH je druhou stranou potvrdzovaná PUBREC, tá je spätne potvrdzovaná správou PUBREL a tá finálne správou PUBCOMP

Úroveň potvrdzovania správ posielaných procesom D2000 KOM definuje parameter protokolu Publish QoS. D2000 KOM proces považuje zápis výstupného meraného bodu za úspešne ukončený v závislosti od QoS:

• QoS_0 - po úspešnom odoslaní dát cez TCP spojenie
• QoS_1 - po prijatí PUBACK
• QoS_2 - po prijatí PUBCOMP

Komunikácia MQTT sa začína správou CONNECT, ktorú pošle klient (D2000 KOM). Správa obsahuje prihlasovacie meno (User Name), heslo (Password) a ďalšie parametre, z ktorých sa dá nastaviť Clean Session Flag a Client ID (parameter Will Flag sa nevyužíva, podobne ako Will QoS a Will Retain, parameter Keep Alive je nastavený na 0). Server odpovedá správou CONNACK s návratovým kódom, ktorý obsahuje informáciu o úspešnosti pripojenia.
Následne klient pošle správu SUBSCRIBE s filtrom tém (parameter Topic Filter), o ktoré má záujem a požadovanou maximálnou úrovňou potvrdzovania (parameter Subscribe QoS).
Server odpovedá návratovým kódom, ktorý obsahuje informáciu o úspešnosti a maximálny QoS, ktorý bol pre požadované témy pridelený.
Nasleduje bežná komunikácia, počas ktorej klient aj server posielajú PUBLISH správy (klient s ľubovoľnou témou, server s témami zodpovedajúcimi filtru tém prijatej SUBSCRIBE správy) a podľa hodnoty parametra QoS prijatých správ PUBLISH ich potvrdzujú.
Pokiaľ server nepošle žiadnu správu dlhšie ako Ping Interval sekúnd, klient pošle PING request správu, na ktorú server musí odpovedať PING response správou (do času špecifikovaného parametrom Reply Timeout).
Pokiaľ dôjde k zmene parametrov na linke, spojenie je zavreté a znovu vytvorené.

O protokole MQTT Sparkplug

Protokol MQTT Sparkplug definuje 3 skupiny aplikácií:

• Edge Node - má podporu MQTT protokolu a pripája sa k MQTT serveru. Posiela mu údaje získané z Device alebo vlastné, prípadne agregované
• Device/Sensor - reprezentuje fyzické alebo logické zariadenie pripojené k Edge Node a poskytujúce údaje, procesné dáta alebo metriky
• Host Application - reprezentuje spotrebiteľa dát (SCADA/MES systém, Historian, analytický nástroj), ktorý sa pripája k MQTT serveru a prijíma MQTT dáta od Edge Node/Device a prípadne poveluje. D2000 je Host Application.

MQTT Sparkplug definuje nasledovné typy správ:

• NBIRTH – Birth certificate for Sparkplug Edge Nodes (informácia o pripojení sa Edge Node k MQTT Serveru)
• NDEATH – Death certificate for Sparkplug Edge Nodes (informácia o strate spojenia MQTT Servera s Edge Node)
• DBIRTH – Birth certificate for Devices (informácia o pripojení sa Device k Edge Node)
• DDEATH – Death certificate for Devices (informácia o strate spojenia Edge Node s Device)
• NDATA – Edge Node data message (dáta od Edge Node)
• DDATA – Device data message (dáta od Device)
• NCMD – Edge Node command message (príkaz Host Application pre Edge Node)
• DCMD – Device command message (príkaz Host Application pre Device)
• STATE – Sparkplug Host Application state message (informácia od stave Host Application pre Edge Node/Device [online / offline]).

MQTT Sparkplug definuje pre všetky typy správ syntax tém (Topic), ktorý má tvar namespace/group_id/message_type/edge_node_id/[device_id], kde:

• namespace je konštanta spBv1.0 (pre Sparkplug B)
  
• group_id je ľubovolný názov logickej skupiny (napr. podľa typu zariadenia, prevádzky, organizačnej štruktúry atď)
• message_type je typ správy
• edge_node_id je identifikátor Edge Node
• device_id je identifikátor Device (iba ak je správa od/pre Device)

Príklady: 

• spBv1.0/Sparkplug Devices/DDATA/Marvin/Sensor1
• pBv1.0/MIMIC/DDATA/20:19:AB:F4:0E:0D/sensor11
• spBv1.0/SparkplugDevices/NBIRTH/JsonScada

Poznámka: správy typu STATE majú Topic v tvare spBv1.0/STATE/sparkplug_host_id, kde sparkplug_host_id je unikátny názov hostiteľskej aplikácie.

Príklady:

• spBv1.0/STATE/CFGHUB2023
• spBv1.0/STATE/D2Kom

Všetky typy správ výnimkou STATE majú binárny Sparkplug payload. Správy STATE majú JSON payload (obsahuje iba položky "online" (hodnota True/False) a "timestamp" (časová značka v milisekundách).

Sparkplug payload obsahuje metriky - každú požadovanú metriku je možné nakonfigurovať do jedného meraného bodu. Metrika obsahuje textový identifikátor (zadávaný v konfigurácii meraného bodu ako sparkplug_address), typ hodnoty, hodnotu (jednoduchá/pole/štruktúrovaná), voliteľne časovú značku a ďalšie atribúty. D2000 podporuje prácu s jednoduchými hodnotami a hodnotami typu pole - do meraného bodu sa uloží prvá položka poľa; je možné na meranom bode nakonfigurovať Cieľový stĺpec štruktúry, do ktorého sa uložia všetky položky poľa (resp. prvých N, ak nemá štruktúra dostatočný počet riadkov).
Pozn: ak príde metrika, ktorá má nastavený príznak is_null, meraný bod bude mať Invalid hodnotu.

Konfigurácia komunikačnej linky

• Kategória komunikačnej linky: TCP/IP-TCP.
• Host: IP adresa MQTT servera (prípadne redundantné adresy oddelené čiarkou alebo bodkočiarkou).
• Port: číslo portu je štandardne 1883 alebo 8883 pre kryptovanú SSL/TLS variantu.
• Číslo linky: nepoužité, nastavte hodnotu 0.

Pozn: Štandardný port pre MQTT protokol je 1883 resp. 8883 pre SSL/TLS verziu. D2000 KOM neobsahuje implementáciu SSL/TLS varianty protokolu, ale je možné ju nakonfigurovať s použitím utility stunnel http://www.stunnel.org pracujúcej v klientskom móde (client = yes). Stunnel bežiaci na rovnakom počítači ako D2000 KOM by mal počúvať na lokálnom porte 1883 a po pripojení sa D2000 KOM procesu na tento port by mal komunikáciu zakryptovať pomocou SLL/TLS a poslať na cieľový MQTT server (typicky na port 8883).

Pozn: pri použití SSL/TLS spojenia je treba dať pozor na expiráciu použitých certifikátov. Dátum expirácie je možné na Linuxe zistiť príkazom

openssl x509 -text -in file.crt | grep "After"

Vynútené prerušenie spojenia: Keď sa všetky stanice na linke dostanú do simulácie alebo majú zastavenú komunikáciu, linka sa odpojí (dôjde ku zavretiu komunikačného socketu). Keď aspoň jedna stanica bude mať vypnutú simuláciu a nebude mať zastavenú komunikáciu (záložka Parametre objektu Stanica), linka sa znovu pripojí.

Parametre protokolu linky

Dialóg konfigurácia linky - záložka Parametre protokolu.
Ovplyvňujú niektoré voliteľné parametre protokolu. Môžu byť zadané nasledovné parametre protokolu linky:

Tab. č. 1

Konfigurácia komunikačnej stanice

• Komunikačný protokol "MQTT Client Protocol".
• Adresa stanice: adresa stanice zodpovedá poľu Topic v správe PUBLISH prijatej od MQTT servera. Adresou môže byť konkrétna téma (Topic), regulárny výraz,  znak # reprezentujúci všetky témy, alebo topic .* reprezentujúci všetky témy, ktoré nie sú vhodné pre iné stanice. Priorita spracovania je takáto:
  • Pokiaľ je na linke stanica s adresou #, všetky správy sú smerované na jej merané body a ďalej sa nehľadá.
  • Ďalej sú prehľadávané všetky ostatné stanice na linke (s výnimkou adresy .*). Ak s Topic zhoduje s adresou stanice, správa je určená pre ňu a ďalej sa nehľadá.
  • Ďalej sú prehľadávané všetky ostatné stanice na linke (s výnimkou adresy .*), pričom ich adresa je vyhodnocovaná ako regulárny výraz. Ak s Topic zhoduje s adresou stanice, správa je určená pre ňu a ďalej sa nehľadá. Stanice sú prehľadávané v zostupnom poradí (podľa adresy stanice), takže špecifickejšie výrazy idú ako prvé (napr. status/battery pred status/batt.*)
  • Nakoniec ak existuje stanica s adresou .*, správa je určená pre ňu.
• Parametre pollingu na záložke Časové parametre - odporúčaná je hodnota Delay=0.

Poznámka: v prípade SparkPlug MQTT servera má Topic tvar 'namespace/group_id/message_type/edge_node_id/[device_id]', kde message_type udáva typ správy (napr. DDATA, DBIRTH, DDEATH).
Je možné použiť namiesto message_type regulárny výraz (napr. spBv1.0/Sparkplug Devices/.*/MyDevice/Sensor2), aby boli pokryté všetky typy správ.
Ak Payload Type=Sparkplug, je možné vynechať časti namespace aj message_type a zapísať Topic v skrátenom tvare 'group_id/edge_node_id/[device_id]' (napr. Sparkplug Devices/MyDevice/Sensor2).

Pozn: Ak je adresa stanice v skrátenom tvare, nespracúvajú sa pre ňu príkazy (DCMD, NCMD). Ak je v tvare regulárneho výrazu ('namespace/group_id/.*/edge_node_id/[device_id]'), stanica spracúva aj príkazy (včítane príkazu posielaného D2000 KOM procesom, ak Send Node Control/Rebirth=YES). Preto odporúčame adresu stanice v skrátenom tvare. Pokiaľ je potrebné spracovávať aj príkazy (od iných Host Applications), tak vytvoriť inú stanicu s adresou v tvare regulárneho výrazu (napr. spBv1.0/Sparkplug Devices/DCMD/MyDevice).

Tlačidlo Browse otvorí browsovací dialóg pre adresu stanice. Pokiaľ je komunikácia funkčná, zobrazí sa dialóg s doteraz prijatými Topicmi. Tlačidlom Refresh je možné zoznam prijatých Topicov vyčistiť. Dvojité kliknutie na konkrétny riadok spôsobí, že sa hodnota zo stĺpca Address vloží do konfigurácie stanice, z ktorej bol browsovací dialóg otvorený.
Poznámka: v stĺpci Station sa zobrazuje stanica, ktorej bol Topic priradený (na základe vyššie uvedených priorít spracovania). Pre Sparkplug adresy sa zobrazuje skrátený tvar adresy.

Parametre protokolu stanice

Dialóg konfigurácia stanice - pole "Parameter protokolu".
Ovplyvňujú niektoré voliteľné parametre protokolu. Môžu byť zadané nasledovné parametre protokolu stanice:

Tab. č. 2

Konfigurácia meraných bodov

Možné typy hodnôt meraných bodov: Ci, Co, TxtI, TxtO, Qi, Ci, Co, Ai, Ao, Di, Do, TiR, ToR, TiA, ToA.

MQTT implementácia podporuje tri režimy práce:

• Textový režim: Pôvodná implementácia MQTT protokolu obsahovala iba vstupné textové merané body s adresami IN_TOPIC, IN_DATA a voliteľne dvojicu bodov s adresami IN_ID a ACK_ID. Prvé dva body slúžili na zverejnenie prijatého Topicu a Payloadu (ktorý bolo následne potrebné parsovať v skripte), druhé dva body slúžili na zverejnenie identifikátora paketu a potvrdenie spracovania paketu. Tak bolo možné zabezpečiť, že pre dáta posielané s QoS > QoS_0 bolo poslané potvrdenie až po spracovaní dát v skripte.
  Na zápis slúžia výstupné body s adresami OUT_TOPIC a OUT_VALUE.
• JSON režim: Rozšírenie pre spracovanie Payloadu s JSON dátami (Payload Type=JSON) bolo implementované s pomocou vstupných meraných bodov s adresami JA=json_address. D2000 KOM proces priamo parsuje JSON payload a nastavuje hodnoty meraných bodov s JSON adresami. Merané body s adresami IN_TOPIC, IN_DATA, IN_ID a ACK_ID vôbec nemusia existovať.
  Na zápis slúžia výstupné body s adresami OUT_TOPIC a OUT_VALUE.
• Sparkplug režim: Rozšírenie pre spracovanie Payloadu so Sparkplug dátami (Payload Type=Sparkplug) bolo implementované s pomocou vstupných meraných bodov s adresami SA=sparkplug_address. D2000 KOM proces priamo parsuje Sparkplug payload a nastavuje hodnoty meraných bodov so Sparkplug adresami. Merané body s adresami IN_TOPIC, IN_DATA, IN_ID a ACK_ID vôbec nemusia existovať.
  Zápis jednoduchých hodnôt je možný pomocou výstupných bodov s adresami ST=type;SA=sparkplug_address, kde type je definícia Sparkplug dátového typu (napr. Int8, UInt16, DateTime, String atď). Výstupné body musia byť na stanici so Sparkplug adresou, keďže Topic pri zápise sa odvodzuje od nej.
  Pozn: správy typu STATE, ktoré majú JSON payload, je možné parsovať s meranými bodmi s JSON adresou (typicky meraný bod typu Di s adresou JA=online)

{
    "metrics": [
        {
            "datatype": 3,
            "int_value": 7338992,
            "name": "Corrected Vol Acc Stn",
            "timestamp": 1729664005479
        }
    ],
    "seq": 32,
    "timestamp": 1729664005479
}

{
    "metrics": [
        {
            "datatype": 12,
            "datatype_txt": "String",
            "name": "Node Properties/Configuration",
            "string_value": "{}",
            "timestamp": 1730305529539,
            "timestamp_txt": "30-10-2024 17:25:29.539"
        },
        {
            "alias": 30064771073,
            "datatype": 5,
            "datatype_txt": "Uint8",
            "int_value": 0,
            "name": "Node Properties/Missing Param",
            "properties": {
                "keys": [
                    "usage"
                ],
                "values": [
                    {
                        "string_value": "technical information",
                        "type": 12,
                        "type_txt": "String"
                    }
                ]
            },
            "timestamp": 1730305529537,
            "timestamp_txt": "30-10-2024 17:25:29.537"
        },
        {
            "alias": 0,
            "dataset_value": {
                "columns": [
                    "topic_name",
                    "offset",
                    "length",
                    "crc"
                ],
                "num_of_columns": 4,
                "rows": [
                    {
                        "row": [
                            "N/A",
                            0,
                            0,
                            0
                        ]
                    }
                ],
                "types": [
                    12,
                    7,
                    7,
                    7
                ],
                "types_txt": [
                    "String",
                    "UInt32",
                    "UInt32",
                    "UInt32"
                ]
            },
            "datatype": 16,
            "datatype_txt": "DataSet",
            "name": "Node Control/FW Update",
            "timestamp": 1730305529537,
            "timestamp_txt": "30-10-2024 17:25:29.537"
        }
    ],
    "seq": 0,
    "timestamp": 1730305529536,
    "timestamp_txt": "30-10-2024 17:25:29.536"
}

Poznámka: je možné monitorovať aj stav iných Sparkplug Host Application pripojených k MQTT serveru. Ak je Identifikátor Host Application napr. "ACME", tak je potrebné vytvoriť stanicu s adresou "spBv1.0/STATE/ACME" (alebo v skrátenom tvare "ACME") a na nej meraný bod typu Di s adresou "JA=online" (keďže Host Application posiela STATE správu s JSON payloadom).

Browse

Pre merané body v Sparkplug režime je možné zistiť zoznam metrík, pokiaľ KOM proces beží a komunikácia s MQTT serverom je nadviazaná.
Po kliknutí na tlačidlo Browse sa otvorí okno MQTT Item Browser a zobrazí sa zoznam doteraz načítaných metrík. Zoznam objektov sa vytvára dynamicky v dôsledku prijatých správ .

Zoznam objektov je dynamický, t.j. pri príchode novej hodnoty do KOM procesu je aktualizovaný. Podporované je aj filtrovanie v jednotlivých stĺpcoch (s výnimkou časového stĺpca Time), v maske je možné používať hviezdičky (napr. *Short*).

Dvojité kliknutie na konkrétny riadok spôsobí, že sa hodnota zo stĺpca Address vloží do konfigurácie meraného bodu, z ktorého bolo okno MQTT Item Browser otvorené. Zároveň sa vloží aj do clipboardu.

Tlačidlo Refresh spôsobí vymazanie zoznamu hodnôt v CNF aj v KOM procese a prípadne poslanie Rebirth príkazu (ak Send Node Control/Rebirth=YES).

V stĺpci Value sa nachádza prijatá hodnota.

Literatúra

Odkazy
Oficiálna stránka MQTT protokolu http://mqtt.org

Stránka MQTT Sparkplug protokolu https://sparkplug.eclipse.org

Špecifikácie a štandardy
MQTT 3.1.1 špecifikácia http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html
ISO/IEC 20922:2016 http://www.iso.org/standard/69499.html

Popisy dátových formátov a API
www.loriot.io - Application API Data Format https://www.loriot.io/home/documentation.html#docu/app-data-format
www.thethingsnetwork.org - API Reference https://www.thethingsnetwork.org/docs/applications/mqtt/api.html

Revízie dokumentu

• Ver. 1.0 - 8. august 2017 - vytvorenie dokumentu.
• Ver. 1.1 - 15. október 2021 - podpora parametrov pre LastWill a Retain
• Ver. 1.2 - 27. október 2021 - podpora parsovania JSON správ
• Ver. 1.3 – 1. február 2022 – podpora časových značiek v správach JSON
• Ver. 1.4 – 4. októbra 2024 – podpora pre MQTT Sparkplug, podpora pre browsovanie