mqtt — Client MQTT simplu

Modulul mqtt oferă o implementare minimală de client MQTT v3.1.1 potrivită pentru utilizarea pe dispozitive cu memorie limitată. Acesta acceptă conectarea la un broker (cu TLS opțional), publicarea de mesaje, abonarea la topicuri, configurarea testamentului (last-will) și ping-urile de menținere a conexiunii (keep-alive).

Exemplu:

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

Excepții

exception mqtt.MQTTException

Generată atunci când brokerul respinge cererea CONNECT sau atunci când o cerere SUBSCRIBE este refuzată. Singurul argument este codul numeric de retur furnizat de broker.

Clase

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)

Construiește un client MQTT.

Argumente:

  • client_id – identificator unic de client trimis brokerului.

  • server – numele de gazdă sau adresa IP a brokerului (rezolvat cu socket.getaddrinfo()).

  • port – portul TCP al brokerului (de obicei 1883 simplu sau 8883 pentru TLS).

  • ssl_params – dacă nu este None, socketul este împachetat cu ssl.wrap_socket() iar ssl_params este transmis ca argumente cu nume. Treceți {} pentru a activa TLS cu valorile implicite.

  • user – nume de utilizator opțional pentru autentificarea la broker. Dacă este furnizat, trebuie să fie furnizat și password.

  • password – parolă opțională folosită împreună cu user.

  • keepalive – intervalul de menținere a conexiunii (keep-alive) în secunde (0 dezactivează). Trebuie să fie mai mic decât 65536.

  • callback – apelabil invocat ca callback(topic, msg) pentru fiecare PUBLISH livrat de la broker. Poate fi setat și ulterior cu set_callback().

Metode

set_callback(f: callable) → None

Setează funcția de retroapelare (callback) invocată de wait_msg() și check_msg() atunci când este recepționat un mesaj PUBLISH. Funcția de retroapelare este apelată ca f(topic, msg) cu ambele argumente de tip bytes.

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

Configurează testamentul Last Will and Testament al MQTT. Brokerul publică msg pe topic dacă clientul se deconectează în mod neregulamentar. Trebuie apelată înainte de connect().

Argumente:

  • topic – topicul testamentului (last-will) (trebuie să fie nevid).

  • msg – conținutul util (payload) al testamentului (last-will).

  • retain – dacă este True, brokerul stochează mesajul de testament ca mesaj reținut.

  • qos – nivelul QoS al testamentului (last-will). Trebuie să fie 0, 1 sau 2.

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

Deschide o conexiune TCP (și opțional TLS) către broker și trimite un pachet CONNECT.

Argumente:

  • clean_session – dacă este True, solicită o sesiune curată; în caz contrar brokerul reia orice stare de sesiune anterioară.

  • timeout – timpul de așteptare al socketului în secunde aplicat socketului subiacent.

Returnează indicatorul session present al brokerului (0 sau 1). Generează MQTTException dacă brokerul returnează un cod de retur CONNACK diferit de zero.

disconnect() → None

Trimite un pachet DISCONNECT și închide socketul subiacent.

ping() → None

Trimite un pachet PINGREQ către broker. Ar trebui apelat periodic dacă keepalive este diferit de zero, astfel încât brokerul să nu întrerupă conexiunea.

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

Publică msg pe topic.

Argumente:

  • topic – numele topicului pe care se publică.

  • msg – octeții conținutului util (payload).

  • retain – dacă este True, instruiește brokerul să rețină mesajul pentru noii abonați.

  • qos – nivelul de calitate a serviciului. Sunt acceptate 0 (trimite și uită) și 1 (confirmat). 2 nu este implementat și va genera AssertionError.

Pentru qos 1, apelul se blochează până la recepționarea PUBACK-ului corespunzător. Dimensiunea totală a pachetului trebuie să fie mai mică de 2097152 de octeți.

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

Se abonează la topic la nivelul qos specificat. O funcție de retroapelare (callback) trebuie să fi fost înregistrată prin set_callback() sau furnizată constructorului; în caz contrar se generează AssertionError.

Se blochează până la recepționarea SUBACK-ului corespunzător. Generează MQTTException dacă brokerul refuză abonamentul (codul de retur 0x80).

wait_msg() → int | None

Se blochează în așteptarea unui singur pachet MQTT primit și îl procesează. Pachetele PUBLISH sunt livrate funcției de retroapelare (callback) înregistrate. Pachetele PINGRESP sunt consumate în mod silențios. Pentru alte pachete de control, este returnat primul octet brut. Returnează None dacă nu sunt disponibile date sau dacă a fost procesat un PINGRESP.

check_msg() → int | None

Varianta neblocantă a wait_msg(). Interoghează socketul timp de cel mult ~50 ms folosind select.select(); dacă există date în așteptare efectuează aceeași procesare ca wait_msg(), în caz contrar returnează None.