Skip to main content

An Edge Compute IIoT Demonstrator Part 3: Software

Brainboxes BB-400 Smart Controller

Implementing simple process control with Python and a graphical dashboard with Node-RED.

In Part 1 of this series, we introduced the demonstrator platform hardware configuration, while Part 2 covered the design, simulation and construction of a custom envelope follower module for signal conditioning. In the final article in this series, we now take a look at very simple software which serves to exercise the hardware and demonstrate how to interface with it.

Hardware configuration

Brainboxes BB-400 Hardware configuration

The Brainboxes BB-400 provides a convenient web interface for configuration and this was used to label digital IO and set the mode. The web UI can also be used to check the state of digital inputs and to set the state of outputs.

Brainboxes protocol menu

Brainboxes remote IO modules also have a web interface for configuration and above can be seen the one for the ED-538 digital input and relay output module.

Brainboxes status on analogue inputs

Here we can see the status of the analogue inputs on the ED-549 module.

The remote IO modules and BB-400 are all connected to private LAN, via the DIN rail Ethernet switch fitted inside the demonstrator.

Main application

Since the Brainboxes BB-400 is a full blown Linux computer we are spoilt for choice when it comes to programming languages, libraries and frameworks etc. However, the Python programming language was chosen for the main control application, as this is particularly easy to understand.

When it comes to communicating with IO modules we have the option of the ASCII protocol or Modbus TCP. The BB-400 meanwhile supports ASCII protocol or a REST (HTTP) API. Hence it was decided to use ASCII protocol, as this is supported by both the IO modules and BB-400. ASCII is a pretty simple text-based query/response protocol, has been around for quite some time and would have originally been used over serial connections.

Python modules

import sys
import time
import json
import brainboxes
import paho.mqtt.publish as publish
from binascii import unhexlify

The application uses a number of built-in Python modules, such as sys and time. Brainboxes provide a Python module which facilitates communication with their remote IO modules. We’ll be using MQTT to publish measurements via the Mosqtuitto MQTT broker, which is also installed on the BB-400. Finally, the binascii module will be used to return the binary string represented by hexadecimal string, so that we can see check to see which bits have been set by an IO module.

# MQTT configuration
mqtt_broker = 'localhost'
sensor_topic = 'conveyor/sensors'
state_topic = 'conveyor/state'

# ASCII TCP servers
bb400ip = ''
ed538ip = ''
ed549ip = ''

# Sensor constants

volts_mult = 3
amps_mult = 0.5
temp_mult = 10
vib_mult = 10
meas_dp = 2

# Beacon amber > green transition delay (s)

beaconDelay = 2

After importing the required modules we then define a number of variables, such as the address of the MQTT broker and topics to use, addresses of IO modules, and multiplier constants for calculating voltage, current, temperature and vibration.

Hardware interfacing

def isBitSet(resp, pos):
    intval = int.from_bytes(unhexlify((resp[3:])), byteorder='big')
    return intval & 1 << pos != 0

A function is defined which will parse a hex value returned by an IO module and then check to see if a bit has been set, which will tell us if a digital input is high or low.

def readBB400():
        with brainboxes.AsciiIo(ipaddr=bb400ip, port=9500, timeout=1.0) as bb400:
            rxdata = bb400.command_response(b'@01')
        print('  *** BB-400 read: Failed to connect!')

    if rxdata is None:
        print('  *** BB-400 read: No response received!')

    elif len(rxdata) != 5:
        print('  *** BB-400 read: Bad response received!')

        pb_stop = isBitSet(rxdata, 1)
        pb_start = isBitSet(rxdata, 0)
        return(pb_stop, pb_start)

The readBB400() function posts the @01 command to the BB-400 IO server ASCII protocol port, with the parsed response telling us if the machinery start or stop button is pressed.

