Internet of things today generates a large amount of data. That data gets exchanged between the broker and clients. We have built Python MQTT interface. It enables application developers to hook in the data from the MQTT Broker. These hooks help build a highly unified application. And at the same time reduce the bandwidth required by half.
This documentation is a comprehensive guide for developers on using Python MQTT hooks. It covers topics such as connecting to the broker, and sending messages. Also it provides a deep tech view on configuring the MQTT Broker. Refer to the sample IoT application that is built using the Python MQTT hooks available in the MQTTRoute.
MQTTRoute is a python based MQTT Broker with a publish subscribe messaging pattern. It provides absolute support for the MQTT protocol . Developers use Python's flexibility to enable an interface for extending the implementation. They use it to build complex IoT applications. Users can extend the MQTT Broker's python modules by connecting to any big data engine. The ready-to-use python interfaces include:
Additional Python interfaces to extend the MQTT Broker & develop complex IoT applications include :
We’ll go over each of these in detail here. The mandate of any IoT Application is
The MQTT Broker includes the Python MQTT Interface. It helps you achieve complete control over your IoT application development.
Custom Data Storage allows users to choose a unique database for their application. It collects the data in the database of your choice. With MQTTRoute user has the ability to store data in the following, databases by default.
1 .MySQL
2 .SQLite
3 .PostgreSQL
4 .MSSQL
5 .ElasticSearch
You can enable these data storage options without a single line of code. You can enable these options in the conf/data_store.conf
We have added the Custom Data Storage Hook. It ensures that MQTT Broker supports all the kind of data storage available. You will be able to enable the Custom storage hook on the same data_store.conf file. The next section will explain how to configure the data storage.
Steps to configure Custom Data Store
1. Open the data_store.conf
2. Make the CUSTOMSTORAGE option as ENABLED. Selecting this option sends received messages from clients to the custom_store.py file. It also includes the default data storage.
3. To store data in the Elastic search, the value of DATASTORE needs to be specified as ELASTIC. The same can be done for all databases. HOSTNAME, the host name or the IP address of the MQTT Broker. MQTT Broker address is 127.0.0.1. PORT, the port of the custom data store you are using. Here it is mentioned as 9200 since the port number of Elastic search is 9200. If you want to utilise a different database, you must specify the appropriate port. The port number for MYSQL is 3306, MSSQL is 1434, and PostgreSQL is 5432. INDEX_NAME specify the name under which data should be stored. This looks like the name of a database. The index name is which you want to store data.
4. If you are planning to have your own data storage, the value of DATASTORE need to be specified as CUSTOM. When you configure custom settings, the MQTT data will be sent to the handle_Received_Payload() method. This method handles the data received from clients. And the data sent to the clients from the broker will be sent to the handle_Sent_Payload() method. This method is present in the python file specified in INTERCEPT_FILEPATH.
data_store.conf
#############################CUSTOM STORAGE###########################
# def handle_Received_Payload(data)
[CUSTOM STORAGE]
CUSTOMSTORAGE = DISABLED
# ENABLED || DISABLED
DATASTORE = CUSTOM
# ELASTIC || CUSTOM
[ELASTIC]
HOSTNAME = 127.0.0.1
PORT = 9200
INDEX_NAME = mqtt
BULK_INSERT_TIMING = 2
[CUSTOM] INTERCEPT_FILEPATH = ./../extensions/custom_store.py
Additional object references
The custom_store.py file offers the following Python object references. These references allow for a closer integration with the MQTT Broker.
1. .db_cursor – the DB reference to work with the database. You will be able to read and write data into the database. The reference of the DB set on the conf file will be provided.
2. elastic_search – Elastic Search cursor if the elastic option is enabled
3. .datasend – Python object that can be used to send any MQTT messages to the connected mqtt clients.
4. web_socket – UI Websocket object for sending data to the UI for any event notifications. This will be helpful on Custom UI implementation.
Refer to the files below and the same available in the
global db_cursor
#
# elastic_search cursor
#
global elastic_search
import os, sys
global datasend
global Client_obj
sys.path.append(os.getcwd()+’/../extensions’)
# Called on the initial call to set the SQL Connector
def setsqlconnector(conf):
global db_cursor
db_cursor=conf[“sql”]
# Called on the initial call to set the Elastic Search Connector
def setelasticconnector(conf):
global elastic_search
elastic_search=conf[“elastic”]
def setwebsocketport(conf):
global web_socket
web_socket=conf[“websocket”]
def setclientobj(obj):
global Client_obj
Client_obj=obj[‘Client_obj’]
#Client_obj
# Importing the custom client class into the handler
from customimpl import DataReceiver
datasend = DataReceiver()
def handle_Received_Payload(data):
#
# Write your code here. Use your connection object to
# Send data to your data store
print ” print in the handle_received_payload “,data
result = datasend.receive_data(data)
# if result is none then write failed
def handle_Sent_Payload(data):
#
# Write your code here. Use your connection object to
# Send data to your data store
print ” print in the handle_Sent_payload “,data
result = datasend.sent_data(data)
Implementation Bonus TIP
After configuring the custom data storage, you will receive the python message callback functions in your python file. These callback functions are used for the method handle_Received_Payload(data)
Your implementation should receive the data and store it and return the method. You should do your data analysis in a separate thread.
When data that has already been acquired is better processed, ML and AI perform best. The Scheduling module assists in the processing of data over a period of time. The Custom Scheduler will assist you in creating your own MQTTRoute schedule. This aids you to aggregate your data by allowing you to add your own code on the server-side.
For example, consider checking the water level in a home. The custom scheduler will help to set a reminder to check the status of the water level in the tank.
def schedule_conf():
schedules={}
schedules={
‘STATUS’:’DISABLE’,
‘SCHEDULES’:[
{‘OnceIn’:1,’methodtocall’:oneminschedule},
{‘OnceIn’:5,’methodtocall’:fiveminschedule}]}
return schedules
global elastic_search
# Called on the initial call to set the SQL Connector
global web_socket
# Web_socket
def setsqlconnector(conf):
global db_cursor
db_cursor=conf[“sql”]
def setelasticconnector(conf):
global elastic_search
elastic_search=conf[“elastic”]
def setwebsocketport(conf):
global web_socket
web_socket=conf[“websocket”]
def setclientobj(obj):
global Client_obj
Client_obj=obj[‘Client_obj’]
def oneminschedule():
pass
#Write your code here
#print “extension print”
def fiveminschedule():
pass
#Write your code here
#print “extension print”
Now start adding your query in ‘#Write your code here‘ field.
UI custom server provides an option to customize the user interface. By default, MQTTRoute comes up with the custom dashboard option with rich widgets. Those widgets helps to visualize the data in a way user need. But to customize UI for advanced & high level IoT / IIoT implementation, Custom UI server will be the right option. You can add your own code on the server-side. This allows you to customize the UI of the MQTT server very straightforward. You can alter the code in Custom_ui_server.py file as you need to customize it.
#
# SQL Connector. It will be sqlite / mssql / mysql cursor based
# on your configuration in db.conf
# Please construct your queries accordingly.
#
import sys
global db_cursor
#username’,’client_send_api’,’topic_name’,’message’,1,0,’10’,0
# elasstic_search cursor.
global elastic_search
from datetime import datetime
import json
import ast
import time
try :
from elasticsearch import Elasticsearch
elastic_search = Elasticsearch(host=”localhost”, port=9200)
except Exception as e :
print(e)
import os, sys
#
#Client object. It used to send/publish client message to any active clients
#Simply call the helper functions with parameters like
User_name,Client_id,Topic_name,Message,QOS, to subscribe to a topic / subscribe
client
global Client_obj
# Called on the initial call to set the SQL Connector
def setsqlconnector(conf):
global db_cursor
db_cursor=conf[“sql”]
# Called on the initial call to set the Elastic Search Connector
def setelasticconnector(conf):
global elastic_search
elastic_search=conf[“elastic”]
def setclientobj(obj):
global Client_obj
Client_obj=obj[‘Client_obj’]
The Data connectors, like the SQL Connector, will be available as a cursor global variable for querying the Database. The Elastic Search connector will also be available for querying Elastic if you have enabled the custom storage option.
MQTT Broker has custom authentication functionality. It enables user to integrate MQTT Broker with any identity access management (IAM) & Single Sign on (SSO).
[CONFIG]
PORT_NO = 1883
WS_PORT_NO = 10443
TLS_ENABLED = FALSE
# TLS_PORT must be 88xx.
TLS_PORT_NO = 8883
WSS_PORT_NO = 11443
########################Device Authentication######################
[AUTHENTICATION]
AUTHENTICATION_ENABLED = NO
# YES || NO
######################## User Interface Details ######################
[UI]
UI_Http_Port = 8080
LIST_API_CLIENTS = FALSE
[WEBSOCKET]
WEBSOCKET_PORT=8081
############# Prefix for Random Client id Generation #####################
[MQTT]
CLIENTID_PREFIX = Bevywise-
# The clean session to be handled by the broker. If the clear_session is marked as
# disabled, then every connection will be made fresh.
CLEAR_SESSION = DEFAULT
# DEFAULT || DISABLED
################ ######### WEB LOGIN ############################
# Securing the Web login XXXX Need to be removed XXX
[WEB_LOGIN_PAGE]
WEB_LOGIN = ENABLED
WEB_USERNAME = admin
WEB_PASSWORD = admin
# ENABLED || DISABLED
################ #### REMOTE AUTHENTICATION ##################
[REMOTEAUTH]
REMOTEAUTH_ENABLED = NO
# YES || NO
INTERCEPT_FILEPATH = ./../extensions/custom_auth.py
How to enable custom authentication option?
Requesting Retries Count
When we attempt to connect to the server, some connection failures may happen. This may be due to entering incorrect login credentials. In that case providing countable retries will be helpful. By entering request retries count, you can add or limit the retries attempt of the user.
# Request Retries Count
requests.adapters.DEFAULT_RETRIES = 3
# Request URL
url = “https://www.bevywise.com/auth”
# Request Timeout
request_timeout = 0.1
# Request Method
request_auth_method = “POST”
# POST | GET | PUT
Enter the URL of your authentication landing page. This authenticates the user attempting to connect with their login credentials.
In custom_auth.py file, provide the URL,
url = “https://www.bevywise.com/auth”
It is the time duration or interval that an application waits for the response from the client. These values are probably given in seconds or milliseconds.
To set the request timeout,
Open custom_auth.py file in extensions folder and enter timeout value in the space given.
request_timeout = 0.1 (By default it carries the value of 0.1)
There are a set of HTTP's request methods. You can select anyone to indicate the desired method to be performed.
GET – Requesting data from a specified resource.
POST – Submit or publish messages to the specified resource.
PUT – Replacing the existing data of the target resource.
Open custom_auth.py file in extensions folder and enter auth method in the space given.
request_auth_method = “POST”
Set all your configurations, save the file & start running the broker.
MQTTRoute offers a complete Internet of Things application. It includes user interface customization, data aggregation, and analysis. Additionally, it enables event data comparison with the processed data. This IoT application framework will help you build & manage industrial IoT applications faster. It also makes the process much easier within a single process.
The ready to use plugins help you connect MQTT broker to the Elastic Search, Mongo DB & Redis.You can test trial these plugins by connecting MQTT Broker with any standard MQTT client. For instance, you can use the Paho Python client or Mosquitto client. Otherwise, you can also download one from the client library.
MongoDB is one of the most widely used document storage engines for IoT data analysis. This plugin connects Bevywise MQTT Broker with MongoDB. It allows the storage of received payload data from connected clients into MongoDB. It helps you handle complex data in an easy manner and for powerful analysis. The below documentation explains how to configure and setup MQTT Broker in MongoDB.
1. Open plugin.conf and configure the
[MONGO]
HOSTNAME = 127.0.0.1
PORT = 27017
DB_NAME = bevywise
COLLECTION = mqttroute
[AUTHENTICATION]
AUTHENTICATION_ENABLED = FALSE
# TRUE || FALSE
USERNAME = root
PASSWORD = root
[LOG]
LOG_FILE_PATH = ../extensions
2. Copy the folder mongo and paste it into Bevywise/MQTTRoute/extensions .
3. Copy the folder plugin.conf and paste it into Bevywise/MQTTRoute/extensions.
4. Replace custom_store.py with Bevywise/MQTTRoute/extensions/custom_store.py.
5. Open Bevywise/MQTTRoute/conf/data_store.conf.
6. Start the MQTTRoute and it will start storing all the payload into the Mongo DB server.
In the database, the Redis connector uses both clustered and nonclustered connections. Only password-based authentication is supported by Redis Connector. This Python MQTT plugin connects MQTTRoute with the Redis server. It allows to store all the payloads to the Redis server for further processing.
1. Replace”custom_store.py” with Bevywise/MQTTRoute/lib/custom_store.py.
2. In custom_store.py change the server name & port of the Redis if you are running Redis on a different server or port.
redishost=‘localhost’
redisport=6379
3.Then, Open Bevywise / MQTTRoute / conf / data_store.conf
4. Start the MQTTRoute. It will start to store all the payload into the Redis server with clientId_unixtime as the key.
MQTT Broker store data to Elastic search via custom implementation. This ensures better data visualization. The published payload only push to the Elastic search. This helps you hook it and send to your data visualization tool.
1. Open plugin.conf and configure the
[ELASTIC]
HOSTNAME = 127.0.0.1
PORT = 9200
INDEX_NAME = mqttroute
[LOG]
LOG_FILE_PATH = ../extensions
2. Copy the folder plugin.conf and paste it into Bevywise/MQTTRoute/extensions.
3. Copy the folder Elastic and paste it into Bevywise/MQTTRoute/extensions.
4. Replace custom_store.py with Bevywise / MQTTRoute / extensions / custom_store.py.
5. Open Bevywise / MQTTRoute / conf / data_store.conf.
6. Start the MQTTRoute and it will start storing all the payload into the Elastic search server.
We are with all ears waiting to hear from you. Post us with your questions and feedback.
Start Your Free Trial