mqtt — Jednostavni MQTT klijent#

Modul mqtt pruža minimalnu implementaciju MQTT v3.1.1 klijenta prikladnu za uporabu na uređajima s ograničenom memorijom. Podržava povezivanje s brokerom (s opcijskim TLS-om), objavljivanje poruka, pretplatu na teme, konfiguraciju posljednje volje i keep-alive pingove.

Primjer:

from mqtt import MQTTClient

def callback(topic, msg):
    print(topic, msg)

client = MQTTClient("openmv", "broker.example.com", 1883)
client.set_callback(callback)
client.connect()
client.subscribe(b"openmv/in")
client.publish(b"openmv/out", b"hello")
while True:
    client.check_msg()

Iznimke#

exception mqtt.MQTTException#

Podiže se kada broker odbije CONNECT zahtjev ili kada se SUBSCRIBE zahtjev odbije. Jedini argument je numerički povratni kod koji isporučuje broker.

Klase#

class mqtt.MQTTClient(client_id: bytes | str, server: str, port: int, ssl_params: dict | None = None, user: bytes | str | None = None, password: bytes | str | None = None, keepalive: int = 0, callback: callable | None = None)#

Konstruira MQTT klijent.

Argumenti:

  • client_id – jedinstveni identifikator klijenta koji se šalje brokeru.

  • server – naziv računala (hostname) ili IP adresa brokera (razrješava se pomoću socket.getaddrinfo()).

  • port – TCP port brokera (obično 1883 za nešifrirano ili 8883 za TLS).

  • ssl_params – ako nije None, soket se omata pomoću ssl.wrap_socket(), a ssl_params se prosljeđuje kao imenovani argumenti. Proslijedite {} za omogućavanje TLS-a sa zadanim postavkama.

  • user – opcijsko korisničko ime za autentifikaciju s brokerom. Ako je navedeno, mora se navesti i password.

  • password – opcijska lozinka koja se koristi zajedno s user.

  • keepalive – interval keep-alive u sekundama (0 onemogućuje). Mora biti manji od 65536.

  • callback – pozivni objekt koji se poziva kao callback(topic, msg) za svaki PUBLISH isporučen od brokera. Može se postaviti i kasnije pomoću set_callback().

Metode#

set_callback(f: callable) → None#

Postavlja povratni poziv koji pozivaju wait_msg() i check_msg() kada se primi PUBLISH poruka. Povratni poziv se poziva kao f(topic, msg) s oba argumenta tipa bytes.

set_last_will(topic: bytes | str, msg: bytes | str, retain: bool = False, qos: int = 0) → None#

Konfigurira MQTT posljednju volju i oporuku (Last Will and Testament). Broker objavljuje msg na topic ako se klijent neispravno odspoji. Mora se pozvati prije connect().

Argumenti:

  • topic – tema posljednje volje (ne smije biti prazna).

  • msg – sadržaj posljednje volje.

  • retain – ako je True, broker pohranjuje poruku posljednje volje kao zadržanu poruku.

  • qos – razina QoS posljednje volje. Mora biti 0, 1 ili 2.

connect(clean_session: bool = True, timeout: float = 5.0) → int#

Otvara TCP (i opcijski TLS) vezu s brokerom i šalje CONNECT paket.

Argumenti:

  • clean_session – ako je True, traži čistu sesiju; u suprotnom broker nastavlja bilo koje prethodno stanje sesije.

  • timeout – istek vremena soketa u sekundama primijenjen na temeljni soket.

Vraća brokerovu zastavicu session present (0 ili 1). Podiže MQTTException ako broker vrati povratni kod CONNACK različit od nule.

disconnect() → None#

Šalje DISCONNECT paket i zatvara temeljni soket.

ping() → None#

Šalje PINGREQ paket brokeru. Treba ga pozivati periodički ako je keepalive različit od nule kako broker ne bi prekinuo vezu.

publish(topic: bytes | str, msg: bytes | str, retain: bool = False, qos: int = 0) → None#

Objavljuje msg na topic.

Argumenti:

  • topic – naziv teme na koju se objavljuje.

  • msg – bajtovi sadržaja.

  • retain – ako je True, upućuje broker da zadrži poruku za nove pretplatnike.

  • qos – razina kvalitete usluge (quality-of-service). Podržani su 0 (pošalji i zaboravi) i 1 (potvrđeno). 2 nije implementiran i podići će AssertionError.

Za qos 1, poziv se blokira dok se ne primi odgovarajući PUBACK. Ukupna veličina paketa mora biti manja od 2097152 bajtova.

subscribe(topic: bytes | str, qos: int = 0) → None#

Pretplaćuje se na topic na zadanoj razini qos. Povratni poziv mora biti registriran putem set_callback() ili proslijeđen konstruktoru; u suprotnom se podiže AssertionError.

Blokira dok se ne primi odgovarajući SUBACK. Podiže MQTTException ako broker odbije pretplatu (povratni kod 0x80).

wait_msg() → int | None#

Blokira čekajući jedan dolazni MQTT paket i obrađuje ga. PUBLISH paketi se isporučuju registriranom povratnom pozivu. PINGRESP paketi se tiho troše. Za ostale upravljačke pakete vraća se sirovi prvi bajt. Vraća None ako nema dostupnih podataka ili ako je obrađen PINGRESP.

check_msg() → int | None#

Neblokirajuća varijanta metode wait_msg(). Provjerava soket najviše ~50 ms pomoću select.select(); ako podaci čekaju, izvodi istu obradu kao wait_msg(), u suprotnom vraća None.