In this article, I will show you how to use MQTT to publish/subscribe using Python code.
1. Set up your MQTT server
2. Install MQTT client library for Python
pip install paho-mqtt
2.1. What is paho-mqtt
There is only a mqtt broker (server) and multiple mqtt clients in a mqtt system.
Now, we already have a server (Mosquitto) running on our computer if you followed the first step. And now what we need is a mqtt client to publish/subscribe messages to/from a certain topic. So your Python program will be a mqtt client.
3. Subscribe Topic
When a client subscribes to a topic, it is basically telling the broker to send messages to it that are sent to the broker on that topic. —paho-mqtt
Most of the examples just give some codes that can only handle messages from one topic, which is not so good. Because most time, we want to handle different messages from different topics separately, e.g., use different functions.
So the code below, not only can handle messages from different topics but also run indefinitely.
import paho.mqtt.client as mqtt # a callback function def on_message_temperature(client, userdata, msg): # Message is an object and the payload property contains the message data which is binary data. # The actual message payload is a binary buffer. # In order to decode this payload you need to know what type of data was sent. # If it is JSON formatted data then you decode it as a string and then decode the JSON string as follows: # decoded_message=str(message.payload.decode("utf-8"))) # msg=json.loads(decoded_message) print('Received a new temperature data ', str(msg.payload.decode('utf-8'))) print('message topic=', msg.topic) print('message qos=', msg.qos) def on_message_humidity(client, userdata, msg): print('Received a new humidity data ', str(msg.payload.decode('utf-8'))) # Give a name to this MQTT client client = mqtt.Client('greenhouse_server') client.message_callback_add('greenhouse/temperature', on_message_temperature) client.message_callback_add('greenhouse/humidity', on_message_humidity) # IP address of your MQTT broker, using ipconfig to look up it client.connect('192.168.1.109', 1883) # 'greenhouse/#' means subscribe all topic under greenhouse client.subscribe('greenhouse/#') client.loop_forever() # stop the loop # client.loop_stop()
3.2. Some functions need to be understood
loop_forever() above blocks the program, It also handles automatic reconnects. When you want to run your program indefinitely, try it.
loop_forever() will make our program run forever, but it will block the program! This means that the other codes below
client.loop_forever will never be executed. So what if we need to display the data on a screen which requires a
while loop. This means we have to process data from the mqtt server and display it on screen at the same time. And now we need
loop_start() starts a new thread, that calls the loop() method at regular intervals for you. It handles re-connects automatically.
import paho.mqtt.client as mqtt import time # a callback function def on_message_temperature(client, userdata, msg): print('Received a new temperature data ', msg.payload.decode('utf-8')) def on_message_humidity(client, userdata, msg): print('Received a new humidity data ', str(msg.payload.decode('utf-8'))) client = mqtt.Client("greenhouse_server_123") client.message_callback_add('greenhouse/temperature', on_message_temperature) client.message_callback_add('greenhouse/humidity', on_message_humidity) client.connect('192.168.1.109', 1883) # start a new thread client.loop_start() client.subscribe("greenhouse/#") while True: time.sleep(6) # do something you like
4. Publish Message
MQTT is a binary based protocol where the control elements are binary bytes and not text strings. Topic names, Client ID, User names and Passwords are encoded as stream of bytes using UTF-8.
import time import paho.mqtt.client as mqtt def on_publish(client, userdata, mid): print("sent a message") mqttClient = mqtt.Client("greenhouse_alarm") mqttClient.on_publish = on_publish mqttClient.connect('192.168.1.101', 1883) # start a new thread mqttClient.loop_start() # Why use msg.encode('utf-8') here # MQTT is a binary based protocol where the control elements are binary bytes and not text strings. # Topic names, Client ID, Usernames and Passwords are encoded as stream of bytes using UTF-8. while True: msg = "hello" info = mqttClient.publish( topic='greenhouse/alarm', payload=msg.encode('utf-8'), qos=0, ) # Because published() is not synchronous, # it returns false while he is not aware of delivery that's why calling wait_for_publish() is mandatory. info.wait_for_publish() print(info.is_published()) time.sleep(3)
And there you have it. Thank you for reading.