9.19. MQTT em Python

O módulo mqtt incluído em toda OpenMV cam com rede encapsula o protocolo de fio MQTT em uma única classe, mqtt.MQTTClient. A classe abre o socket TCP, executa o handshake CONNECT, empacota e desempacota os pacotes no nível de bytes, trata o keepalive PINGREQ e despacha as mensagens PUBLISH recebidas para um callback. O código da aplicação chama connect(), publish(), subscribe() e wait_msg() / check_msg().

9.19.1. Um publicador em quinze linhas

O menor programa útil é uma única publicação. Conectar, publicar uma mensagem e desconectar:

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 é o broker público de teste mantido pelo projeto Eclipse Mosquitto. Ele aceita conexões em TCP puro na porta 1883 sem credenciais. Não o use para nada sério; ele não oferece garantias de privacidade e o namespace de tópicos é compartilhado com todos os outros testadores da internet.

client_id deve ser único por conexão com o broker – o broker o utiliza para rastrear sessões. Os tópicos e as cargas úteis das mensagens são bytes; passe str se for mais conveniente, e o cliente o codificará como UTF-8.

9.19.2. Conectando via TLS

Para qualquer coisa além de experimentos rápidos, MQTT sobre TLS é um argumento extra. O dicionário ssl_params é encaminhado para ssl.wrap_socket(), então tudo o que funciona lá funciona aqui:

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

A porta 8883 é a porta TLS-MQTT reservada pela IANA. server_hostname ativa o SNI para que brokers atrás de um IP compartilhado possam rotear para o certificado correto – o mesmo mecanismo que o HTTPS usa. user / password mapeiam para os campos de nome de usuário/senha do pacote CONNECT; o broker decide se essas credenciais concedem direitos de publicação ou de assinatura a tópicos específicos.

9.19.3. Assinando e recebendo

Para receber mensagens, um cliente fornece um callback e chama subscribe(). O callback recebe dois argumentos do tipo bytes, o tópico e a carga útil:

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() bloqueia até que um pacote MQTT chegue, faz seu parsing, chama o callback se for um PUBLISH em um tópico assinado e retorna. Os callbacks das assinaturas são disparados de dentro dessa chamada – não há thread em segundo plano.

Para um loop interativo da câmera que precisa continuar fazendo outro trabalho, check_msg() é a mesma lógica em forma não bloqueante. Ele usa select.select() com um timeout de 50 ms e retorna imediatamente se não houver nada pendente:

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

9.19.4. Reconectando de forma limpa

Qualquer cliente MQTT de longa duração precisa lidar com conexões interrompidas. Desconexões de Wi-Fi, reinicializações do broker, timeouts de NAT ou simplesmente ultrapassar o keepalive sem tráfego encerram o socket. O cliente incluído lança OSError (ou uma exceção simples com o código de retorno do broker) a partir da chamada que percebeu a queda, e o padrão usual é um loop de retentativa:

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)

As assinaturas não são preservadas entre reconexões, a menos que o cliente passe clean_session=False na conexão, então o connect interno também deve reemitir quaisquer chamadas subscribe() antes de entrar no loop de publicação.

9.19.5. O hook de última vontade (last will)

Uma câmera que reporta status deve dizer ao broker qual mensagem enviar em nome da câmera caso a conexão caia inesperadamente. Defina a vontade (will) antes de 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)

Agora qualquer dashboard assinado em yard-cam/status vê online no momento em que a câmera se conecta e offline sempre que o broker notar que a câmera caiu. A mensagem offline retida persiste no broker, de modo que um dashboard que se conecta dez minutos depois ainda vê o estado atual correto.

9.19.6. Quando escolher MQTT em vez de HTTP

O capítulo sobre servidores web aborda a câmera atuando como um servidor HTTP e, na página de upload para a nuvem, como um cliente HTTP enviando JPEGs (POST) para uma URL fixa. Ambos têm seu lugar. O momento certo de recorrer ao MQTT em vez disso:

  • Os mesmos dados precisam ir para vários ouvintes (um dashboard, um serviço de notificação, um gravador) sem que a câmera conheça a lista de antemão.

  • Os ouvintes podem entrar e sair sem que a câmera seja reiniciada.

  • A câmera quer assinar – receber comandos de um controlador – algo que um cliente HTTP não consegue fazer sem long polling ou sem um servidor empurrando dados em uma URL de callback.

  • A conexão precisa sobreviver a longos períodos ociosos de forma barata.

O momento certo de continuar com HTTP: uma câmera, um servidor, um padrão fixo de requisição/resposta com um corpo grande demais para um único tópico MQTT (quadros JPEG sobre MQTT funcionam, mas são deselegantes com o broker; o HTTP POST é o ajuste natural).

Referência cruzada: a página de upload para a nuvem no capítulo sobre servidores web mostra a versão HTTP de “câmera → arquivo na nuvem”. A versão MQTT do mesmo problema mantém a câmera desacoplada da URL do arquivo e permite que um segundo consumidor (um aplicativo de alertas no celular, digamos) acesse o mesmo fluxo.