#1 MQTT client for Arduino-based PLC as an I/Os module

In chapter #1 is explained how to configure your Industrial Arduino-based PLC as a MQTT I/Os Module.

Before starting this chapter, it would be interesting to take a look at how MQTT protocol works, what is JSON and their applications.

After that, take a fast look at the next libraries:

MQTT >>

ArduinoJson Library >>

Note: In order to work properly with our devices, install version 5.13.5 of the Arduino JSON Library. A newer version could present some issues.

The architecture of the code is made for an M-Duino21+, but is designed to be extended for other M-Duino's. There are four different arrays that include all available I/Os in M-Duino 21+. These arrays are:

• digitalOutputs
• analogOutputs
• digitalInputs
• analogInputs

So, every array will be structured as follows:

digitalOutputs[NUM_ZONES][NUM_DIGITAL_OUTPUTS_PER_ZONE] = {{Q0_0, Q0_1... }, {Q1_0, Q1_1...}, {Q2_0, Q2_1...}

This last example will be for an M-Duino 58+. Every zone is subdivided with their related pins nomenclature.

After the explanation of how to access to the mapping of the inputs, we are ready to explain the rest of the code.

In setup(), we are just making the initialization of the Ethernet and MQTT client called MQTT, relating the callback function when we receive a message through MQTT and all the actual input values. After that we just have three main functions in the loop() function. We are calling the reconnect() function if there is no connection with the server, if there is a connection we are just using the mqtt.loop() function that will maintain the connectivity with our server. Apart from these two functions, we have the updateInputs() that will update the real value of the inputs.

Let's go deeper into the functions to have a better understating of what they do.

Outputs Management

First of all, we have the reconnect() function that will subscribe to the "Q" and "A" topics. One topic is dedicated to Analog outputs and the other is dedicated to Digital outputs. If is not connected to the server, then they will stop the client connection in order to not cause any issue with the server ports.

Secondly, to understand how the outputs control works we need to look into the callback function. In this program, the callback function is receiveMqttMessage(). The receiveMqttMessage() function will run every time that we receive a message. So, when we receive a message the callback function will decide if the topic is for Analog or Digital pins and then will call the right function for it, setDigitalOutput() function or setAnalgoOutput() function.

setDigitalOutput() extract and analyzes the JSON file, confirm that the zone and the pin exist and then execute the digitalWrite() function to update the desired output.

The setAnalogOutput() function works exactly in the same way as setDigitalOutput().

ANALOG_OUTPUTS_OFFSET is used to offset the value of the analog pins. Analog pins are placed in positions  5, 6, 7. This is why the ANALOG_OUTPUTS_OFFSET is 5.

Inputs Management

In the main loop, we have the function updateInputs(), that will update all the values of the inputs. This function will be comparing all the previous values to the current ones and if the new value has changed, then the program will publish the new value through MQTT. The Digital values will be comparing if there is a 0 or a 1. The function that makes this work is updateDigitalInput().

In the Analog side, the program will compare if the Analog value has changed more than 5 points. If the current value is higher or lower than the previous one, then it will publish the new value. The function that makes this work is updateAnalogInput().

Note: Make sure that the ArduinoJson library is in version 5. Go to Sketch > Include Library > Manage Libraries… > Type: ArduinoJson and make sure the version 5 is installed.

Code Example

/*
   Copyright (c) 2019 Boot&Work Corp., S.L. All rights reserved
   This program is free software: you can redistribute it and/or modify
   it under the terms of the GNU Lesser General Public License as published by
   the Free Software Foundation, either version 3 of the License, or
   (at your option) any later version.
   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU Lesser General Public License for more details.
   You should have received a copy of the GNU Lesser General Public License
   along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
// including Ethernet library
#include <Ethernet.h>
// libraries needed for MQTT communication
#include <ArduinoJson.h>
#include <PubSubClient.h>
#define MQTT_ID "demo"
#define NUM_ZONES 1
#define NUM_DIGITAL_OUTPUTS_PER_ZONE 5
#define DIGITAL_OUTPUTS_OFFSET 0
const int digitalOutputs[NUM_ZONES][NUM_DIGITAL_OUTPUTS_PER_ZONE] = {
  {Q0_0, Q0_1, Q0_2, Q0_3, Q0_4},
};
#define NUM_ANALOG_OUTPUTS_PER_ZONE 3
#define ANALOG_OUTPUTS_OFFSET 5
const int analogOutputs[NUM_ZONES][NUM_DIGITAL_OUTPUTS_PER_ZONE] = {
  {A0_5, A0_6, A0_7},
};
#define NUM_DIGITAL_INPUTS_PER_ZONE 7
#define DIGITAL_INPUTS_OFFSET 0
const int digitalInputs[NUM_ZONES][NUM_DIGITAL_INPUTS_PER_ZONE] = {
  {I0_0, I0_1, I0_2, I0_3, I0_4, I0_5, I0_6},
};
#define NUM_ANALOG_INPUTS_PER_ZONE 6
#define ANALOG_INPUTS_OFFSET 7
#define ANALOG_INPUTS_THRESHOLD 5 // Filtering threshold
const int analogInputs[NUM_ZONES][NUM_ANALOG_INPUTS_PER_ZONE] = {
  {I0_7, I0_8, I0_9, I0_10, I0_11, I0_12},
};

byte mac[] = { 0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xAE };
IPAddress broker(10, 10, 10, 60);
/* Set an IP address for the MQTT-Broker. Remember to set the same static IP address in the device
set as a broker.*/
IPAddress ip(10, 10, 10, 70); //SET HERE THE IP ADDRESS FOR YOUR M-DUINO
unsigned port = 1883;
// Initialize client
EthernetClient client;
PubSubClient mqtt(client);
int digitalInputsValues[NUM_ZONES][NUM_DIGITAL_INPUTS_PER_ZONE];
int analogInputsValues[NUM_ZONES][NUM_ANALOG_INPUTS_PER_ZONE];
////////////////////////////////////////////////////////////////////////////////////////////////////
void setup(){
  Ethernet.begin(mac, ip);
  mqtt.setServer(broker, port);
  mqtt.setCallback(receiveMqttMessage);
  // Init variables
  for (int i = 0; i < NUM_ZONES; ++i) {
    for (int j = 0; j < NUM_DIGITAL_INPUTS_PER_ZONE; ++j) {
      digitalInputsValues[i][j] = digitalRead(digitalInputs[i][j]);
    }
    for (int j = 0; j < NUM_ANALOG_INPUTS_PER_ZONE; ++j) {
      analogInputsValues[i][j] = analogRead(analogInputs[i][j]);
    }
  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void loop() {
  if (!mqtt.connected()) {
    reconnect();
  } else {
    mqtt.loop();
  }
  updateInputs();
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void updateInputs() {
  for (int i = 0; i < NUM_ZONES; ++i) {
    for (int j = 0; j < NUM_DIGITAL_INPUTS_PER_ZONE; ++j) {
      updateDigitalInput(i, j);
    }
    for (int j = 0; j < NUM_ANALOG_INPUTS_PER_ZONE; ++j) {
      updateAnalogInput(i, j);
    }
  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void updateDigitalInput(int zone, int index) {
  int value = digitalRead(digitalInputs[zone][index]);
  if (value != digitalInputsValues[zone][index]) {
    digitalInputsValues[zone][index] = value;
    publishInput(zone, index + DIGITAL_INPUTS_OFFSET, value);
  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void updateAnalogInput(int zone, int index) {
  int value = analogRead(analogInputs[zone][index]);
  int minValue = value > ANALOG_INPUTS_THRESHOLD ? value - ANALOG_INPUTS_THRESHOLD : 0;
  int maxValue = value < 1023 - ANALOG_INPUTS_THRESHOLD ? value + ANALOG_INPUTS_THRESHOLD : 1023;
  if ((analogInputsValues[zone][index] < minValue) || (analogInputsValues[zone][index] > maxValue)) {
    analogInputsValues[zone][index] = value;
    publishInput(zone, index + ANALOG_INPUTS_OFFSET, value);
  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void setDigitalOutput(char *payload, unsigned int len) {
  DynamicJsonBuffer json(JSON_OBJECT_SIZE(3));
  JsonObject &root = json.parseObject(payload, len);
  if (root.success()) {
    int zone = root["zone"];
    if (zone >= NUM_ZONES) {
      // Invalid zone
      return;
    }
    int index = root["index"];
    index -= DIGITAL_OUTPUTS_OFFSET;
    if (index >= NUM_DIGITAL_OUTPUTS_PER_ZONE) {
      // Invalid digital output
      return;
    }
    int value = root["value"];
    digitalWrite(digitalOutputs[zone][index], value);
  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void setAnalogOutput(char *payload, unsigned int len) {
  DynamicJsonBuffer json(JSON_OBJECT_SIZE(3));
  JsonObject &root = json.parseObject(payload, len);
  if (root.success()) {
    int zone = root["zone"];
    if (zone >= NUM_ZONES) {
      // Invalid zone
      return;
    }
    int index = root["index"];
    index -= ANALOG_OUTPUTS_OFFSET;
    if (index >= NUM_ANALOG_OUTPUTS_PER_ZONE) {
      // Invalid analog output
      return;
    }
    int value = root["value"];
    analogWrite(analogOutputs[zone][index], value);
  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void reconnect() {
  if (mqtt.connect(MQTT_ID)) {
    mqtt.subscribe("Q");
    mqtt.subscribe("A");
  } else {
    // MQTT connect fail
    client.stop();
  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void receiveMqttMessage(char* topic, byte* payload, unsigned int len) {
  if (strcmp(topic, "Q") == 0) {
    // Set digital output
    setDigitalOutput((char*) payload, len);
  } else if (strcmp(topic, "A") == 0) {
    // Set analog output
    setAnalogOutput((char*) payload, len);
  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void publishInput(int zone, int index, int value) {
  DynamicJsonBuffer json(JSON_OBJECT_SIZE(3));
  JsonObject &root = json.createObject();
  if (root.success()) {
    root["zone"] = zone;
    root["index"] = index;
    root["value"] = value;
    publish("I", root);
  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void publish(const char *topic, JsonObject &root) {
  unsigned len = root.measureLength();
  if (len > 0) {
    char *payload = new char[len + 1];
    if (payload) {
      root.printTo(payload, len + 1);
      publish(topic, payload);
      delete[] payload;
    }
  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////
void publish(const char *topic, const char *payload) {
  if (mqtt.connected()) {
    mqtt.publish(topic, payload);
  }
}