【MQTT/Python】PythonでMQTTのPub/Subをするクラスを実装した

はじめに

MQTTをPython/Javaで使ってみたので投稿しています。
実際にほかのシステムから呼び出すときは、どうやるんだろう、をやってみました
本記事は以下の2にあたります。まだの方は1をまずご覧ください。

1. 【MQTT】コマンドベースでMQTTの導入(前回)
2. 【Python】PythonでMQTTのPub/Subをするクラスを実装した(本記事)
3. 【Java】JavaでMQTTのPub/Subをするクラスを実装した(次回)
4. 【ROS】MQTT通信するノードを実装した(次々回)

動作環境

Python 2.7
Python 3.7

ライブラリ、ブローカーのインストール

これは前回の記事にあるのでまだインストールしていない人のために書いておきます。

ライブラリ、ブローカーのインストール

1. windowsの場合

以下のサイトの、「Binary Installation」⇒「Windows」のところから自身の環境に合わせてインストーラーをダウンロードしてください。
https://mosquitto.org/download/

2. Linuxの場合

以下の2つのコマンドを実行してください。

# Mosquitto(Broker)をインストール
$ sudo apt-get install mosquitto

# Mosquittoクライアントをインストール
$ sudo apt-get install mosquitto-clients

クライアントライブラリのインストール

次回以降に使うPythonとJavaのクライアントライブラリのインストール方法も記載しておきます。

Python

ライブラリpahoのインストール

$ pip install paho-mqtt

PythonでMQTT通信

公式のサンプルがあります。

Publisherは毎回繋げに行けばよいので以下の一行で済みます。連続でやりたかったら3行目をループさせてください。

simplepub.py

import paho.mqtt.publish as publish

publish.single("トピック名", "メッセージ内容", hostname="ホスト名")

simplepub_loop.py

import paho.mqtt.publish as publish
import time
i = 0
while True:
    time.sleep(3)
    i += 1
    print(i)
    publish.single("testTopic2", i, hostname="localhost")

Subscriberはcallback関数を設定して処理を行います。(ここもROS1と同じですね)

simplesub.py

import paho.mqtt.subscribe as subscribe

topics = 'test'

def print_msg(client, userdata, message):
    print("%s : %s" % (message.topic, message.payload))

while True:
    subscribe.callback(print_msg, "test", hostname="localhost")

実行

前回の記事と同じように、ブローカーを起動してからクライアントを起動しましょう。

$ cd C:\Program Files (x86)\mosquitto
$ mosquitto -v

以下のようにそれぞれ実行して、publishしたメッセージがsubscriberの画面に出てくれば成功です。

$ python simple_sub.py

$ python simple_pub.py

また、MQTTのクライアント(paho.mqtt.client)にはデフォルトのメソッドがいくつか用意されていてます。
上記のように一行で済ませるのではなく、それぞれのメソッドを書く実装例もあります。
pythonでMQTT送受信
MQTTライブラリ Paho Python を理解しようとしてみる

MQTT通信用クラス

以上はPub/Subともに単体での実装例でしたが、ROSやOpenRTMなど、別で動作しているシステムから呼び出したかったので、
自作モジュールとして簡単に読み込めるようにクラスにまとめました。

以下がそのクラスになります。
mqtt.Clientを継承しています。参考は公式のGitHubの、client_sub-class.pyになります。

MQTTClient.py

import paho.mqtt.client as mqtt
import paho.mqtt.publish as publish
import time

class MyMQTTClass(mqtt.Client):
    def __init__(self):
        super().__init__()
        self.recieve_data = ""
        self.recieve_time = ""
        self.lasttime     = ""

    def on_connect(self, mqttc, obj, flags, rc):
        print("rc: "+str(rc))

    def on_message(self, mqttc, obj, msg):
        print(msg.topic+" "+str(msg.qos)+" "+str(msg.payload))
        self.recieve_time = time.time()
        self.recieve_data = (msg.payload).decode()

    def run(self, hostname, topic):
        self.connect(hostname, 1883, 60)
        self.subscribe(topic, 0)

        self.loop_start()

        rc = 0

        return rc

    def publish_message(self, host_name, topic, message):
        publish.single(topic, message, hostname=host_name)

    def isNew(self):
        flag = False
        if self.lasttime==self.recieve_time: flag =  False
        else: flag = True
        self.lasttime = self.recieve_time
        return flag

# If you want to use a specific client id, use
# mqttc = MyMQTTClass("client-id")
# but note that the client id must be unique on the broker. Leaving the client
# id parameter empty will generate a random id for you.
mqttc = MyMQTTClass()
rc = mqttc.run("localhost","testTopic1")

print("rc: "+str(rc))

i=0
while(1):
    i+=1
    print(i)
    mqttc.publish_message("localhost", "testTopic2",i)

    if mqttc.isNew(): print(mqttc.recieve_data)

※Python 2系では、super().__init__()の部分はsuper(MyMQTTClient, self).__init()__にしないとエラーが出るので注意してください。

それぞれのメソッドの役割は以下の通りです。

メソッド名	役割
on_connect	ブローカーと接続時に呼び出されるメソッド
on_message	メッセージ受け取り時に呼び出されるメソッド(recieve_dataに受け取ったメッセージを代入)
run	外部から呼び出されるメソッド、subscriberのループをスタートさせる
publish_message	メッセージをpublishするメソッド、上記のsimplepub.pyそのまま
isNew	受け取ったメッセージが新しいものか判断するメソッド

実は親クラスにpublish()メソッドがあるのですが、connect()した際のホストにしかできないのかな、と思ったのと、一つのクラスに収めたかったのでこのような形で実装しました。ほかに方法はありそうです。

使用例

クラスの呼び出しを用いた実装例は以下のようになります。
先ほどのMQTTClient.pyと同じディレクトリにファイルを置いてください。

sample_mqtt_lient.py

import MQTTClient

mqttc = MQTTClient.MyMQTTClass()
mqttc.run("ホスト名","トピック名")

# 1回だけpublish
mqttc.publish_message("ホスト名", "トピック名","メッセージ")

if(mqttc.isNew()):
    print(mqttc.receive_data)

mqttc.run()でSubscribe開始、mqttc.publish_message()で1回だけメッセージをpublishします。
また、recieve_dataにsubscribeした変数がはいります。mqttc.isNew()関数によってsubscribeしたかを判断しているので上記のようにセットで使ってください。(Open RTMと一緒ですね)

mqttc = MQTTClient.MyMQTTClass()
mqttc.run("ホスト名","トピック名")

上の2行は一度実行される場所に、以下の3行はメインループの中にいれるイメージです。

# 1回だけpublish
mqttc.publish_message("ホスト名", "トピック名","メッセージ")

if(mqttc.isNew()):
    print(mqttc.receive_data)

おわりに

PythonでMQTT通信ができるクラスを実装してみました。
ROSで使ってみたところ、うまく使えたので目的は達成できたかと思います。