9.19. MQTT dalam Python

Modul mqtt bawaan di setiap OpenMV cam yang terhubung ke jaringan membungkus protokol kawat MQTT dalam satu kelas, mqtt.MQTTClient. Kelas ini membuka soket TCP, melakukan jabat tangan CONNECT, mengemas dan membuka paket pada tingkat byte, menangani keepalive PINGREQ, dan mengirimkan pesan PUBLISH masuk ke sebuah callback. Kode aplikasi memanggil connect(), publish(), subscribe(), dan wait_msg() / check_msg().

9.19.1. Publisher dalam lima belas baris

Program paling sederhana yang berguna adalah satu publish. Hubungkan, publikasikan satu pesan, putuskan koneksi:

from mqtt import MQTTClient

client = MQTTClient(
    client_id='yard-cam',
    server='test.mosquitto.org',
    port=1883,
)
client.connect()
client.publish(b'yard-cam/motion', b'detected at 14:02', qos=0)
client.disconnect()

test.mosquitto.org adalah broker pengujian publik yang dijalankan oleh proyek Eclipse Mosquitto. Broker ini menerima koneksi plain-TCP pada port 1883 tanpa kredensial. Jangan gunakan untuk hal serius; tidak ada jaminan privasi dan ruang nama topik dibagi dengan semua penguji lain di internet.

client_id harus unik per koneksi broker -- broker menggunakannya untuk melacak sesi. Topik dan payload pesan adalah bytes; teruskan str jika lebih praktis dan klien akan mengodekannya sebagai UTF-8.

9.19.2. Menghubungkan melalui TLS

Untuk apa pun di luar eksperimen cepat, MQTT melalui TLS hanya membutuhkan satu argumen tambahan. Dict ssl_params diteruskan ke ssl.wrap_socket(), sehingga apa pun yang berfungsi di sana juga berfungsi di sini:

import ssl

client = MQTTClient(
    client_id='yard-cam',
    server='broker.example.com',
    port=8883,                          # TLS-MQTT default port
    ssl_params={'server_hostname': 'broker.example.com'},
    user='yard-cam',
    password=load_token(),
)

Port 8883 adalah port TLS-MQTT yang dicadangkan oleh IANA. server_hostname mengaktifkan SNI sehingga broker di balik IP bersama dapat mengarahkan ke sertifikat yang tepat -- mekanisme yang sama dengan yang digunakan HTTPS. user / password dipetakan ke bidang username/password paket CONNECT; broker memutuskan apakah kredensial tersebut memberikan hak publish atau subscribe ke topik tertentu.

9.19.3. Berlangganan dan menerima

Untuk menerima pesan, klien menyediakan callback dan memanggil subscribe(). Callback menerima dua argumen bytes, topik dan payload:

def on_message(topic, msg):
    print('received on', topic.decode(), ':', msg.decode())

client = MQTTClient(
    client_id='dashboard',
    server='test.mosquitto.org',
    port=1883,
    callback=on_message,
)
client.connect()
client.subscribe(b'yard-cam/motion', qos=0)
while True:
    client.wait_msg()

wait_msg() memblokir hingga satu paket MQTT tiba, menguraikannya, memanggil callback jika itu adalah PUBLISH pada topik yang dilanggani, dan kembali. Callback yang dilanggani dipanggil dari dalam panggilan tersebut -- tidak ada thread latar belakang.

Untuk loop kamera interaktif yang perlu terus melakukan pekerjaan lain, check_msg() adalah logika yang sama dalam bentuk non-blocking. Ini menggunakan select.select() dengan timeout 50 ms dan segera kembali jika tidak ada yang tertunda:

while True:
    client.check_msg()
    run_frame()                  # capture + processing
    check_motion_threshold()

9.19.4. Menghubungkan kembali dengan bersih

Setiap klien MQTT yang berjalan lama harus menangani koneksi yang terputus. Putusnya Wi-Fi, restart broker, timeout NAT, atau sekadar melewati keepalive tanpa lalu lintas semuanya mengakhiri soket. Klien bawaan memunculkan OSError (atau eksepsi biasa dengan kode kembalian broker) dari panggilan yang mendeteksi pemutusan, dan pola standarnya adalah loop percobaan ulang:

import time

def keep_publishing(client, topic, get_message):
    while True:
        try:
            client.connect()
            while True:
                client.publish(topic, get_message())
                time.sleep(5)
        except OSError:
            print('connection lost, reconnecting in 5s')
            time.sleep(5)

Langganan tidak dipertahankan di seluruh koneksi ulang kecuali klien melewatkan clean_session=False saat connect, sehingga connect di dalam loop juga harus mengeluarkan kembali panggilan subscribe() apa pun sebelum masuk ke loop publish.

9.19.5. Hook last-will

Kamera yang melaporkan status harus memberi tahu broker pesan apa yang harus dikirim atas nama kamera jika koneksi mati secara tak terduga. Atur will sebelum connect()

client = MQTTClient(
    client_id='yard-cam',
    server='broker.example.com',
    port=8883,
    ssl_params={'server_hostname': 'broker.example.com'},
)
client.set_last_will(
    b'yard-cam/status',
    b'offline',
    retain=True,
    qos=0,
)
client.connect()
client.publish(b'yard-cam/status', b'online', retain=True)

Sekarang setiap dashboard yang berlangganan ke yard-cam/status melihat online pada saat kamera terhubung dan offline kapan pun broker mendeteksi kamera terputus. Pesan offline yang dipertahankan tetap ada di broker sehingga dashboard yang terhubung sepuluh menit kemudian masih melihat status terkini yang benar.

9.19.6. Kapan memilih MQTT daripada HTTP

Bab webserver mencakup kamera yang bertindak sebagai server HTTP dan, di halaman cloud-upload, sebagai klien HTTP yang memposting JPEG ke URL tetap. Keduanya memiliki tempatnya masing-masing. Waktu yang tepat untuk memilih MQTT sebagai gantinya:

  • Data yang sama perlu dikirim ke beberapa pendengar (dashboard, layanan notifikasi, perekam) tanpa kamera mengetahui daftarnya terlebih dahulu.

  • Pendengar dapat datang dan pergi tanpa kamera perlu di-restart.

  • Kamera ingin berlangganan -- untuk menerima perintah dari pengontrol -- yang tidak dapat dilakukan klien HTTP tanpa long polling atau server yang mendorong ke URL callback.

  • Koneksi harus bertahan dalam periode idle yang panjang dengan biaya rendah.

Waktu yang tepat untuk tetap menggunakan HTTP: satu kamera, satu server, pola permintaan/respons tetap dengan body yang terlalu besar untuk satu topik MQTT (bingkai JPEG melalui MQTT berfungsi tetapi memberatkan broker; HTTP POST adalah pilihan yang lebih alami).

Tautan silang: halaman cloud-upload di bab webserver menunjukkan versi HTTP dari "kamera → arsip cloud". Versi MQTT dari masalah yang sama membuat kamera tetap terpisah dari URL arsip dan memungkinkan konsumen kedua (misalnya aplikasi notifikasi ponsel) memanfaatkan aliran yang sama.