■Micropythonを使い、MQTTでサブスクライブする手順についてのネット情報
【狙い】
発信元であるRPP2WがMQTTでメッセージを送った際に、メッセージが受信されて
DB等に正常に登録されたことを「発信元が知る」方法を理解する。
【出典】
ESP32 などのマイコンで MicroPython のモジュール umqtt を使い、MQTT サーバーへの接続、
サブスクライブ、メッセージの送受信などの機能を実現する方法
https://docs.emqx.com/ja/cloud/latest/connect_to_deployments/esp32_with_micropython.html
【要点１】
出典はEMQXにおけるMQTTの例だが、ライブラリが、
from umqtt.simple import MQTTClient
なので、サブスクライブ手順はほぼそのまま使える。
【要点２】
１：ネットワーク接続先を定義
２：ネットワーク接続
３：MQTTブローカー接続
４：新しいメッセージを受信したときの処理を関数定義（以降説明のon_messageの部分）
すなわち、受信メッセージを解読し、適切に分岐して処理をするのはここ！
５：コールバック関数に受信処理関数を組み込む
client.set_callback(on_message)
６：サブスクトピック（例：topic_msg_sub）を指定してMQTTサブスクライブ設定
例：client.subscribe(topic_msg)
７：待ち受けループ
例１：
先にメッセージをパブリッシュし、同じトピックで待ち受けすると、
送ったメッセージを受信することになる。
例２：
デバイスＰがメッセージをトピックtopoc_msg_pubでパブリッシュし、
デバイスＳがトピックtopoc_msg_pubでサブスクしてメッセージを解読して所定処理を行い、
その結果をメッセージとしてデバイスＳがトピックtopoc_msg_subでパブリッシュし、
デバイスＰがトピックtopoc_msg_subでサブスクしてデバイスＳでの処理結果をメッセージで得る
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
●（出典から）コールバック設定とトピックをサブスクライブ
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
次に、メッセージ到着時に呼ばれる on_message コールバック関数を実装します。ここにメッセージ処理のコードを記述できます。この例では、トピックとメッセージの内容を単純に表示しています。メッセージの取りこぼしを防ぐため、トピックをサブスクライブする前にクライアントのコールバックを on_message に設定することを推奨します。
def on_message(topic, msg):
print("Received '{payload}' from topic '{topic}'\n".format(
payload = msg.decode(), topic = topic.decode()))
def subscribe(client):
client.set_callback(on_message)
client.subscribe(TOPIC)
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
●（出典から）循環的にパブリッシュと受信
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
この例では、同じクライアントを使ってメッセージの循環的なパブリッシュと受信を行います。以下のコードの主な役割は、新しいメッセージ内容を継続的に構築してパブリッシュし、その後 wait_msg() を呼び出して MQTT サーバーから転送されるメッセージをブロックして待つことです。
メッセージが届くと on_message コールバックが呼ばれてメッセージ内容を表示し、コールバック終了後は 1 秒待ってから次のループに入ります。
def loop_publish(client):
msg_count = 0
while True:
msg_dict = {
'msg': msg_count
}
msg = json.dumps(msg_dict)
result = client.publish(TOPIC, msg)
print("Send '{msg}' to topic '{topic}'".format(msg = msg, topic = TOPIC))
client.wait_msg()
msg_count += 1
time.sleep(1)
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
●（出典から）メイン関数
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
メイン関数では、まず wifi.connect() を呼んでデバイスをネットワークに接続し、その後に接続、
サブスクライブ、循環パブリッシュの各関数を順に呼び出します。
def run():
wifi.connect()
client = connect()
subscribe(client)
loop_publish(client)
if __name__ == "__main__":
run()
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
●（別の出典から）KeepAlive処理
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
一般的にKeepAliveは２〜３分（180秒）程度だそうです。
Pingreq/Pingrespによる接続維持、切断時の再接続処理をしておくのが良さそうです。
出典：
↑この出典にメインループ内で接続維持と再接続の実コード例があります。
MQTTブローカーはクライアントからメッセージあるいはPingを受けると
KeepAliveカウンタを0にして、接続維持に至るそうです。
↓以下、受信専用のデバイスでのメインループ内での処理コード例
（送受信とも動作させるデバイスの場合はメインループで受信無限待受では不都合なので、
このままのコードは適用できない。）
# --- メインループ ---
connect_wifi() # まずWi-Fiに接続
while True:
try:
client = mqtt_connect() # MQTTサーバーに接続
while True:
client.check_msg() # 新着メッセージを確認（受信があればon_messageが発火）
time.sleep(1) # 1秒ごとにポーリング
# 10秒ごとにpingを送信し、接続を維持
if time.time() % 10 == 0:
client.ping()
except Exception as e:
# 接続が切れた場合、5秒後に再接続を試みる
print(f"Connection lost ({e}). Retrying in 5 seconds...")
led.on(); time.sleep(0.2); led.off() # エラー時はLEDを短く点滅
time.sleep(5)
connect_wifi() # Wi-Fiから再接続し直す
# ループ先頭に戻り、MQTTも再接続
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
●実際のコード例（自作コード、開発環境、再接続関連は未設定）
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
# Status indicater
# ------ kabayon 260711 -------
# micropython 1.27.0 or later
# RaspberryPiW/2W
# MQTT network for publishing
# Programable RGB-LED WS2811 8X8-allay
#標準ライブラリーーーーーーーーーーーーーーーーーーーーー
import sys
import machine
from machine import Pin
#NeoPixelライブラリ
from neopixel import NeoPixel
import time
#wifi-networkライブラリ（PicoW/2Wファームウェアを使う）
import network
#外部ライブラリーーーーーーーーーーーーーーーーーーーーー
#mqttライブラリ
from umqtt.simple import MQTTClient
#初期設定ーーーーーーーーーーーーーーーーーーーーーーーー
# 8X8NeoPixelマトリクス設定ーーーーーーーーーーーー
num_row_matrix = 8 #8行
num_column_matrix = 8 #8列
num_row_zone = 2 #2行1ゾーンを同色にするため
total_zone = 4 #ゾーン数4
pin_np = Pin(28, Pin.OUT) # NeoPixel 駆動のための GPIO 28 を出力に設定
np = NeoPixel(pin_np,num_row_matrix*num_column_matrix )
#標準的な色のRGB設定配列
#勝手に決めた色番号,R,G,B）。RGBは最大255
#255は明るすぎるので、実際は最大16程度で十分。
#RGBそれぞれ光度が異なるので色味を観ながら調整
colour_neopixel = [
[0,8,8,8,"white"],
[1,0,0,0,"black"],
[2,8,0,0,"red"],
[3,0,0,16,"blue"],
[4,8,8,0,"yellow"],
[5,0,8,0,"green"],
[6,8,3,0,"orange"],
[7,8,2,6,"pink"]
]
#64個のLEDそれぞれ色を決められるが、ここでは4ゾーン（64個）まるごと同色にセット
num_colour_zone_all_white = [0,0,0,0]
num_colour_zone_all_black = [1,1,1,1]
num_colour_zone_all_red = [2,2,2,2]
num_colour_zone_all_blue = [3,3,3,3]
num_colour_zone_all_yellow = [4,4,4,4]
num_colour_zone_all_green = [5,5,5,5]
num_colour_zone_all_orange = [6,6,6,6]
num_colour_zone_all_pink = [7,7,7,7]
#ゾーンごとに色をセットして発光させる自作関数
def show_np(num_colour_zone,total_zone):
for num_zone in range (0,total_zone):
num_colour = num_colour_zone[num_zone]
total_pixcel_zone = num_row_matrix*num_row_zone
start_pixel = num_zone*total_pixcel_zone
end_pixel = start_pixel + total_pixcel_zone
for num_pixel in range (start_pixel,end_pixel):
np[num_pixel] = (colour_neopixel[num_colour][1],colour_neopixel[num_colour][2],colour_neopixel[num_colour][3])
np.write()
time.sleep(0.2)
#とりあえず４ゾーンとも消灯
show_np(num_colour_zone_all_black,total_zone)
#ネットワーク接続ーーーーーーーーーーーーーーーーーーーーーー
# Wi-Fi接続関数
def connect():
# 自分の固定IPアドレス設定
IPaddress_mine = '192.168.10.81'
# Wifi起動
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
# Wifi接続
wlan.connect(ssid, password)
while wlan.isconnected() == False:
print('Waiting for connection...')
time.sleep(1)
wlan_status = wlan.ifconfig()
wlan.ifconfig((IPaddress_mine, wlan_status[1], wlan_status[2], wlan_status[3]))
ip = wlan.ifconfig()[0]
print(f'Connected on {ip}')
return ip
# Wi-Fi接続SSID、パスワード設定
ssid = 'kabanet24'
password = 'kabanet24'
# MQTTブローカーサーバーIP設定
address_kaban5105 = '192.168.10.24'
mqttBroker = address_kaban5105
# ConnectID（ユニーク）設定
myId = 'rpp01'
# トピック設定（バイト列で表記する。/で構造化してある）
topic_mqtt_pub= b"test_mqtt/dpt201"
topic_mqtt_sub= b"test_mqtt/dpt201"
# 接続維持時間設定
time_keepalive_MQTTClient = 36000
# MQTTクライアント設定
client = MQTTClient(myId, mqttBroker, keepalive=time_keepalive_MQTTClient)
# 先にネットワーク接続し、次にMQTTブローカーに接続ーーーーーーー
# Wi-Fi接続
connect()
# ブローカーに接続
try:
# ブローカー接続処理
client.connect()
# Subトピック登録
#client.subscribe(topic_mqtt)
except:
# ブローカー接続失敗時、プログラム終了
msg_system = "mqtt NG!"
msg_error = "DPT down"
show_np(num_colour_zone_all_red,total_zone)
print(msg_system,msg_error)
sys.exit()
msg_system = "mqtt OK!"
show_np(num_colour_zone_all_green,total_zone)
print(msg_system)
# サブスクライブ設定ーーーーーーーーーーーーーーー
#まず先にメッセージ受信時の処理内容の関数を定義
def on_message(topic, msg):
#print((topic, msg))
#print(topic.decode(), msg.decode()) # 文字列として使いたい場合
#受信文字列を変数に入れる
msg_mqtt_sub = msg.decode()
#print(msg_mqtt_sub)
#受信文字列変数（数値文字の0〜7）に応じた色で8X8NeoPixelマトリクス発光
if msg_mqtt_sub == str(0):
show_np(num_colour_zone_all_white,total_zone)
elif msg_mqtt_sub == str(1):
show_np(num_colour_zone_all_black,total_zone)
elif msg_mqtt_sub == str(2):
show_np(num_colour_zone_all_red,total_zone)
elif msg_mqtt_sub == str(3):
show_np(num_colour_zone_all_blue,total_zone)
elif msg_mqtt_sub == str(4):
show_np(num_colour_zone_all_yellow,total_zone)
elif msg_mqtt_sub == str(5):
show_np(num_colour_zone_all_green,total_zone)
elif msg_mqtt_sub == str(6):
show_np(num_colour_zone_all_orange,total_zone)
elif msg_mqtt_sub == str(7):
show_np(num_colour_zone_all_pink,total_zone)
#コールバック関数のなかに受信時処理関数を設定
client.set_callback(on_message)
#受信用のトピックを使ってサブスクライブ設定
client.subscribe(topic_mqtt_sub)
#送信・受信ループーーーーーーーーーーーーーーーーーーー
i_count = 0
while True:
# Pubの前になんらかの処理を行い、それの結果として送信メッセージを作成し、Pub実行
# ここではカウンタ値に対して7の剰余を色番号として送信メッセージとする。
i_count = i_count +1
msg_mqtt_pub = str(i_count % 7)
# Pub実行
client.publish(topic_mqtt_pub, msg_mqtt_pub)
time.sleep(1) # デバイスが少なく、かつ無線AP近傍であれば0.1秒でも動作する
# Subメッセージrecieve
client.wait_msg() # 受信時の処理はon_messeage関数に記述されており、ここは受信待ちだけでよい。
time.sleep(0.1)