def readED538():
        with brainboxes.AsciiIo(ipaddr=ed538ip, port=9500, timeout=1.0) as ed538:
            rxdata = ed538.command_response(b'@01')
        print('  *** ED-538 read: Failed to connect!')

    if rxdata is None:
        print('  *** ED-538 read: No response received!')

    elif len(rxdata) != 5:
        print('  *** ED-538 read: Bad response received!')

        enable = isBitSet(rxdata, 0)
        pb_volt = isBitSet(rxdata, 4)
        pb_cur = isBitSet(rxdata, 5)
        pb_temp = isBitSet(rxdata, 6)
        pb_vib = isBitSet(rxdata, 7)
        return(enable, pb_volt, pb_cur, pb_temp, pb_vib)

A similar function is defined for reading the ED-538 module inputs so that we can see if the machine is enabled (its safety relay is not tripped) and if any of the fault simulation buttons have been pressed. It should be noted that at this point fault simulation itself has not been implemented, but the buttons and indicator are available for use in projects.

def readED549():
        with brainboxes.AsciiIo(ipaddr=ed549ip, port=9500, timeout=1.0) as ed549:
            rxdata = ed549.command_response(b'#01')
        print('  *** ED-549 read: Failed to connect!')

    if rxdata is None:
        print('  *** ED-549 read: No response received!')

    elif len(rxdata) != 57:
        print('  *** ED-549 read: Bad response received!')

        sup_volts = round(float(rxdata[2:6]) * volts_mult, meas_dp)
        sup_amps = round(float(rxdata[9:15]) * amps_mult, meas_dp)
        motor_tempc = round(float(rxdata[16:22]) * temp_mult, meas_dp)
        motor_vib = round(float(rxdata[23:27]) * vib_mult, meas_dp)
        return(sup_volts, sup_amps, motor_tempc, motor_vib)

The ED-549 module provides analogue measurement inputs and these are configured for 0-10v input. When queried this returns a long string with measurements for each channel. Hence we slice the string at appropriate points to select the value of interest, before turning this into a float, multiplying to get the correct units value, and then finally reducing it to two decimal places.

def getRun():
        stop, start = readBB400()
        if stop == False and start == True:
            return False
        elif stop == True and start == False:
            return True
        print('  *** getRun: Failed to read BB-400!')

We have a simple helper function which only returns True if the start button is pressed and not the stop button.

def setBeacon(colour):
    if colour == 'red':
        msgs = (b'#011600', b'#011500', b'#011401')
    elif colour == 'amber':
        msgs = (b'#011600', b'#011400', b'#011501')
    elif colour == 'green':
        msgs = (b'#011400', b'#011500', b'#011601')
        print('  *** BB-400 setBeacon invalid colour: {0}'.format(colour))

        with brainboxes.AsciiIo(ipaddr=bb400ip, port=9500, timeout=1.0) as bb400:
            for txdata in msgs:
                data = bb400.command_noresponse(txdata)
        print('  *** BB-400 setBeacon: Failed to connect!')

The three-colour tower beacon is wired to BB-400 digital outputs and a function is defined which takes an argument of red, amber or green to set this accordingly.

def setConveyor(state):
    if state == 'stop':
        txdata = b'#010A00'
    elif state == 'run':
        txdata = b'#010A01'
        print('  *** ED-538 setConveyor invalid state: {0}'.format(state))

        with brainboxes.AsciiIo(ipaddr=ed538ip, port=9500, timeout=1.0) as ed538:
            data = ed538.command_noresponse(txdata)
        print('  *** ED-538 setConveyor: Failed to connect!')

The coil of the contactor for the conveyor motor is connected to a digital output on the BB-400 and there is similarly a function defined which starts or stops this, by passing run or stop.

Process state

def setStep():
    global step, tick
    print('  Running step: {0}'.format(step))

    if step == 0:
    elif step == 1:
    elif step == 2:
        step = 0

    tick = time.time()

The process is incredibly simple and the conveyor has states of stopped, starting and running. We set the process step by calling the setStep() function, which configures the beacon colour and starts or stops the conveyor.


Prior to entering the main loop, we call this function once to set the process to step 0.

Main loop

