mqtt --- عميل MQTT بسيط

توفّر الوحدة mqtt تنفيذًا أدنى لعميل MQTT v3.1.1 مناسبًا للاستخدام على الأجهزة محدودة الذاكرة. وهي تدعم الاتصال بوسيط (مع TLS اختياري)، ونشر الرسائل، والاشتراك في المواضيع، وتهيئة الوصية الأخيرة، ونبضات الإبقاء على الاتصال.

مثال:

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

الاستثناءات

exception mqtt.MQTTException

يُطلق عندما يرفض الوسيط طلب CONNECT أو عندما يُرفض طلب SUBSCRIBE. الوسيط الوحيد هو رمز الإرجاع العددي الذي يوفّره الوسيط.

الفئات

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)

ينشئ عميل MQTT.

الوسائط:

  • client_id -- معرّف العميل الفريد المُرسَل إلى الوسيط.

  • server -- اسم مضيف الوسيط أو عنوان IP (يُحلّ باستخدام socket.getaddrinfo()).

  • port -- منفذ TCP الخاص بالوسيط (عادةً 1883 للنص الصريح أو 8883 لـ TLS).

  • ssl_params -- إذا لم يكن None، يُغلَّف المقبس باستخدام ssl.wrap_socket() ويُمرَّر ssl_params كوسائط مفتاحية. مرّر {} لتمكين TLS بالإعدادات الافتراضية.

  • user -- اسم مستخدم اختياري لمصادقة الوسيط. إذا أُعطي، فيجب توفير password أيضًا.

  • password -- كلمة مرور اختيارية تُستخدم مع user.

  • keepalive -- فترة الإبقاء على الاتصال بالثواني (0 يعطّلها). يجب أن تكون أقل من 65536.

  • callback -- قابل للاستدعاء يُستدعى بالصيغة callback(topic, msg) لكل رسالة PUBLISH تُسلَّم من الوسيط. يمكن أيضًا ضبطه لاحقًا باستخدام set_callback().

الدوال

set_callback(f: callable) → None

تضبط دالة رد النداء التي يستدعيها wait_msg() و check_msg() عند استقبال رسالة PUBLISH. تُستدعى دالة رد النداء بالصيغة f(topic, msg) مع كلا الوسيطين من النوع bytes.

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

تهيئ وصية MQTT الأخيرة (Last Will and Testament). ينشر الوسيط msg على topic إذا انقطع اتصال العميل بشكل غير سليم. يجب استدعاؤها قبل connect().

الوسائط:

  • topic -- موضوع الوصية الأخيرة (يجب ألا يكون فارغًا).

  • msg -- حمولة الوصية الأخيرة.

  • retain -- إذا كان True، يخزّن الوسيط رسالة الوصية كرسالة محتفظ بها.

  • qos -- مستوى QoS للوصية الأخيرة. يجب أن يكون 0 أو 1 أو 2.

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

يفتح اتصال TCP (و TLS اختياريًا) بالوسيط ويرسل حزمة CONNECT.

الوسائط:

  • clean_session -- إذا كان True، يطلب جلسة نظيفة؛ وإلا يستأنف الوسيط أي حالة جلسة سابقة.

  • timeout -- مهلة المقبس بالثواني المطبَّقة على المقبس الأساسي.

يُرجع علم session present (وجود الجلسة) الخاص بالوسيط (0 أو 1). يُطلق MQTTException إذا أرجع الوسيط رمز إرجاع CONNACK غير صفري.

disconnect() → None

يرسل حزمة DISCONNECT ويغلق المقبس الأساسي.

ping() → None

يرسل حزمة PINGREQ إلى الوسيط. ينبغي استدعاؤها دوريًا إذا كان keepalive غير صفري حتى لا يقطع الوسيط الاتصال.

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

ينشر msg على topic.

الوسائط:

  • topic -- اسم الموضوع المراد النشر عليه.

  • msg -- بايتات الحمولة.

  • retain -- إذا كان True، يوجّه الوسيط للاحتفاظ بالرسالة للمشتركين الجدد.

  • qos -- مستوى جودة الخدمة. المستويان 0 (أطلق وانسَ) و 1 (مؤكَّد) مدعومان. المستوى 2 غير منفَّذ وسيُطلق AssertionError.

بالنسبة إلى qos 1، يحجب الاستدعاء حتى يُستقبل PUBACK المطابق. يجب أن يكون الحجم الكلي للحزمة أقل من 2097152 بايت.

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

يشترك في topic عند مستوى qos المعطى. يجب أن تكون دالة رد نداء قد سُجِّلت عبر set_callback() أو مُرِّرت إلى المُنشئ؛ وإلا يُطلق AssertionError.

يحجب حتى يُستقبل SUBACK المطابق. يُطلق MQTTException إذا رفض الوسيط الاشتراك (رمز الإرجاع 0x80).

wait_msg() → int | None

يحجب منتظرًا حزمة MQTT واردة مفردة ويعالجها. تُسلَّم حزم PUBLISH إلى دالة رد النداء المسجَّلة. تُستهلك حزم PINGRESP بصمت. أما حزم التحكم الأخرى فيُرجَع أول بايت خام منها. يُرجع None إذا لم تتوفر بيانات أو إذا عُولجت PINGRESP.

check_msg() → int | None

متغير غير حاجب من wait_msg(). يستقصي المقبس لمدة ~50 مللي ثانية على الأكثر باستخدام select.select()؛ إذا كانت هناك بيانات معلقة فإنه يجري نفس المعالجة التي يجريها wait_msg()، وإلا يُرجع None.