9.19. Python에서의 MQTT¶

네트워크 기능이 있는 모든 OpenMV 카메라에 기본 포함된 mqtt 모듈은 MQTT 와이어 프로토콜을 하나의 클래스 mqtt.MQTTClient로 감쌉니다. 이 클래스는 TCP 소켓을 열고, CONNECT 핸드셰이크를 수행하며, 바이트 수준의 패킷을 묶고 풀고, PINGREQ 킵얼라이브를 처리하고, 들어오는 PUBLISH 메시지를 콜백으로 디스패치합니다. 애플리케이션 코드는 connect(), publish(), subscribe(), 그리고 wait_msg() / check_msg()를 호출합니다.

9.19.1. 열다섯 줄짜리 발행자¶

가장 작으면서도 유용한 프로그램은 단 한 번의 발행입니다. 연결하고, 메시지 하나를 발행하고, 연결을 끊습니다:

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는 Eclipse Mosquitto 프로젝트가 운영하는 공개 테스트 브로커입니다. 자격 증명 없이 포트 1883에서 일반 TCP 연결을 받습니다. 진지한 용도로는 사용하지 마십시오. 프라이버시 보장이 전혀 없고, 토픽 네임스페이스가 인터넷상의 다른 모든 테스터와 공유됩니다.

client_id는 브로커 연결마다 고유해야 합니다. 브로커가 이를 사용해 세션을 추적하기 때문입니다. 토픽과 메시지 페이로드는 바이트입니다. 더 편리하다면 str를 전달해도 되며, 클라이언트가 이를 UTF-8로 인코딩합니다.

9.19.2. TLS를 통한 연결¶

간단한 실험 단계를 넘어선 거의 모든 경우, TLS를 통한 MQTT는 인자 하나만 더 붙이면 됩니다. ssl_params 딕셔너리는 ssl.wrap_socket()으로 전달되므로, 거기서 동작하는 것은 무엇이든 여기서도 동작합니다:

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

포트 8883은 IANA가 예약한 TLS-MQTT 포트입니다. server_hostname은 SNI를 켜서 공유 IP 뒤에 있는 브로커가 올바른 인증서로 라우팅할 수 있게 해 줍니다. HTTPS가 쓰는 것과 같은 메커니즘입니다. user / password는 CONNECT 패킷의 사용자 이름/비밀번호 필드에 매핑되며, 그 자격 증명이 특정 토픽에 대한 발행 또는 구독 권한을 부여할지는 브로커가 결정합니다.

9.19.3. 구독과 수신¶

메시지를 받으려면 클라이언트는 콜백을 제공하고 subscribe()를 호출합니다. 콜백은 두 개의 바이트 인자, 즉 토픽과 페이로드를 받습니다:

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()는 하나의 MQTT 패킷이 도착할 때까지 블록되고, 이를 파싱한 뒤, 그것이 구독한 토픽에 대한 PUBLISH였다면 콜백을 호출하고 반환합니다. 구독 콜백은 그 호출 안에서 실행됩니다. 백그라운드 스레드는 없습니다.

다른 작업을 계속해야 하는 대화형 카메라 루프의 경우, check_msg()가 같은 로직을 논블로킹 형태로 제공합니다. 이는 50 ms 타임아웃으로 select.select()를 사용하며, 대기 중인 것이 없으면 즉시 반환합니다:

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

9.19.4. 깔끔하게 재연결하기¶

장시간 실행되는 MQTT 클라이언트는 연결 끊김을 반드시 처리해야 합니다. Wi-Fi 연결 끊김, 브로커 재시작, NAT 타임아웃, 또는 단순히 트래픽 없이 keepalive를 초과하는 경우 모두 소켓이 종료됩니다. 번들로 제공되는 클라이언트는 끊김을 감지한 호출에서 OSError(또는 브로커의 반환 코드를 담은 일반 예외)를 발생시키며, 표준 패턴은 재시도 루프입니다:

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)

클라이언트가 연결 시 clean_session=False를 전달하지 않는 한 구독은 재연결을 거쳐 유지되지 않습니다. 따라서 내부의 connect는 발행 루프로 들어가기 전에 모든 subscribe() 호출도 다시 발행해야 합니다.

9.19.5. 라스트 윌 훅¶

상태를 보고하는 카메라는 연결이 예기치 않게 끊겼을 때 브로커가 카메라를 대신해 어떤 메시지를 보낼지 브로커에게 알려야 합니다. 윌은 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)

이제 yard-cam/status를 구독하는 모든 대시보드는 카메라가 연결되는 순간 online을, 브로커가 카메라의 끊김을 감지할 때마다 offline을 보게 됩니다. 보존된 offline 메시지는 브로커에 유지되므로, 10분 뒤에 연결하는 대시보드도 여전히 올바른 현재 상태를 보게 됩니다.

9.19.6. HTTP 대신 MQTT를 선택해야 할 때¶

웹서버 챕터에서는 카메라가 HTTP 서버로 동작하는 경우와, 클라우드 업로드 페이지에서는 고정된 URL로 JPEG를 게시하는 HTTP 클라이언트로 동작하는 경우를 다룹니다. 둘 다 각자의 쓰임새가 있습니다. 대신 MQTT를 꺼내 들기에 적절한 때는 다음과 같습니다:

같은 데이터를 여러 리스너(대시보드, 알림 서비스, 레코더)에게 보내야 하는데, 카메라가 그 목록을 미리 알지 못하는 경우.
카메라를 재시작하지 않고도 리스너가 들어왔다 나갔다 할 수 있어야 하는 경우.
카메라가 구독을 하고 싶은 경우, 즉 컨트롤러로부터 명령을 받고 싶은 경우. 이는 HTTP 클라이언트로는 롱 폴링이나 콜백 URL로 푸시하는 서버 없이는 할 수 없습니다.
연결이 긴 유휴 기간 동안 저렴하게 유지되어야 하는 경우.

HTTP를 그대로 쓰기에 적절한 때는 다음과 같습니다. 카메라 하나, 서버 하나, 그리고 단일 MQTT 토픽에 담기에는 너무 큰 본문을 가진 고정된 요청/응답 패턴인 경우입니다(MQTT를 통한 JPEG 프레임 전송은 동작하기는 하지만 브로커에게 무례한 일입니다. HTTP POST가 자연스러운 선택입니다).

교차 링크: 웹서버 챕터의 클라우드 업로드 페이지는 “카메라 → 클라우드 아카이브”의 HTTP 버전을 보여 줍니다. 같은 문제의 MQTT 버전은 카메라를 아카이브의 URL로부터 분리된 상태로 유지하고, 두 번째 소비자(예를 들어 휴대폰 알림 앱)가 같은 스트림을 끌어다 쓸 수 있게 해 줍니다.