Create Your Own IIoT Tech Stack Project | Part 1: MQTT Client Setup

There is quite a bit of complexity that goes into building Industrial IoT solutions for use in production. The solution must span many different technologies that live in both the OT and IT worlds.

Generally, a full IIoT project begins at the “edge” in the form of low-level hardware residing by the physical system. The final destination point is often some form of real-time dashboard hosted on an IT server that can be accessed by engineers, technicians, managers, and operators. The data is also often used to predict or analyze possible points of optimization or troubleshooting for a more efficient plant operation.

Some examples of dashboarding tools include Ignition Vision, ThingWorx, and Grafana. The flow of data from the edge device to the dashboard passes through many supporting applications along its journey. These can include databases and data pipeline tools. The taxi transporting the data is usually a protocol such as MQTT, HTTP, or Modbus TCP.

 

Figure 1. IoT tools bucketed into 4 categories. Image used courtesy of Zipit

 

IoT Solution Components

This solution will ride on the coattails of my prior article series demonstrating the setup of the Raspberry Pi 4 and the basic utilization of the Sense HAT. It will also utilize the paho-mqtt library which I displayed in an older article using a mock python script. This series will showcase how to blend these 2 Python libraries together to illustrate a useful, functioning real-world IoT device. Yes, no more simulated or virtual devices!

This series will venture into potentially intimidating new territory, such as Docker containers, networking, and databases. If some of these technologies are unfamiliar, not a problem! I personally had not worked with several of the technologies before writing this series, so we are all learning together. A high-level overview of the technology stack can be seen below. Notice the sheer variety of tools required.

 

Figure 2. Tech stack overview. Image used courtesy of the author

 

Jumping-Off Point

As with any complex project, once the scope is roughly defined, start small. This article will focus on the edge layer of the solution, that is, the layer that exists on the machine that is directly connected to the actuator and input devices, on the edge of physical and virtual.

The hardware at this layer will be composed of a Raspberry Pi 4 and Sense HAT (the recently-announced Raspberry Pi 5 should also work, but as with all new updates, not every component and application has been tested in all configurations). The Pi will be modeled as an MQTT client publishing Sense HAT sensor data to an MQTT broker. There is a separate series dedicated to these sensor devices on Control.com. If you haven’t already, please read those articles if you are interested in following along, as I will assume you have set up the Raspberry Pi and Sense HAT as outlined in those articles. Perfect!

To start, we will use the SSH protocol to remote into the Raspberry Pi. Within the remote terminal, we will create a new directory and a Python file within that directory to house the code for the Pi.

 

Figure 3. Example of a new directory and Python file being created in a remote terminal window via SSH protocol. Image used courtesy of the author

 

To begin the script, we will open the Nano editor by entering the command followed by enter/return:

nano publisher.py

 

In the Nano editor window, initialize the script with the import of the required modules:

import paho.mqtt.client as mqtt
from sense_hat import SenseHat
import socket
import json
import time

 

SenseHat and Callback Function

Here’s where things get interesting in combining the sense_hat and paho-mqtt libraries. We’ll begin by initializing the Raspberry Pi as a SenseHat() class instance. However, we’ll proceed by defining a callback function that will be utilized by the paho-mqtt library.

Within this callback function called on_connect(), we are building instructions for how the Raspberry Pi and Sense HAT will behave when connecting to an MQTT broker. If the connection is successful, the LED matrix will flash green 3 times. If another rc code is returned, the LED matrix will flash yellow 3 times.

sense = SenseHat()

# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, rc):

    print("Connected with result code "+str(rc))
    green = (0,255,0)
    yellow = (255,255,0)
    if rc == 0:
        for _ in range(3):
            sense.clear(green)
            time.sleep(0.5)
            sense.clear()
            time.sleep(0.5)
    else:
        for _ in range(3):
            sense.clear(yellow)
            time.sleep(0.5)
            sense.clear()
            time.sleep(0.5)

 

Connecting to the MQTT Broker

We’ll start by defining the connect function and supplying input parameters. In an attempt to make the script modular, we’ll parameterize the topic, broker URI, and hostname of the device.

The first two lines within the definition block in the code below will instantiate the Pi as an MQTT client and tell the client to use the previously defined ‘on_connect’ callback function when connecting to the broker.

The try block attempts to connect the device to the broker and maintain that network connection with the loop_start() method. If a TCP connection cannot be made (if the broker is unreachable or down), we’ll receive an [Errno 111] Connection refused instead of an rc code. Therefore, the exception will be handled by flashing red 3 times on the sense HAT and returning the full traceback.

def connect(topic: str, broker: str, hostname = socket.gethostname()):
    client = mqtt.Client()
    client.on_connect = on_connect
    try:
        client.connect(broker, 1883)
        client.loop_start()
        return client
    except:
        print("Cannot establish connection to %s" % (broker))
        red = (255,0,0)
        for _ in range(3):
            sense.clear(red)
            time.sleep(0.5)
            sense.clear()
            time.sleep(0.5)
        raise

 

Data Payload pushData Function

The previous function is used to instantiate the client and connection in the first line of the pushData function. Next, we actually need to incorporate the sense HAT environment sensors in order to build a real-world data payload. We’ll use the sense_hat library’s “get” functions for temperature, humidity, and pressure to get that real-time environment data.

The environmental data will be packaged as a JSON payload given by the variable called payload. This payload will then be published to the broker on the specified topic with a quality of service (qos) of 0. This indicates that the message will be fired to the broker “at most once”; it is the least robust qos where message loss is acceptable.

 def pushData(topic: str, broker: str, hostname = socket.gethostname()):
    client = connect(topic, broker)
    scan = 1
    while True:

        temp = sense.get_temperature()
        pressure = sense.get_pressure()
        humidity = sense.get_humidity()

        payload = json.dumps(
            {
            "hostname": str(hostname),
            "temperature": str(temp),
            "pressure": str(pressure),
            "humidity": str(humidity)
        }
        )

        client.publish(topic, payload, qos=0)
        scan += 1
        time.sleep(60)

 

Creating a Config File

We’ve been calling parameterized variables throughout the script thus far but haven’t defined these variables yet. In consideration of the modularity, wouldn’t it be great if we stored our secrets or keys in a single file? This would make any changes to the broker or topic over time simple to manage.

We can consolidate these parameters by leveraging a configuration file. For now, we will just define the function that will read that configuration file and return the values associated with our keys. The config.json file will be covered in the next article.

def config(file: str = "config.json"):
    with open(file, "r") as jsonfile:
        data = json.load(jsonfile)
        topic = data["topic"]
        broker = data["broker"]
        print("Configuration file read successfully")
        return topic, broker

 

Triggering the Data Push Event

An ‘if’ statement ensures that the script executes when called directly, and not when imported as a module in an external script.

When triggered, the config() function is called. It uses the default “config.json” parameter to pull the topic and broker values from the file. The pushData() function is then called and supplied the key values for the broker and topic. This begins the broker connection and publishing of data to the broker.

if __name__ == "__main__":
    topic, broker = config()
    pushData(topic,broker)

 

At this point in the operation, the edge computer (the Raspberry Pi) is prepared to push a packet of environmental data from sensors to a provided MQTT broker.

There’s some complexity behind provisioning an MQTT broker, so in the next article in this series (Part 2), we’ll use open, free sandbox brokers to test our script. We’ll also cover the config.json file to get our publisher script up and running. This will help us take one step at a time in the process.