while True:
        # Save previous run state
        prevRunst = runst

        # Read ED-538 inputs (enable circuit and fault push buttons)
            enable, pb_volt, pb_cur, pb_temp, pb_vib = readED538()
            print('  *** Failed to read ED-538')

        # Set the enable state
        if enable != None:
            enst = enable

        # Read and obey start/stop buttons if we're enabled
        if enst == True:
            run = getRun()
            if run != None:
                runst = run
            runst = False

        # Stop if required
        if runst == False and prevRunst == True:
            print('  Stopping! Run: {0} // Enable: {1}'.format(runst, enst))
            runst = False
            step = 0

       # Run the plant process
        if runst == True:
            if step == 0:
                step = 1

            elif step == 1:
                if time.time()-tick > beaconDelay:
                    step = 2

            print('  Run: {0} // Enable: {1}'.format(runst, enst))

In the main loop we first:

  1. Save the current run state
  2. Check and set the enable state
  3. Read the start/stop buttons and set the new run state
  4. Stop the conveyor if the state has transitioned to stopped
  5. Run the plant process


In running the plant process we check to see if we are set to run and if so and the current step is 0 (stopped/red), we advance to step 1 (starting/amber). If already in the starting state we check to see if sufficient time has elapsed since transitioning from step 0, before then progressing to step 2 (running/green). If we are already at step 2 we just pass.

            volts, amps, tempc, vib = readED549()
            print('  *** Failed to read ED-549!')

        # Fomat process data and publish
            process_state = {
                'enable_state': enst,
                'run_state': runst,
                'step': step,
                'voltage_fault': pb_volt,
                'current_fault': pb_cur,
                'temperature_fault': pb_temp,
                'vibration_fault': pb_vib

            process_sensors = {
                'supply_voltage': volts,
                'supply_current': amps,
                'motor_temperature': tempc,
                'motor_vibration': vib

            msgs = [{'topic':state_topic, 'payload':json.dumps(process_state)},
                (sensor_topic, json.dumps(process_sensors), 0, False)]

            publish.multiple(msgs, hostname=mqtt_broker, client_id='conveyor')
            print('  *** Failed to publish to MQTT broker!')

Next, we read the analogue inputs and construct a JSON object with key/value pairs. A second object is also constructed with key/value pairs for the enable and run states, plus fault push button states. Both JSON objects are then published to MQTT topics.


Finally, we pause before returning to the beginning of the loop.


Node-RED Dashboard

It was decided to use Node-RED to create a simple web dashboard with sensor readings. We could have also used Python for this, or alternatively used Node-RED for both process control and the dashboard. However, using a combination of the two demonstrates how different programming languages and tools can be used together, where some might excel in one area or another, or be supplied by different vendors, for example.

In our case, since we have sensor measurements available via MQTT, it is very easy indeed to construct a simple web interface. As can be seen above, we have a node which subscribes to the relevant MQTT topic, with messages then being converted from a JSON string to an object.

volts = {payload: msg.payload.supply_voltage};
amps = {payload: msg.payload.supply_current};
temperature = {payload: msg.payload.motor_temperature};
vibration = {payload: msg.payload.motor_vibration};

return [volts, amps, temperature, vibration];

Following this we have a function node with multiple outputs, which splits out measurements.

Each of the measurements is routed to a Gauge node

Finally, each of the measurements is routed to a Gauge node.

Resultant Dashboard

Above we can see the resulting simple dashboard.

Both the Python code and Node-RED flow have been published to GitHub.

Potential improvements

As we've seen this software is extremely simple and as previously noted it was developed for the purposes of exercising the hardware and demonstrating how to interface with it — and there is clearly a great deal of room for improvement. For example, by implementing fault simulation and much more sophisticated MQTT functionality, or perhaps a web API for integrating with other systems.

Providing sensor sampling rates are not too high it should also be possible to host a time series database on the BB-400, such as InfluxDB, along with an analytics web interface such as Grafana. Although it would likely make more sense to host such capabilities elsewhere and focus on more typical edge functionality, such as local processing and alerting.

In any case, there is plenty of scope for the students and researchers who will be making use of the platform!

  — Andrew Back

Open source (hardware and software!) advocate, Treasurer and Director of the Free and Open Source Silicon Foundation, organiser of Wuthering Bytes technology festival and founder of the Open Source Hardware User Group.
DesignSpark Electrical Logolinkedin