mqtt --- MQTT client đơn giản

Module mqtt cung cấp một cài đặt MQTT v3.1.1 client tối giản phù hợp cho các thiết bị bị giới hạn bộ nhớ. Nó hỗ trợ kết nối đến broker (có tùy chọn TLS), xuất bản thông điệp, đăng ký chủ đề, cấu hình last-will và ping giữ kết nối.

Ví dụ:

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

Ngoại lệ

exception mqtt.MQTTException

Được phát sinh khi broker từ chối yêu cầu CONNECT hoặc khi một yêu cầu SUBSCRIBE bị từ chối. Tham số duy nhất là mã trả về dạng số do broker cung cấp.

Các lớp

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)

Khởi tạo một MQTT client.

Tham số:

  • client_id -- định danh client duy nhất gửi đến broker.

  • server -- tên máy chủ hoặc địa chỉ IP của broker (được phân giải bằng socket.getaddrinfo()).

  • port -- cổng TCP của broker (thường là 1883 cho kết nối thông thường hoặc 8883 cho TLS).

  • ssl_params -- nếu không phải None, socket được bọc bằng ssl.wrap_socket() và ssl_params được chuyển tiếp dưới dạng tham số từ khóa. Truyền {} để bật TLS với các cài đặt mặc định.

  • user -- tên người dùng tùy chọn để xác thực với broker. Nếu được cung cấp, password cũng phải được cung cấp.

  • password -- mật khẩu tùy chọn dùng cùng với user.

  • keepalive -- khoảng thời gian giữ kết nối tính bằng giây (0 để vô hiệu hóa). Phải nhỏ hơn 65536.

  • callback -- callable được gọi dưới dạng callback(topic, msg) cho mỗi PUBLISH nhận được từ broker. Cũng có thể được đặt sau đó bằng set_callback().

Phương thức

set_callback(f: callable) → None

Đặt hàm gọi lại được gọi bởi wait_msg() và check_msg() khi nhận được thông điệp PUBLISH. Hàm gọi lại được gọi dưới dạng f(topic, msg) với cả hai tham số là bytes.

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

Cấu hình MQTT Last Will và Testament. Broker sẽ xuất bản msg trên topic nếu client ngắt kết nối không đúng cách. Phải được gọi trước connect().

Tham số:

  • topic -- chủ đề last-will (phải không rỗng).

  • msg -- nội dung last-will.

  • retain -- nếu là True, broker lưu trữ thông điệp will dưới dạng thông điệp retained.

  • qos -- mức QoS của last-will. Phải là 0, 1 hoặc 2.

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

Mở kết nối TCP (và tùy chọn TLS) đến broker và gửi gói CONNECT.

Tham số:

  • clean_session -- nếu là True, yêu cầu phiên làm việc sạch; ngược lại broker tiếp tục bất kỳ trạng thái phiên trước đó.

  • timeout -- timeout của socket tính bằng giây áp dụng cho socket bên dưới.

Trả về cờ session present của broker (0 hoặc 1). Phát sinh MQTTException nếu broker trả về mã trả về CONNACK khác không.

disconnect() → None

Gửi gói DISCONNECT và đóng socket bên dưới.

ping() → None

Gửi gói PINGREQ đến broker. Nên được gọi định kỳ nếu keepalive khác không để broker không ngắt kết nối.

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

Xuất bản msg lên topic.

Tham số:

  • topic -- tên chủ đề để xuất bản.

  • msg -- các byte nội dung.

  • retain -- nếu là True, hướng dẫn broker lưu giữ thông điệp cho những người đăng ký mới.

  • qos -- mức chất lượng dịch vụ. 0 (gửi và quên) và 1 (có xác nhận) được hỗ trợ. 2 chưa được cài đặt và sẽ phát sinh AssertionError.

Với qos 1, lệnh gọi bị chặn cho đến khi nhận được PUBACK phù hợp. Tổng kích thước gói phải nhỏ hơn 2097152 byte.

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

Đăng ký topic ở mức qos đã cho. Một hàm gọi lại phải được đăng ký qua set_callback() hoặc cung cấp cho hàm khởi tạo; nếu không sẽ phát sinh AssertionError.

Bị chặn cho đến khi nhận được SUBACK phù hợp. Phát sinh MQTTException nếu broker từ chối đăng ký (mã trả về 0x80).

wait_msg() → int | None

Chặn để chờ một gói MQTT đến và xử lý nó. Các gói PUBLISH được gửi đến hàm gọi lại đã đăng ký. Các gói PINGRESP được tiêu thụ silently. Với các gói điều khiển khác, byte đầu tiên thô được trả về. Trả về None nếu không có dữ liệu hoặc nếu PINGRESP đã được xử lý.

check_msg() → int | None

Biến thể không chặn của wait_msg(). Thăm dò socket trong tối đa ~50 ms bằng select.select(); nếu có dữ liệu đang chờ, nó thực hiện xử lý giống như wait_msg(), ngược lại trả về None.