The paho MQTT python client from Eclipse supports MQTT v 3.1 and 3,1.1, and now MQTTv5 and works with Python 3.x.
Tutorial Outline
In this tutorial we look at the main client object, and it’s methods.
We will then create a simple Python example script that subscribes to a topic and publishes messages on that topic.
If all goes well we should see the published messages.
The example scripts are kept simple, and I don’t include any error checking.
I use my own locally installed broker, but you will probably find it easier when starting to use a free online broker like:
- test.mosquitto.org
- broker.hivemq.com
- iot.eclipse.org
Installing The Client
You can Install the MQTT client using PIP with the command:
It usually isn’t as straightforward as using the command
pip install paho-mqtt
as most machines have multiple versions of python installed and there are several versions of pip and the actual command depends on whether you are on Windows or Linux.
Therefore use the command:
pip --version
before installing to find out where pip will install the files.
The screen shot below is taken from my Windows 10 machine where I have two versions of Python installed (3.4 and 3.6)
If I ran
pip install paho-mqtt
It would install the client in the 3.6 site packages. To install it for the 3.4 version I would need to run.
pip3.4 install paho-mqtt
On my Raspberry pi (linux) using the command
pip install paho-mqtt
would install the client for use my python version 2.7
To install for version 3.5 I would need to run:
pip3 install paho-mqtt
Note: if you have multiple versions of python on your machine then take a look at my Python Notes.
Note: On the PI and maybe other linux versions if you get an error on install then use sudo pip install paho-mqtt.
Video- Installing The Mqtt Python Client and Other Modules Using PIP
You will find the online client documentation here. and also the install files if you need them.
The Python MQTT Client
The core of the client library is the client class which provides all of the functions to publish messages and subscribe to topics.
If you want to look at the code for this class you should find the code in the client.py file in the mqtt directory. (windows machine)
This directory is located in python34\Lib\site-packages\paho\mqtt (windows see Python Notes.)
Where python34 is the root of my python install.
Main Client Methods
The paho mqtt client class has several methods.The main ones are:
- connect() and disconnect()
- subscribe() and unsubscribe()
- publish()
Each of these methods is associated with a callback. See Later.
Importing The Client Class
To use the client class you need to import it. Use the following:
Import paho.mqtt.client as mqtt
Creating a Client Instance
The client constructor takes 4 optional parameters, as shown below .but only the client_id is necessary, and should be unique.
Client(client_id=””, clean_session=True, userdata=None, protocol=MQTTv311, transport=”tcp”)
To create a instance use:
client =mqtt.Client(client_name)
See Working with Client objects for more details
Connecting To a Broker or Server
Before you can publish messages or subscribe to topics you need to establish a connection to a broker.
To do this use the connect method of the Python mqtt client.
The method can be called with 4 parameters. The connect method declaration is shown below with the default parameters.
connect(host, port=1883, keepalive=60, bind_address="")
Note: You only need to supply the broker name/IP address.
The general syntax is
client.connect(host_name)
See Working with Client Connections for more details.
Publishing Messages
Once you have a connection you can start to publish messages.
To do this we use the publish method.
The publish method accepts 4 parameters. The parameters are shown below with their default values.
publish(topic, payload=None, qos=0, retain=False)
The only parameters you must supply are the topic, and the payload.
The payload is the message you want to publish.
The general syntax is:
client.publish("house/light","ON")
Example Python Script:
We are now in a position to create our first Python Script to Publish a message.
The script below publishes the message OFF to topic house/main-light
import paho.mqtt.client as mqtt #import the client1
broker_address="192.168.1.184"
#broker_address="iot.eclipse.org" #use external broker
client = mqtt.Client("P1") #create new instance
client.connect(broker_address) #connect to broker
client.publish("house/main-light","OFF")#publish
Note: I am using my own local broker but you can use an online broker like the one at iot.eclipse.org.
Subscribing To Topics
To subscribe to a topic you use the subscribe method of the Paho MQTT Class object.
The subscribe method accepts 2 parameters – A topic or topics and a QOS (quality of Service) as shown below with their default values.
We will now subscribe to topics and in this example we will subscribe to the topic house/bulb1 which is also the same topic that I’m publishing on.
Doing this lets us see the messages we are publishing but we will need to subscribe before we publish.
So our script outline becomes.
- Create new client instance
- Connect to broker
- Subscribe to topic
- Publish message
Our new example script is shown below, and I have inserted some print statements to keep track of what is being done.
import paho.mqtt.client as mqtt #import the client1
broker_address="192.168.1.184"
#broker_address="iot.eclipse.org"
print("creating new instance")
client = mqtt.Client("P1") #create new instance
print("connecting to broker")
client.connect(broker_address) #connect to broker
print("Subscribing to topic","house/bulbs/bulb1")
client.subscribe("house/bulbs/bulb1")
print("Publishing message to topic","house/bulbs/bulb1")
client.publish("house/bulbs/bulb1","OFF")
If we run the script this is what we see:
So where is the message that I published?
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.
When the client receives messages it generate the on_message callback.
To view those messages we need to activate and process the on_message callback.
Aside: Callbacks are an important part of the Python Client and are covered in more detail in Understanding Callbacks.
Callbacks also depend on the client loop which is covered in Understanding the Client Loop.
However at this stage it may be better to just accept them and proceed with the script.
To process callbacks you need to:
- Create callback functions to Process any Messages
- Start a loop to check for callback messages.
The client docs describe the on_message callback and the parameters it excepts.
Here is my callback function, which basically just prints the received messages:
def on_message(client, userdata, message):
print("message received " ,str(message.payload.decode("utf-8")))
print("message topic=",message.topic)
print("message qos=",message.qos)
print("message retain flag=",message.retain)
Note the message parameter is a message class with members topic, qos, payload, retain.
I.e message.topic will give you the topic.
Now we need to attach our callback function to our client object as follows:
client.on_message=on_message #attach function to callbackand finally we need to run a loop otherwise we won’t see the callbacks. The simplest method is to use loop_start() as follows.
client.loop_start() #start the loopWe also need to stop the loop at the end of the script (loop_stop()), and in addition wait a little to give the script time to process the callback, which we accomplish using the time.sleep(4) function.
This what our completed example script now looks like:
import paho.mqtt.client as mqtt #import the client1
import time
############
def on_message(client, userdata, message):
print("message received " ,str(message.payload.decode("utf-8")))
print("message topic=",message.topic)
print("message qos=",message.qos)
print("message retain flag=",message.retain)
########################################
broker_address="192.168.1.184"
#broker_address="iot.eclipse.org"
print("creating new instance")
client = mqtt.Client("P1") #create new instance
client.on_message=on_message #attach function to callback
print("connecting to broker")
client.connect(broker_address) #connect to broker
client.loop_start() #start the loop
print("Subscribing to topic","house/bulbs/bulb1")
client.subscribe("house/bulbs/bulb1")
print("Publishing message to topic","house/bulbs/bulb1")
client.publish("house/bulbs/bulb1","OFF")
time.sleep(4) # wait
client.loop_stop() #stop the loop
If you run the script you should see the following
Note: logically you should be able to start the loop before you create a client connection, but it you do then you get unexpected results.
Useful Exercises
You should try commenting out, one by one, the lines:
- client.on_message=on_message
- client.loop_start()
- client.Loop_stop()
and run the script to see the results.
Receiving Messages outside of the on_message Callback
A common question is how do you get received messages into the main script from the on-message callback. There are several ways of doing this as explained in Receiving Messages with the Paho MQTT Python Client
Troubleshoot using Logging
To help troubleshoot your applications you can use the built in client logging callback.
To use it you create a function to process the logging callback. My function is shown below and it simply prints the log message.
def on_log(client, userdata, level, buf):
print("log: ",buf)
and then attach it to the callback:
client.on_log=on_log
You should then see details of connections,publish and subscribe messages like that shown below:
The above is a quick overview to get started you can find out more details in the tutorials below:
Video – Using the Paho Python MQTT Client.
Common Problems
1. Not seeing any messages or not seeing all expected messages.
Possible causes
- You haven’t started a network loop or called the loop() function. Or you haven’t registered or created the callback functions.
- You haven’t subscribed to the correct topics or subscription has failed.
- Access restrictions are in place.
2.- My messages don’t appear in the order I expected?
Possible causes
- The callback functions are async functions which can be called at any time. Use a queue to store the messages and print in one place. I use the Python logging module.
——–> MQTT Python Beginners Course
Important Changes for MQTTv5
Although you may not currently be working with MQTTv5 I would advise you to take a look at the Client changes for MQTTv5 tutorial as by making slight changes to your code you can make it future proof.
MQTT Python Kindle Book
If you prefer all of my MQTT pythons tutorials all in one place then you might be interested in my Kindle Book.
Working with the Paho Python MQTT Client
Related Tutorials and Resources
I didn’t see messages but i think code is true. what am i missing out on? I add code and window run code.
import paho.mqtt.client as mqtt #import the client1
import time
############
def on_message(client, userdata, message):
print(“message received ” ,str(message.payload.decode(“utf-8”)))
print(“message topic=”,message.topic)
print(“message qos=”,message.qos)
print(“message retain flag=”,message.retain)
########################################
def on_log(client, userdata, level, buf):
print(“log: “,buf)
#######################################
broker_address=”192.168.137.1″
#broker_address=”iot.eclipse.org”
print(“creating new instance”)
client = mqtt.Client(“P1”) #create new instance
client.on_message=on_message #attach function to callback
print(“connecting to broker”)
client.connect(broker_address) #connect to broker
client.loop_start() #start the loop
print(“Subscribing to topic”,”house/bulbs/bulb1″)
client.subscribe(“house/bulbs/bulb1”)
print(“Publishing message to topic”,”house/bulbs/bulb1″)
client.publish(“house/bulbs/bulb1″,”OFF”)
client.on_log=on_log
time.sleep(4) # wait
client.loop_stop() #stop the loop
Run window:
creating new instance
connecting to broker
Subscribing to topic house/bulbs/bulb1
Publishing message to topic house/bulbs/bulb1
log: Sending CONNECT (u0, p0, wr0, wq0, wf0, c1, k60) client_id=b’P1′
log: Received CONNACK (0, 5)
log: Sending CONNECT (u0, p0, wr0, wq0, wf0, c1, k60) client_id=b’P1′
log: Received CONNACK (0, 5)
Process finished with exit code 0
You are getting an authentication failure probably because you are using mosquitto v2 see note here and basic config file to fix it.
http://www.steves-internet-guide.com/mosquitto-broker/
I solved . Thanks Mr. Steve.
I have few devices and want to connect to a particular device to subscribe to all the attribues. I can subscribe and get all the attributes by giving the IP address as a broker and access token of that particular device to the username. But where as if I try localhost as a broker, we can subscribe but don’t know for which device we are subscribing. Can you please help me how to connect and subscribe to a particular device by using localhost or 127.0.0.1
Hi
You cannot send to a device only to a topic. I think you may not understand how MQTT works. Take a quick look at this.
http://www.steves-internet-guide.com/mqtt-basics-course/
Hi Steve, thanks for the guide. I’ve written 2 scripts to use the broker to continuously send screen shots of one machine to all connected clients. This works really well (I can send you the scripts if you would like to see how I did it). However, when I run my client script from another machine on my network, it fails to connect to the mosquitto broker. When I run mosquitto from DOS in verbose mode, it doesn’t detect anything when I try to connect. I’ve opened up the ports in the firewall to allow incoming and outgoing traffic but that had no effect. I’ve allowed anonymous connections and setup an empty password file and that had no effect. Do you have any thoughts on why a machine on the local network can’t connect to my local installation of the mosquitto broker? Thanks! Andy
Yes like to see them. I’ll drop you an email so you can add them as an attachment.Can you ping the machine?
Yes, the machine pings. I didn’t look at the routers firewall. Do you think that could be it?
Yes, I can ping the machine. I added a rule to my router to allow the unencrypted mosquito port. That didn’t seem to help.
If you can ping it then it can only be that the broker isn’t running on that port of the port is blocked by a firewall.
I can’t remember if you said it was running on windows or Linux.
rgds
Steve
Windows. FYI – Didn’t receive an email about sending those scripts.
Hi Steve, first of all many thanks! very insightful and well structured guide.
I have some doubts regarding the callback functions in scenarios when I have to wait until a certain condition happen. In my app, I want to be listening in 2 different topics and once I have received a message in both of them, perform some calculations with both of them.
Eg:
(main thread) mqtt_client.sub(topic1)
(thread Nº2) mqtt_client2.sub(topic2))
(thread Nº3) do_calculations(message_from_topic1, message_from_topic2)
And then send some messages to both topics:
mqtt_client.publish(topic1, “5”)
mqtt_client.publish(topic2, “20”)
To be able receive messages from both topics, the main thread is listening to topic1, and a parallel threads is listening to topic2. I’ve thought using a third thread that is in parallel waiting to obtain the values from the messages in both topics, but with the callback at the middle, I cant sort out with a solution… do you have any hint on dealing with this kind of situation?
BR,
Adrian
Use a dictionary from store the messages from each thread but you don’t need separate threads to listen on two topics.
periodically check the dictionary values in the main thread and call the calculate function when both are present then clear the values from the dictionary.
You could also extend the client class simply by using
client.topic1_message= #use for topic 1 and client.topic2_message= in the on message callback
if topic==topic1:
client.topic1_message=msg
if topic==topic2:
client.topic2_message=msg
if client.topic1_message!=”” && client.topic2_message!=””:
do_(calculation(client.topic1_message,client.topic2_message)
rgds
steve
Steve, Again Great App.
I have another question: The broker I am accessing has a config topic and a data topic. I can subscribe to the config topic to see what data is valid in the data topic. config looks kind of like
config/subtopic1/… like 7000 topics here
config/subtopic2/… like 7000 topics here
I subscribe to config/#, and sometimes I get all 14000 messages back quite quickly, and then sometimes I only get a subset like 1500, and no matter how long I wait (10 minutes) it doesn’t send anymore messages. The information in config is slow changing, as the valid data doesn’t often change. How do I force the broker to send all topics under config/#? I suspect it’s to do either of the following:
qos setting (1)
clean_session (True)
I suspect that you may have had clean session false and so when you reconnected you got messages for all of the topics you were previously subscribed to.
MQTT on sends you a message if it has one for that topic.
It will only hold messages if the they were published with a retain flag set or qos 1 or 2 and you connected with clean sessions set to false.
Rgds
Steve
Steve,
Thanks for the great tutorials!!
I have an application that has a mqtt setup. The basic application works fine. What I am wondering, I have multiple topics and would like to check the state of one particular topic from a different application.
What is the best way to do just a one time check for a topic state. And get the state information and then exit.
Hi
Just connect,subscribe and check for a message and then disconnect.
Rgds
Steve
Thanks Steve,
I have hopefully a quick question:
subscribing to something/something/something/#
there are say 20,000 topics here:
I care about like 10,000 of them:
on_message() gives me the actual topic:
something/something/something/topic1
If I don’t want topic1 can I unsubscribe to it specifically, and still stay subscribed to the rest.
I actually created a 10,000 element array of tuples of the topics I care about, but it seems way slower than using /#.
I don’t think you can unsubscribe to a sub topic when using the wild card but is interesting and I will check.I would just filter out the topics I didn’t want.
Hope that helps.
Rgds
Steve
I tried it and unsubscribing to a sub topic doesn’t work
rgds
steve
Hi Steve,
Following your code as in the video: https://www.youtube.com/watch?v=QAaXNt0oqSI&t=620s
everything works using “broker.hivemq.com” as broker.
When I put localhost or 127.0.0.1 the program does not give any error and it also waits 4 seconds, then it finishes. The only problem is that it does not write the message, in fact when I am in localhost the publish does not work (I think).
Can you tell me why this happens and how can I solve it?
If I put back the correct broker everything works.
Also mqtt.Client I have not understood well what it is for, it could be also for this that it gives error, maybe I use it wrong?
Thank you very much for the tutorials and especially if you can answer my question.
Have a nice day,
Daniel
Localhost and 127.0.0.1 means that the broker is running on the machine that you are running the script on.
The client code is in a class called Client. At the top of the code you see this
import paho.mqtt.client as mqtt #import the client
this imports the file containing the code and assigns it the name mqtt.
to create an instance of the Client class with use
client=mqtt.Client
Thanks Steve, it works!
I’m new to mqtt paho. I learned a lot from your tutorial. I try to configure the client and test it out. My question in what file should I configure? Thanks.
I would get the pub and subscribe script from my site and work with that to get you started
Hello Steve, I’d like to thank you for the immense help you provide for people and the amount of resources on this site is amazing. I’ve started to work on MQTT and I do have a couple of questions. First of all, if I get a reading from a temperature/heart sensor, how would I transfer that reading to MQTT and secondly, would publish and subscribe files be in one, as certain examples added them together, or do we keep it different and make a main file and import the publish and subscribe files?
I appreciate your time.
Thank you
It depends on how you are getting the reading is it coming using http etc. As regards publishing and subscribing they can be the same script and often are.
Rgds
Steve
Hi Steve,
There is a way to accept a specific ip adress to connect on mqtt server or get the origin ipadress in on_message callback without using userdata or ipAdress on the subject
Thanks,
No. The sender would need to send the data as art of the message or in userdata if MQTTv5
Rgds
Steve
Hello,
I want to know if it is possible to know if an mqtt client with a particular id has disconnected.
i have a python script which connects to an mqtt server and i have a javascript which connects to an mqtt server. i want to know if from python script i can know if my javascript client disconnected from mqtt server
Hi
This is not built into the protocol but you can do it yourself take a look at this article
http://www.steves-internet-guide.com/checking-active-mqtt-client-connections/
How to develop application MQTT with Java and how to send data to MQTT server and how to get data from MQTT server
Hi
Sorry but I don’t use Java.
very good tutorial thank you.
it s easy to understand
I wish you can make one for MQTT V5.0 with python
Hi
I’m waiting for the paho client to support it.
Rgds
Steve
Hi Steve, I have some doubts about the latency of mqtt, I want to measure the latency time of the messages in each QoS, but I’m not sure how to do it. I’m thinking of using Wireshark but I don’t know if there is another way to do it by programming the clients from python, for example.
Thanks for your help.
Regards
Hi
I wrote some python scripts that did this and logged the data to a database or log file some time ago but never published them. I just took a look at them and they use an old function that I don’t longer use so they need a little rewrite. Here is a sample output
[{“time”: 1531239454.9252646, “time_taken”: “0.025”, “count”: 1, “broker”: “192.168.1.157”}, {“time”: 1531239454.9262555, “time_taken”: “0.206”, “count”: 1, “broker”: “iot.eclipse.org”}, {“time”: 1531239454.9267304, “time_taken”: “0.067”, “count”: 1, “broker”: “test.mosquitto.org”}, {“time”: 1531239454.9271038, “time_taken”: “0.088”, “count”: 1, “broker”: “broker.hivemq.com”}, {“time”: 1531239454.927502, “time_taken”: “0.098”, “count”: 1, “broker”: “m21.cloudmqtt.com”}]
[{“time”: 1531239484.9253147, “time_taken”: “0.020”, “count”: 2, “broker”: “192.168.1.157”}, {“time”: 1531239484.9263163, “time_taken”: “0.422”, “count”: 2, “broker”: “iot.eclipse.org”}, {“time”: 1531239484.9268985, “time_taken”: “0.297”, “count”: 2, “broker”: “test.mosquitto.org”}, {“time”: 1531239484.9274166, “time_taken”: “0.308”, “count”: 2, “broker”: “broker.hivemq.com”}, {“time”: 1531239484.927918, “time_taken”: “0.318”, “count”: 2, “broker”: “m21.cloudmqtt.com”}]
If that is of interest You can have them as is and modify accordingly or I can rewrite the code so that they work without the old functions but I probably need a week to do this.
Get in touch using the ask steve page.
Rgds
Steve
Thanks a lot for the tutorial! I’ve been playing around with ESP8266 and ESP32 chips, and have so far stayed away from MQTT. Your tutorial made me realized that MQTT is a great complementary to using raw tcp/ip!
As a first exercice, I wanted to create a “mailbox” service through a persistent session. The incentive is a low power ESP8266 device that sleeps most of the time and periodically wakes up and checks if there are any pending commands for it.
Meanwhile I implemented this as a receiver and sender on my Linux host. The two scripts `mbox_receiver.py` and `mbox_sender.py` can be found in the following gist: https://gist.github.com/dov/d0dd06d702e5e456f8022774b4089f1b
Even though my solution seems to work works, I feel that the way the receiver wakes up and asks the broker for new messages is very hackish. I’m not even sure that I might not miss a message. What I currently do is the following:
“`
broker_address=”127.0.0.1″
client_name=’mbox’
is_first=True
while 1:
client = mqtt.Client(client_name, clean_session=is_first)
is_first=False
print(“polling”)
client.on_message=on_message #attach function to callback
client.connect(broker_address) #connect to broker
client.subscribe(‘mbox/#’,qos=1)
client.loop_start()
time.sleep(0.1) # How long should this time be?
client.loop_stop()
# client.loop(0.1) # why doesn’t this do the same as the previous three lines?
client.disconnect()
time.sleep(5) # Here the esp chip will go into deep sleep, e.g. for 5minutes.
“`
My questions are:
– Is there no direct way of polling for a message, instead of the indirect method of using `loop_start();…;loop_stop()`.
– Why doesn’t the receiver work when I do `loop(0.1); instead of `loop_start(); sleep(0.1); loop_stop()`? What is the difference?
– Is the receiver guaranteed to receive all the messages?
– Is there a better way to implement this?
Thanks a lot!
Hi
There is no need for polling. If the client connects with a clean session of false and subscribes with a qos of 1 and then the commands are published with a qos of 1 then when the client wakes up and connects it will receive all of the queued commands.
As for the loop just use loop_start() and loop_stop(). So my code would look like
Wake up
process messages
wait a bit
go to sleep
Hope that helps
Rgds
Steve
Hi Steve,
I am trying to parse the payload from a BLE Gateway. Using MQTTBox, I’ve verified incoming messages from the BLE Gateway to the Broker. But if I try to parse the message, the code doesn’t seem to work properly. I could not see the “failure”… must be somewhere in the function “def on_message” / the “for-loop”.
I tried to debug the code, but I can’t find the failure. Would be glad, if you could help me. Thx a lot – Joern
###########################
import time
import paho.mqtt.client as mqtt
import msgpack
import logging
from beacontools import parse_packet
def on_subscribe(mosq, obj, mid, granted_qos):
print(“Subscribed: ” + str(mid))
def on_log(client, userdata, level, buf):
print(buf)
def on_message(mosq, obj, msg):
for d in msgpack.unpackb(msg.payload)[b’devices’]:
#parse iBeacon data
advData = d[8:]
adv = parse_packet(advData)
if adv == None:
continue
print(“=============================================”)
print(“mac:{:02X}{:02X}{:02X}{:02X}{:02X}{:02X}”.format(ord(d[1]), ord(d[2]), ord(d[3]), ord(d[4]), ord(d[5]), ord(d[6])))
print(“rssi:”, ord(d[7]) – 256)
print(“UUID: %s” % adv.uuid)
print(“Major: %d” % adv.major)
print(“Minor: %d” % adv.minor)
print(“TX Power: %d” % adv.tx_power)
def on_connect(mosq, obj,flags, rc):
mqttTopic = “AlteZiegelei”
print(“Connected with result code “+str(rc))
mqttc.subscribe(mqttTopic, 0)
print(“Connected”)
mqttHost = “mqtt.bconimg.com”
mqttPort = 1883
mqttc = mqtt.Client()
#mqttc.on_log = on_log
mqttc.on_connect = on_connect
mqttc.on_subscribe = on_subscribe
mqttc.on_message = on_message
mqttc.connect(mqttHost, mqttPort, 60)
mqttc.loop_forever()
Hi
Can you send me the code as an attachment to steve.w.cope@gmail.com and also an example of the payload you are trying to decode if possible. Change the extension to .txt and .py will be blocked by spam filters
Rgds
Steve
Hi Steve,
thanks a lot for your guideline – very helpful for me as a beginner (in python as well as mqtt). While trying your example (Pycharm, Python 3.6), I don’t get any output regarding the successful messages. With the logging command I recieve this:
creating new instance
connecting to broker
log: Sending CONNECT (u0, p0, wr0, wq0, wf0, c1, k60) client_id=b’P1′
Subscribing to topic Test
log: Sending SUBSCRIBE (d0, m1) [(b’Test’, 0)]
Publishing message to topic Test
log: Sending PUBLISH (d0, q0, r0, m2), ‘b’Test”, … (3 bytes)
log: Received CONNACK (0, 0)
Process finished with exit code 0
I am using test.mosquitto.org as broker.
Q: Why is there a ‘b’ upfront of my topic “test” and what could be the problem, that I didn’t recieve the message from the function “on_message”?
Thank you very much
The b stands for binary and is nothing to worry about.
Usually the reason you don’t see a message is that you aren’t running a loop
Rgds
Steve
Thank you for your answer. Regarding the failed message – I am using your code – so where might the error could be? Thx a lot again! Joern
import paho.mqtt.client as mqtt # import the client1
import time
############
def on_message(client, userdata, message):
print(“message received “, str(message.payload.decode(“utf-8”)))
print(“message topic=”, message.topic)
print(“message qos=”, message.qos)
print(“message retain flag=”, message.retain)
def on_log(client, userdata, level, buf):
print(“log: “,buf)
########################################
broker_address = “test.mosquitto.org”
#broker_address=”iot.eclipse.org”
print(“creating new instance”)
client = mqtt.Client(“P1”) # create new instance
client.on_message = on_message # attach function to callback
client.on_log = on_log
print(“connecting to broker”)
client.connect(broker_address) # connect to broker
client.loop_start() # start the loop
print(“Subscribing to topic”, “Test”)
client.subscribe(“Test”,0)
print(“Publishing message to topic”, “Test”)
client.publish(“Test”, “OFF”)
time.sleep(4) # wait
client.loop_stop() # stop the loop
Hi
The script is probably ending before the message arrives increase the time.sleep to 20 secs
rgds
steve
That was helpful! Thx a lot!
Best, Joern
Switching to broker.hivemq.com worked for me.
Hi steve, I want to ask about this
what is “(u0, p0, wr0, wq0, wf0,c1, k60) client_id=b’control”,?
can you explain one by one?
Thankyou
it is user,password, retain,qos,not sure,clean session and keep alive=60
Amazing introduction, That’s help me a lot to get a first idea.
Thank you very much Steve!
I have 1 problem, in my script I subscribe to the Charge topic, when it receives a message about the start of charging and the amount of energy, I use the on_message method, check if the topic matches (msg.topic == ‘Charge’) and if they match, I call my own charging function, in this function in an infinite loop I constantly increase the value of the buffer variable, and at certain times I want to publish, but instead of publishing at the right time, it waits for the whole cycle to work (it outputs to the console as if it were publishing ) and only after the cycle is done, he sends all my publishing to the broker
Hi
Can you use the ask steve page and send me the script and I’ll take a quick look.
Also what do you mean by cycle is done?
Rgds
Steve
Hey Steve,
what happens if the broker receives a message on the same channel at the exact same time?
By channel I assume you mean topic. The broker processes it as normal. there is no conflict
Alright. Thank you for your answer.
One more Question:
What happens if a client receives a multiple messages on the same topic on the exact same time?
For example when I have two raspberry pies as clients which are publishing data to each other?
Kind regards, John
It still works ok as they can’t actually be at the same time because they need to come in via the tcp socket which will hand them over to the client or broker depending on what is running.
rgds
steve
Thank you! But there will be a bigger delay then? I calculated the delay with seven different clients a, all publishing to one client b on a specific topic and the delay till this client b publishes back an answer is sometimes over 1.5s?
The delay is getting bigger the more a clients are sending a message to the client b.
That is to be expected I didn’t realise that you were measuring the delay I thought you were only worried that the client would receive all the messages.
Rgds
Steve
is there a way to minimize this delay?
Can you explain to me exactly the process and I’ll have a think about it.Maybe better to use the ask steve page
rgds
Steve
Hello Steve,
I want to store the sensor data which is published on MQTT Broker into a database.
I followed your article and it is working properly. I used PAHO Mqtt client to subscribe to the topic.
When this paho mqtt client is subscribed to the topic, the data is stored in database. So how this paho mqtt client can be subscribed to the mqtt broker all the time so that the data will always be stored in the database.
Is there any way that paho client is subscibed to the mqtt broker 24×7? Or is there any other way so that each and every message published on the mqtt broker will be saved in the database?
In the subscribe script you use loop_forever or a infinite while loop at the end of the script to keep it running.
Rgds
Steve
Hi Steve
I have installed MQTT broker (mosquitto) on my windows and created a MQTT service port 1883. However when i am trying to run your script, the client is unable to set up connection with broker. I guess I am missing something in broker_address. In broker_address what address I should use?
Thanks
you need the address of the windows machine running mosquitto. Go the the command line and type ipconfig it will show you the address.
rgds
steve
Hi Steve, hope everything is going well, I have a problem with MQTT, as I imported paho.mqtt.client as MQTT and it is working only on pycharm, but when I run a publish.py or subscribe.py files on VSCode it gives me: ImportError: No module named paho.mqtt.client, how can I fix it? I prefer to work using VSCode.
regards
Hassan Majdalawi
Hi
Don’t know vscode but the error is because the module is not in the search path for vscode. you may need to install the module using vscode
rgds
steve
Hi ..iam stared to do a project to create a dashboard in cumilocity IOT server .mqtt box is using for testing cumilocity APIs ..do you have any material or idea where to start with ..all your vedios are so informative
Sorry but I’ve never used cumilocity IOT server.
Rgds
Steve
Hey, Steve,
thank you very much for this ingenious and easy to understand tutorial.
I try to connect small model cars via MQTT to a broker. Every time they pass a certain section of the track, they have to connect to the broker, subscribe, send a message with the section and after they receive a message from the broker, they unsubscribe and disconnect again. This works quite well so far, but only the first time. As soon as the client (car) is unsubscribed, disconnected and the loop is terminated with loop_stop(), nothing happens anymore if they cross the section again. It is not due to the flags. After loop_stop they are set to the initial value again. Have you got an idea whats the problem here?
Here the important code-segments:
class carClient:
def run(self, position):
self.position = position
assining functions to callback
self.client.on_disconnect = self.on_disconnect
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
self.client.connect(“localhost”, 1883, 60)
time.sleep(1)
self.client.loop_start()
while self.r == True: #in on_message r is set to FALSE
self.client.subscribe(“test_channel”) # subscribe topic
self.client.publish(“test_channel1”, (“ID:” + str(self.id) + ” Position: ” + str(position)))
time.sleep(4)
self.client.disconnect()
self.client.loop_stop()
self.r == True
I call the function from the main method in another class:
if position == 20:
carClient.run() #instance of the class where all the mqtt methods are.
Thank’s a lot and stay healthy.
Here is a link to the whole code:
https://stackoverflow.com/questions/61043898/how-to-handle-the-loop-stop-mosquitto-mqtt
Hi
I took a look at the complete code and it looks ok. There is no need to stop the loop but you should disconnect.
Can you check that the run function is being called more than once.
Rgds
Steve
Hey Steve,
thanks for the fast reply. It’s still not working. It’s working as long I don’t disconnect(). As soon as I use the disconnect() – method, the client (car) is not publishing again if it crosses the section a second time.
I even tested it with a second if loop for track piece == 23, but still the same problem.
Hi
Can you use the ask steve page and send me a copy of the code and we can deal with it via email
rgds
steve
Hi Steves
I want to send mqtt data to other process using multiprocessing and do something in this process (like send it to can bus). But I don’t know how to do that. Could you help me. Please!!!
I try to do this thing, but it not work
def on_message(parent_conn, client, userdata, msg):
print(msg.topic + ” ” + str(msg.payload))
parent_conn.send(msg.payload)
Hi
Are you sending it to a process on the same machine? Do you want to use D-Bus
Rgds
Steve
Yes I send it to the same machine
And I have never used D-Bus. My purpose is send data to other process now i use multiprocessing package in python but if this thing give me the bester way, no problem
Ok I think I get what you mean can you send your script to me using the ask steve page. You don’t need to paste it in as once I get the message I can email you.
Yes I can do that
Hi,
I am in a great trouble in working with MQTT in Raspberry Pi and Arduino. The task now is that I have got RPi-A and RPi-B and they need to communicate through MQTT. RPi-A gets some information and sends it to RPi-B. RPi-B makes the decision and sends back the result to RPi-A. I have seen some examples in sending one way, say RPi-A is the publisher and RPi-B is the subscriber. It’s totally fine. Yet, I can’t find a suitable example with codes in two-way communications as discussed above. Do you have any suggestions. I have tried to include on_publish to publish the message in same topic (and different topics) as in subscription, but both way doesn’t help. I am not good at programming and communication. Could you help suggest an example so that I can solve the problem and understand the logic? Really Thanks
Karl been away for a few days I will take a look
rgds
steve
Hi
I’ve created a tutorial that might help along with python download code
http://www.steves-internet-guide.com/two-way-communication-mqtt-python
rgds
steve
Hi,
my connection was successful and able to PUBLISH, but I do not get any message from my SUBSCRIBE(subscribe topic does show up at AWS IOT test). I have the loop_start() there. the only difference between your example and what I do was via the port 443 with ALPN
I have the following,
def ssl_alpn():
ssl_context = ssl.create_default_context()
ssl_context.set_alpn_protocols([IoT_protocol_name])
ssl_context.load_verify_locations(cafile=ca)
ssl_context.load_cert_chain(certfile=cert, keyfile=private)
mqttc = mqtt.Client()
mqttc.on_message=on_message
ssl_context= ssl_alpn()
mqttc.tls_set_context(context=ssl_context)
mqttc.connect(aws_iot_endpoint, port=443)
mqttc.loop_start()
print(“Subscribing to topic”, topic)
mqttc.subscribe(topic)
any idea what I did wrong?
thanks.
Hi
Looks ok. Did you say you can publish ok? When you say subscribe doesn’t show up do you mean you don’t see it on the amazon server?
rgds
steve
Hello everyone,
I made a SDR-POCSAG receiver with decodes pager messages. It spits out these messages on the terminal. Since I’m a newby and no C++ or Python programmer, I need a little help from you guys to get me in the right direction. to find a simple way to get those messages converted into MQTT messages. Anyone any suggestions on how to achieve this? Much appreciated!
If it’s linux you could pipe it to a python program that retransmits the data using mqtt.
Are you in a position to run a python program on the device.
Rgds
Steve
Sorry for my delayed responce, I thought I’d recieve an e-mail if there was a reaction.
It is is possible to run a Python program om my system, only I don’t have the programming skills to get that working. Any help in this regard will be much appreciated.
Jan
I did reply by email but guess you didn’t get it.
If you are on a linux system then Python is already installed and all you need to do is run it from the command line. If you are on windows then you first need to install python here is the link
https://www.python.org/downloads/
hi, i am getting this error
log: Sending CONNECT (u0, p0, wr0, wq0, wf0, c1, k60) client_id=b’python1′
log: Sending DISCONNECT
log: failed to receive on socket: [WinError 10054] An existing connection was forcibly closed by the remote host
Can you plz help
Hi
It look like the broker isn’t there or on the right port have you tried testing to the broker
test.mosquitto.org
rgds
steve
What IDE are you using ?
The default python ide
Steve, can you tell me what is causing this error?
I think I am doing what you are instructing in this document.
Python 3.5.2 |Anaconda 4.2.0 (64-bit)| (default, Jul 5 2016, 11:41:13) [MSC v.1900 64 bit (AMD64)] on win32
Type “copyright”, “credits” or “license()” for more information.
>>> import paho.mqtt.client as mqtt
>>> Client(client_id=””, clean_session=True, userdata=None,protocol=MQTTv311, transport=”TCP”)
Traceback (most recent call last):
File “”, line 1, in
Client(client_id=””, clean_session=True, userdata=None,protocol=MQTTv311, transport=”TCP”)
NameError: name ‘Client’ is not defined
>>>
Should be
client = mqtt.Client(“client_id”)
Do you mean that I should replace
Client(client_id=””, clean_session=True, userdata=None,protocol=MQTTv311, transport=”TCP”)
with client = mqtt.Client(“client_id”) ?
then the rest of the tutorial should work?
yes
Hi, I want to run a stress test to see the performance of current backend infrastructure and specifically the Mqtt broker feature.
1. Clone account 10 000 (we have a tool to clone users)
2. Get a session for the 10 000 cloned users (client – server).
Use the session to connect to the broker (mqtts://xxxx.xx:xxxx)
Create topics for all the DoSs and publish a message with qos type 2.
We are interested to know:
CPU
Memory
Apdex, response times, network, errors of (login and get a session, connect to the broker, subscribe to the topics and publish messages)
cold u help me out here
Hi
What specifically can you use the ask steve page to reply
rgds
Steve
hi steve,
thanks for this page. as a beginner i want to ask u can we connect multiple publisher and subscribers with the same program above or do it in different way. please help me with this. thankyou
Yes but they all need different client ids. When you are doing this in bulk most people use a random client id
rgds
steve
Hi Steve,
thanks a lot for your very helpful tutorial – it helped me a lot already.
I’m running a mosqitto broker V1.5.7 on a Raspbian Buster and observe a significant difference in behaviour on topic subscriptions between python 2.7.16 and python 3.7.3.
With the same python code, on python 2 the on_message callback is triggered, while on python 3 the callback is not triggered. In both cases the connect and subscribe (mid checked) are successful.
Are you aware of any significant difference between the python 2 and 3 behaviour of the paho-mqtt package?
Is there any kind of diagnostics to check the on_message function or the loop running?
best regards
Gerhard
Hi
I have used several python versions 3.4,3.5, and 3.6 and not noticed anything like that. What version of the client are you using?
Can you send me the script (use ask steve page ) and I’ll take a look
rgds
Steve
Yes I have the same issue with python 3.8.1. I didn’t test with an older version of python but the callbacks aren’t triggered.
Hi
The problem with the callbacks in 3.8 turned out to be that the client loop wasn’t being called. This is common so it is the first thing to check
rgds
steve
With the loop that works. Sorry I don’t see that in your very useful website. Thanks.
Glad you got it working.
Rgds
Steve
Greetings Steve.
Thank you for your site. I have a problem, i want to store my data to database i had create. And the problem is how to store data from esp8266 over mqtt to database. Could explain how to do that?
it’s pleasure if you could help me.
Hi
Take a look here
http://www.steves-internet-guide.com/logging-mqtt-sensor-data-to-sql-database-with-python/
rgds
Steve
First of all, thank you for your site. Here is my very curious problem: If your program turn on a computer, everything works perfectly. Now if the same program run on two computers that need to connect to the same broker, on both Python consoles, the message ” Broker connected” appears cyclically in the same way on both computer. By stopping a one of two programs the other one is working normally again.
The problem persists with different topics.
By inhibiting the “on message” callback function, the problem disappears.
Do you please an explanation to problem and especially a solution? Cordially
Hi
It is because the clients are using the same client id. Change the client id of one of the clients
Rgds
Steve
Minor quibble with the example for installing on Raspberry Pi. When I tried it,
pi@WxPIcture:~ $ pip3 install paho-mqtt
Traceback (most recent call last):
File “/usr/bin/pip3”, line 9, in
from pip import main
ImportError: cannot import name ‘main’
Being a “python newbie” (even though I’ve been programming nearly 50 years), it took me a while to track down that you need to use “sudo pip3 install paho-mqtt”
Sorry I’ll correct it.
Rgds
Steve
Hi,
I am working on my thesis and I have a Problem to connect with the Broker in the very first example:
———————————————————————————–
import paho.mqtt.client as mqtt #import the client1
broker_address=”192.168.178.35″
#broker_address=”iot.eclipse.org” #use external broker
client = mqtt.Client(“P1”) #create new instance
client.connect(broker_address) #connect to broker <-at this po
client.publish("house/main-light","OFF")#publish
———————————————————————————–
As broker address I type in my IP4 address: 192.168.178.35 as shown in my network settings. My Computer is running on Ubuntu and connected to my FritzBox 7330 SL. If I use an external broker address everything works fine. But if I use my address in my own FritzBox network the following is shown in my Terminal:
——————————————————————————————–
Traceback (most recent call last):
File "paho_test.py", line 5, in
client.connect(broker_address) #connect to broker
File “/home/flint/.local/lib/python3.6/site-packages/paho/mqtt/client.py”, line 839, in connect
return self.reconnect()
File “/home/flint/.local/lib/python3.6/site-packages/paho/mqtt/client.py”, line 962, in reconnect
sock = socket.create_connection((self._host, self._port), source_address=(self._bind_address, 0))
File “/usr/lib/python3.6/socket.py”, line 724, in create_connection
raise err
File “/usr/lib/python3.6/socket.py”, line 713, in create_connection
sock.connect(sa)
ConnectionRefusedError: [Errno 111] Connection refused
——————————————————————————————–
I am a very beginner in this topic. It would be awesome, if someone could help me with this
It looks like the broker isn’t running or is using a different port
Rgds
Steve
Thank you Steve!
The broker wasn’t installed correctly. Now it works fine!
And a great thanks for your site!
Artur
Hi Artur,
I hope find you well,
i would ask you how you solve this issue.
yosef.hammad@gmail.com
I want to cutoff voltage using MQTT. How can I achieve it?
You will need to supply more details like what are you controlling what is the end device and what interface. Is it a tasamota switch for example etc
rgds
steve
Hey Steve, I want to replace the communication in my project (HTTP/REST to MQTT). For that I need to realize a Request-Response-Pattern with MQTT. Is there any option to do this?
Thanks in advance
Adrian
In MQTT v5. Take a look here
http://modelbasedtesting.co.uk/2018/04/09/a-story-of-mqtt-5-0/
I’m just starting to create MQTT v5 tutorials as I’ve been waiting for a Python v5 client and the mosquitto broker to support v5.
Hopefully Get most of the done by end July.
Rgds
Steve
Hey Steve,
thanks for your reply.
I also have read about the req-res-pattern in MQTTv5.
Do you see any option to do this with v3.1.1. (Python client ) already or do I have to wait for the Paho implementation as well? 😛
You would have to build it into the App with version 3.1.1. I would imaging v5 will be everywhere by end of this year. You can start development now as mosquitto 1.6 supports v5 and there is a Python client on github gmqtt I think.
Hello Steve!
Thanks for the article! Btw I have a small question: As I can see now, callbacks works only if the method signature is the same as it is required by the paho lib, so this doesn’t work for me:
……
def on_message(message):
print(“message received “, str(message.payload.decode(“utf-8”)))
print(“message topic=”, message.topic)
print(“message qos=”, message.qos)
print(“message retain flag=”, message.retain)
client.on_message = on_message
………
only in case if I add back the client and userdata parameters which are unused at the moment and I dont plan to use them.
Of course I don’t want to learn all the the method signatures for callbacks so do You know a way to handle this? I’m using PyCharm for development but as I can see, it can’t generate automatically the method with the proper parameters. Do you have any advice for this?
Hi
No the callback function that you write need to conform to the callback template for that callback which I get from the docs here https://pypi.org/project/paho-mqtt/
rgds
steve
Hallo Steve,
Thanks for you instructions they are very helpful.
One thing temporarily stopped me was the installation, with the instruction “pip install paho-mqtt”
the installation shall be done for python 2.7.
For python3 what is standard installed on the newer version of rapsberry pi you need pip3.
I am sure you know, but for me it took a while to discover.
Tks for pointing that out I’ll edit the text to make it clear.
You might find this page useful
http://www.steves-internet-guide.com/python-notes/
it goes into running multiple python versions and pip as I had problems with it when I first started and still do from time to time.
Hi Steve,
Great article! Very helpful. In the comments section you wrote:
The on_message callback is what triggers the storing. I wrote a logging script but I used a different thread to store the data so as not to block the main script.
The on_message callback just drops the message in a queue that gets processed by the storage thread.
If you want I will send it to you so you can take a look.
as for the 1000 messages I think it is too many as it could be several hours or days depending on how active the topics are.
Can you send me that code?
Thanks,
Rich
There are two loggers avaialable
teh data logger and topic logger. You can download them here
http://www.steves-internet-guide.com/simple-python-mqtt-data-logger/
and here
http://www.steves-internet-guide.com/simple-python-mqtt-topic-logger/
Hi Steve.
I have garden lights and internal lights that have sonoff smart switches connected but they have been flashed with tasmota. This gives off mqtt results. I have set up a broker with paho-mqtt and your script above in python works and I can turn on and off the lights with a python script. I would love to add the status of each light on a webpage and even be able to turn it on and off from my webpage.
I have a pi nginx webserver which runs my paho-mqtt/mosquitto on too.
Do you know how to get this info on to a webpage ?
Thanks in advance
There are many ways of doing this but I assume you want a python solution running locally.
If so then I have script that was developed for something else but will do what you want. Use the ask steve page and contact me and I’ll send it to you.
You might also want to consider node-red as it makes it very easy to display data. You could also use it for control.
If you need to do this across the Internet then thingsboard will also do it but I’ve only used that for display.
You might find these videos useful
Node-red
https://youtu.be/e70ta8jI_nM
https://youtu.be/Gu0Vq2kVNzw
Thingsboard
https://youtu.be/eqbTNXf2m7s
Hi
Is there any way i can get the list of all clients and their IPs , connected to broker, by altering the same code. Please mention if there is any other way of doing this.
thanks.
Unless you have configured the clients to announce themselves by some mechanism then you can’t do what you want to do.
See this article
http://www.steves-internet-guide.com/checking-active-mqtt-client-connections/
rgds
Using the on_message callback in paho, when i’m subscribed to a topic. i’m trying to display a message if the last message received was over 15 seconds earlier. I’m trying to control a relay based and if i get flooded with messages, the relays opens and closes nonstop.
Any idea on how to do this?
Do you want the relay to only change state every 16 seconds at most?
Hi Steve,
In my callback function, I want to store the messages in the database. I want to store the messages in two databases parallely. Is it okay to create and run threads in my callback? Suppose the threads which I created take more time, will it affect client.loop_forever()
client.on_message= onMessage #attach function to callback
client.loop_forever()
def onMessage(client, userdata, message):
t1 = Thread(target = saveToDatabase ,args=(msg))
t2 = Thread(target = saveToDatabase ,args=(msg))
t1.start()
t2.start()
It is probably better to start a thread for the databases outside the on_message callback and keep it running in the background.
In the on_message callback drop the message in a queue and let the database thread process the queue.
I’ve got code that does it. It’s not tidy but you are welcome to it.Just use the ask steve page to send me your email address
I am quite new to MQTT. I was wondering if we could do a video conferencing using MQTT. If yes, could you please explain in detail how abouts of the process?
Thanks in advance
Probably but MQTT isn’t designed for it and so I doubt if the quality would be any good. I can imagine that one day security cameras might use it but I haven’t really given much thought to it.
Sorry I can’t be of more help.
rgds
steve
Hi Steve,
I have written a script for the python client which works perfectly. Now I need to extend my application wherein I need to pass the client object within multiple python scripts to subscribe and publish to different topics.
Would appreciate if you could help on how can this be done.
TIA
Can you explain a little more on the actual setup.
rgds
steve
Hi steve,
I tried with the script given in this tutorial to connect to broker and I get this error “socket.error: [Errno 111] Connection refused” whenever I try to connect to broker. I am using python version 2.7.12
import paho.mqtt.client as mqtt #import the client1
import time
############
def on_message(client, userdata, message):
print(“message received ” ,str(message.payload.decode(“utf-8”)))
print(“message topic=”,message.topic)
print(“message qos=”,message.qos)
print(“message retain flag=”,message.retain)
########################################
broker_address=”iot.eclipse.org”
print(“creating new instance”)
client = mqtt.Client(“P1”) #create new instance
client.on_message=on_message #attach function to callback
print(“connecting to broker”)
client.connect(broker_address) #connect to broker
client.loop_start() #start the loop
print(“Subscribing to topic”,”house/bulbs/bulb1″)
client.subscribe(“house/bulbs/bulb1”)
print(“Publishing message to topic”,”house/bulbs/bulb1″)
client.publish(“house/bulbs/bulb1″,”OFF”)
time.sleep(4) # wait
client.loop_stop() #stop the loop
Output:
creating new instance
connecting to broker
Traceback (most recent call last):
File “mqttclient.py”, line 16, in
client.connect(broker_address) #connect to broker
File “/usr/local/lib/python2.7/dist-packages/paho/mqtt/client.py”, line 768, in connect
return self.reconnect()
File “/usr/local/lib/python2.7/dist-packages/paho/mqtt/client.py”, line 895, in reconnect
sock = socket.create_connection((self._host, self._port), source_address=(self._bind_address, 0))
File “/usr/lib/python2.7/socket.py”, line 575, in create_connection
raise err
socket.error: [Errno 111] Connection refused
Hi
I just tried it using python2.7 and it worked OK>have you tried using another broker.
Greetings Steve.
Great tutorial steve
I want to create a sensor network using 6 esp8266( in which 4 esp8266 are for sensor and 2 esp8266 for actuators) to send sensor data to mosquito broker hosted in raspberry pi.So l want to create python script which can store all sensor data in mysql database and the python script should compare the received sensor values with the threshold value so that it publish control messages to actuators esp8266.
Can you help me with how best can i do it..if you have any python scriptd related to this ,can you help. I need a starting point.i am a newbi to python and mqtt.
Stay well
I’ve got a script that doe similar I’ll email it you
sorry where can i add the login stmt in the program
You use this
username_pw_set(username, password=None)
before you connect.
I have subscribed to multiple topics using paho mqtt client. On receiving the messages from broker, I want to store the messages into mysql database. I want to gather the messages altogether before inserting into DB.I have set the threshold say 1000 messages. Only when the threshold is reached the messages has to be inserted into DB altogether at once.For each topic, there is corresponding table in the database. Which callback function Should I use on_message() callback or message_callback_add()? Which is better in such scenario? Is there a chance to lose the message because I am collecting the messages in bulk and after some time delay storing it in the database.?
//MySQLDBClass.py
def __init__(self):
self.insertcounter = 0
self.insertStatement = ”
self.bulkpayload = ”
self.maxInsert = 1000
def onMessage(self, client, userdata, msg):
if msg.topic.startswith(“topic1/”):
self.bulkpayload += “(” + msg.payload.decode(“utf-8”) + “,” + datetime + “),”
elif msg.topic.startswith(“topic2/”):
self.insertStatement += “INSERT INTO mydatabase.table1 VALUES (” + msg.payload.decode(“utf-8”) + “,” + datetime + “);”
elif msg.topic.startswith(“topic3/”)
self.insertStatement += “INSERT INTO mydatabase.table2 VALUES (” +msg.payload.decode(“utf-8”) + “,” + datetime + “);”
elif msg.topic.startswith(“messages”):
self.insertStatement += “INSERT INTO mydatabase.table3 VALUES (‘” + msg.topic + “‘,” + msg.payload.decode(“utf-8”) + “,” + datetime + “);”
else:
return # do not store in DB
self.insertcounter += 1
if ( self.insertcounter > self.maxInsert ):
if ( self.bulkpayload != ” ):
self.insertStatement += “INSERT INTO mydatabase.table4 VALUES” + self.bulkpayload + “;”
self.bulkpayload = ”
for result in cursor.execute(insertStatement, multi=True):
print result.row_count
The on_message callback is what triggers the storing. I wrote a logging script but I used a different thread to store the data so as not to block the main script.
The on_message callback just drops the message in a queue that gets processed by the storage thread.
If you want I will send it to you so you can take a look.
as for the 1000 messages I think it is too many as it could be several hours or days depending on how active the topics are.
Okay thank you so much for quick response. Could you please send that script?
sent the scripts by email
Okay Thank you so much. Suppose, I want to log the number of messages received by on_message callback during a time span of 60 seconds. For this should I create a separate own thread outside the on_message callback to log every minute? or I can add time.sleep(60) inside on_message callback? I want to log the number of messages received in time span before sending to the database and log number of messages inserted into database during that time span.
Don’t use time.sleep(). Just keep track of time in the database thread and do write every 60 seconds.
e.g if lastimewrite-time.time()>60
In the on_message callback just drop the messages into a list or use the queue module.
Hi Steve,
Could you please send me the script.
Thanks,
Nag
Hi
You can copy the script direct from the page. If you have problems use the ask-steve page and I’ll email it to you.
Rgds
Steve
Hi Steve,
Can you please share the logger script with me too for reference and understanding.?
The data logger is here
http://www.steves-internet-guide.com/simple-python-mqtt-data-logger/
I want to ask is it necessary to use client.loop_start() and client.loop_stop()? Can I use while True? If I use client.loop_start(), can I exclude the while True loop that is running my Raspberry Pi client program? Thanks.
Yes you need to call the loop function to process the callbacks. see here
http://www.steves-internet-guide.com/loop-python-mqtt-client/
I mainly use a manual loop call inside the while loop rather than using the loop_start() call but it is a matter of preference.
Thanks for your answer! 🙂 I will try your advice.
Hi Steve,
I am trying my first steps with Python and MQTT and maybe you can give me a hint for my beginner problem.
How can i access to message.payload from outside the “def” part?
Concrete i want to use an “if query”, which reacts on a special keyword published by a second client.
Thank you!
I assume you mean that you want to access it outside the on_message callback.
def on_message(client, userdata, msg):
topic=msg.topic
m_decode=str(msg.payload.decode(“utf-8″,”ignore”))
You could use a global variable to hold the message. Using globals is frowned upon but easy to implement
def on_message(client, userdata, msg):
global m_decode
topic=msg.topic
m_decode=str(msg.payload.decode(“utf-8″,”ignore”))
This could be overwritten before you get to process it. A better option is to
append it to a list and pop the message of the list anywhere in the script.
def on_message(client, userdata, msg):
topic=msg.topic
m_decode=str(msg.payload.decode(“utf-8″,”ignore”))
message_list.append( m_decode)
You could also use the queue class which is probably the better solution especially if you are using threads.
I too have some problems processing the incoming payload.
If I try:
def on_message(mosq, obj, msg):
global m_payload
print “Topic: ” + str(msg.topic)
print “QoS: ” + str(msg.qos)
print “Payload: ” + str(msg.payload)
m_payload=str(msg.payload)
if m_payload==’ON’:
GPIO.output(LED1,True)
if m_payload==’OFF’:
GPIO.output(LED1,False)
I get the error: “NameError: global name ḿ_payload’ is not defined ”
that error is triggered by my “if m_payload==’ON’: statement.
if I exactly follow your definition with “m_payload=str(msg.payload.decode(“utf-8″,”ignore”))”
then I get an even bigger error:
“SyntaxError: Non-ASCII character ‘\xe2’ in file mqttpublish.py on line 32, but no encoding declared; see http://python.org/dev/peps/pep-0263/ for details”
That link however doesnot make anything clear.
So my question remains: how to process the payload message??
Any help is appreciated
You are using python2 by the looks of your code.
Maybe the payload is already a string so try
m_payload=msg.payload and see if it works.
If not try to get the payload type and let me know what it is.
rgds
steve
OK solved like this:
def on_message(mosq, obj, msg):
if (msg.payload==’ON’):
GPIO.output(LED1,True)
print ‘lamp aan’
print “Topic: ” + str(msg.topic)
print “QoS: ” + str(msg.qos)
if (msg.payload==’OFF’):
GPIO.output(LED1,False)
print ‘lamp uit’
print “Topic: ” + str(msg.topic)
print “QoS: ” + str(msg.qos)
Well done. I was wondering why you are using python2.
rgds
steve
Using Python 2 because that just happened to be in Jessie and Stretch 🙂
Sadly in my solution the indents have disappeared and indents being are an important part of the code, so I will try to clarify:
The two “if” blocks need to be indented from the “def on_message()” statement
and the GPIO.outlet print “QoS” blocks need to be indented from the “if” statements
Thanks, this is really simplified.
How do I split the publisher and subscriber entities into two separate files?
Publishing from one end and subscribing on the other receiving published info on it?
I’ve sent you two scripts by email
Steve
Can I have the scripts as well?Thanks
Sent by email
Hii steve my code is not running in pycharm.Can you help me
Have you tried the Pythom IDE
Rgds
Steve
I installed the broker on the raspberry pi. Do I have to manually start the broker everytime? Also, should the client always be on the same machine?
It should start automatically on startup
rgds
steve
Thanks. This helped me a lot.
hey , thanks for the tutorial. I would like to post an issue which I came across in mqtt.
I am running a client which connects to the broker running in the same PC using connect function. Then am running loop_start() function and then subscribing to the topic and getting the message . so far so good. Now I want to get disconnected from the broker , so am using disconnect() function. But the Disconnect function returns the value ‘4 ‘ and it is not calling any on_disconnect function. Can you look into this and resolve my issue .
Thanking You
I faced problem I cant cope with. Executing a script:
#!/usr/bin/python
import paho.mqtt.client as mqtt
server = “192.168.88.20”
port = 1883
user = “jXXXX”
passw = “XXXX”
client = mqtt.Client(client_id=”Rpi3″, clean_session=True, userdata=None, protocol=’MQTTv311′)
client.username_pw_set(user,passw)
client.connect(server)
client.publish(“home/rpi3″,”Hello”, qos=0, retain=False)
I got:
Traceback (most recent call last):
File “mqqt.py”, line 10, in
client.connect(server)
File “/usr/local/lib/python2.7/dist-packages/paho/mqtt/client.py”, line 760, in connect
return self.reconnect()
File “/usr/local/lib/python2.7/dist-packages/paho/mqtt/client.py”, line 932, in reconnect
return self._send_connect(self._keepalive, self._clean_session)
File “/usr/local/lib/python2.7/dist-packages/paho/mqtt/client.py”, line 2114, in _send_connect
keepalive))
struct.error: cannot convert argument to integer
Somebody can help?
Have you tried running it with python 3
this question might be silly, but is the broker address referring to my pc ip address?
The broker address is the address of the machine running the NQTT broker. You can use public brokers like “test.mosquitto.org”,”broker.hivemq.com”,”iot.eclipse.org”.
If i’m using my PC as a localhost for the broker then the address should be my pc ip address, am i right? Or can i just type in localhost when i assign the value into the broker_address variable?
Either should work. You might find these two articles useful
http://www.steves-internet-guide.com/host-name-resolution-methods-explained/
http://www.steves-internet-guide.com/hosts-file/
thanks a lot.
Hi Steve,
I have a few question, hope you could help me out.
Q1. Which party generates the Callback? regarding the “on_connect()”. It looks like the client generates after it gets ACK message from broker. Is my understanding correct?
How about “on_message()”
Q2. I am not clear of the meaning of
“def on_connect(Client, userdate, flag, rc):”
“client.on_connect=on_connect”
It looks On_Connect() has been defined in client module already. To me , it just need to check the value of “rc” to make judgment.
Thanks very much!
corydon
Yes you are correct the CONNACK triggers the callback (on_connect).
The client.on_connect() method is defined in the MQTT class.
the statement
client.on_connect=on_connect
replaces the method defined in the class with the one you define in your script called on_connect().
Does that Make Sense?
Rgds
Steve
P.S I just did a video on client connections which you may find helpful
https://youtu.be/vT4FTRgipOM
This was a massive help to me, just starting with mqtt (and Python) and this really got things moving. Thank you.
Chris
Tks for your comments.Glad you found it useful
Steve
Hey! Amazing tutorial. Thank you.
I am trying out some of the things that you have shown. I am able to subscribe to a topic, and have the message show up in my Python console. So the loops, and the call back configuration seems to be all set.
I am trying to compare the message that I am receiving to a known string. How do I do that?
So basically, I need message.payload.decode(“utf-8”) to be equal to a string variable(that’s just variable for Python I guess) and then in an if-statement inside the loop, I am trying to get the code to compare it for every iteration of the loop to a string. Once it is equal, the if statement will run clientName.disconnect() and .loop_stop()
I tried, under the def on_message, to do that by doing this:
messCon = str(message.payload.decode(“utf-8”)) and also down inside the loop. At both places it is either “Invalid syntax” or that my variable ‘messCon’ is not defined.
I am at a loss on how to get this to work.
Any help is appreciated.
-Sohum
Hi
that looks about right it may just be the if statement missing the ==
if messConn==str(message.payload(“uft-8”):
clientName.disconnect()
loop_stop()
how to write our own mosquitto broker? could you explain the code needed there
Hi
I’ve never written one but if you download the testing suite here
there is a python broker which you can study.
Hello,
i’m trying to write o code which gets tempreture value and write it to a file. But my code getting stuck if there is no broker and then when i try to execute code with broker code does not give any response. i’m using loop_start() func. And i understood that there is a thread on the backround working. So is there a way to stop loop properly when there is no broker or stop the code with “ctrl + c”. Thank you
Hi
When the broker disconnects you get a on_disconnect callback.
Put the loop_stop() call in there.
Here is a copy of my on_dicconect()
def on_disconnect(client, userdata, rc):
m=”disconnecting reason ” ,str(rc)
logging.info(m)
client.connected_flag=False
client.disconnected_flag=True
Note: I use a flag to stop the loop because I normally don’t use the loop_start.
So yous would be:
def on_disconnect(client, userdata, rc):
loop_stop()
rgds
steve
Hello. I have a curious problem with a python program similar to that of this site that works alone wonderfully but when I run the same program on two computers simultaneously having to connect to the same broker, both start to connect and
disconnect constantly. By disabling the callback of one of the two, everything works correctly again except that one of the two can obviously no longer receive. This disfunction is observed only when the two programs connect to the same broker even with different topics.
I am curious to know the reason but also to find a solution. Thanks for your help. cordially
You are probably using the same client id on both clients you need to change the name on one of the clients
rgds
steve
Thanks very much. This is the cause of my problem. To the instruction “client = mqttClient.Client(idClient)” , idClient must be different for each computer.
cordially