Back in April I started working on an MQTT LoRa Field gateway which was going to support a selection of different Software as a service(SaaS) Internet of Things(IoT) platforms.
After a long pause in development I have a working AdaFruit.IO client and have 3 proof of concept (PoC) integrations for AskSensors, Losant and Ubidots. I am also working on Azure IoT Hub, Azure IoT Central and MyDevice Cayenne clients. The first iteration is focused on Device to Cloud (D2C) messaging in the next iteration I will add Cloud to Device where viable(C2D).
My applications use a lightweight, easy to implemented protocol which is intended for hobbyist and educational use rather than commercial applications (I have been working on a more secure version as yet another side project)
I have a number of sample Arduino with Dragino LoRa Shield for Arduino, MakerFabs Maduino, Dragino LoRa Mini Dev, M2M Low power Node and Netduino with Elecrow LoRa RFM95 Shield etc. clients. These work with both my platform specific (Adafruit.IO, Azure IoT Central) gateways and protocol specific field gateways.
When the application is first started it creates a minimal configuration file which should be downloaded, the missing information filled out, then uploaded using the File explorer in the Windows device portal.
{
"MQTTUserName": "AdaFruitIO user",
"MQTTPassword": "AIO Key",
"MQTTClientID": "MQTTLoRaGateway",
"MQTTServer": "io.adafruit.com",
"Address": "LoRaIoT1",
"Frequency": 915000000.0,
"MessageHandlerAssembly": "Mqtt.IoTCore.FieldGateway.LoRa.Adafruit",
"PlatformSpecificConfiguration": "mqttloragateway"
}
The application logs debugging information to the Windows 10 IoT Core ETW logging Microsoft-Windows-Diagnostics-LoggingChannel
The SaaS platform specific interface has gained an additional parameter for platform specific configuration.
namespace devMobile.Mqtt.IoTCore.FieldGateway
{
using System;
using Windows.Foundation.Diagnostics;
using devMobile.IoT.Rfm9x;
using MQTTnet;
using MQTTnet.Client;
public interface IMessageHandler
{
void Initialise(LoggingChannel logging, IMqttClient mqttClient, Rfm9XDevice rfm9XDevice,string platformSpecificConfiguration);
void Rfm9XOnReceive(object sender, Rfm9XDevice.OnDataReceivedEventArgs e);
void MqttApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e);
void Rfm9xOnTransmit(object sender, Rfm9XDevice.OnDataTransmitedEventArgs e);
}
}
This is used for the AdaFruit.IO GroupName so Adafruit.IO feed values are not all in a single group.
public class MessageHandler : IMessageHandler
{
private LoggingChannel Logging { get; set; }
private IMqttClient MqttClient { get; set; }
private Rfm9XDevice Rfm9XDevice { get; set; }
private string PlatformSpecificConfiguration { get; set; }
void IMessageHandler.Initialise(LoggingChannel logging, IMqttClient mqttClient, Rfm9XDevice rfm9XDevice, string platformSpecificConfiguration)
{
LoggingFields processInitialiseLoggingFields = new LoggingFields();
this.Logging = logging;
this.MqttClient = mqttClient;
this.Rfm9XDevice = rfm9XDevice;
this.PlatformSpecificConfiguration = platformSpecificConfiguration;
}
async void IMessageHandler.Rfm9XOnReceive(object sender, Rfm9XDevice.OnDataReceivedEventArgs e)
{
LoggingFields processReceiveLoggingFields = new LoggingFields();
processReceiveLoggingFields.AddString("PacketSNR", e.PacketSnr.ToString("F1"));
processReceiveLoggingFields.AddInt32("PacketRSSI", e.PacketRssi);
processReceiveLoggingFields.AddInt32("RSSI", e.Rssi);
string addressBcdText = BitConverter.ToString(e.Address);
processReceiveLoggingFields.AddInt32("DeviceAddressLength", e.Address.Length);
processReceiveLoggingFields.AddString("DeviceAddressBCD", addressBcdText);
string payloadBcdText = BitConverter.ToString(e.Data);
processReceiveLoggingFields.AddInt32("PayloadLength", e.Data.Length);
processReceiveLoggingFields.AddString("DeviceAddressBCD", payloadBcdText);
this.Logging.LogEvent("Rfm9XOnReceive", processReceiveLoggingFields, LoggingLevel.Information);
}
void IMessageHandler.MqttApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
LoggingFields processReceiveLoggingFields = new LoggingFields();
processReceiveLoggingFields.AddString("ClientId", e.ClientId);
#if DEBUG
processReceiveLoggingFields.AddString("Payload", e.ApplicationMessage.ConvertPayloadToString());
#endif
processReceiveLoggingFields.AddString("QualityOfServiceLevel", e.ApplicationMessage.QualityOfServiceLevel.ToString());
processReceiveLoggingFields.AddBoolean("Retain", e.ApplicationMessage.Retain);
processReceiveLoggingFields.AddString("Topic", e.ApplicationMessage.Topic);
this.Logging.LogEvent("MqttApplicationMessageReceived topic not processed", processReceiveLoggingFields, LoggingLevel.Error);
}
void IMessageHandler.Rfm9xOnTransmit(object sender, Rfm9XDevice.OnDataTransmitedEventArgs e)
{
this.Logging.LogMessage("Rfm9xOnTransmit", LoggingLevel.Information);
}
}
The message handler uploads all values in an inbound messages in one MQTT message using the AdaFruit.IO Group Feed format.
async void IMessageHandler.Rfm9XOnReceive(object sender, Rfm9XDevice.OnDataReceivedEventArgs e)
{
LoggingFields processReceiveLoggingFields = new LoggingFields();
char[] sensorReadingSeparators = { ',' };
char[] sensorIdAndValueSeparators = { ' ' };
processReceiveLoggingFields.AddString("PacketSNR", e.PacketSnr.ToString("F1"));
processReceiveLoggingFields.AddInt32("PacketRSSI", e.PacketRssi);
processReceiveLoggingFields.AddInt32("RSSI", e.Rssi);
string addressBcdText = BitConverter.ToString(e.Address);
processReceiveLoggingFields.AddInt32("DeviceAddressLength", e.Address.Length);
processReceiveLoggingFields.AddString("DeviceAddressBCD", addressBcdText);
string messageText;
try
{
messageText = UTF8Encoding.UTF8.GetString(e.Data);
processReceiveLoggingFields.AddString("MessageText", messageText);
}
catch (Exception ex)
{
processReceiveLoggingFields.AddString("Exception", ex.ToString());
this.Logging.LogEvent("PayloadProcess failure converting payload to text", processReceiveLoggingFields, LoggingLevel.Warning);
return;
}
// Chop up the CSV text
string[] sensorReadings = messageText.Split(sensorReadingSeparators, StringSplitOptions.RemoveEmptyEntries);
if (sensorReadings.Length < 1)
{
this.Logging.LogEvent("PayloadProcess payload contains no sensor readings", processReceiveLoggingFields, LoggingLevel.Warning);
return;
}
JObject payloadJObject = new JObject();
JObject feeds = new JObject();
// Chop up each sensor read into an ID & value
foreach (string sensorReading in sensorReadings)
{
string[] sensorIdAndValue = sensorReading.Split(sensorIdAndValueSeparators, StringSplitOptions.RemoveEmptyEntries);
// Check that there is an id & value
if (sensorIdAndValue.Length != 2)
{
this.Logging.LogEvent("PayloadProcess payload invalid format", processReceiveLoggingFields, LoggingLevel.Warning);
return;
}
string sensorId = string.Concat(addressBcdText, sensorIdAndValue[0]);
string value = sensorIdAndValue[1];
feeds.Add(sensorId.ToLower(), value);
}
payloadJObject.Add("feeds", feeds);
string topic = $"{MqttClient.Options.Credentials.Username}/groups/{PlatformSpecificConfiguration}";
try
{
var message = new MqttApplicationMessageBuilder()
.WithTopic(topic)
.WithPayload(JsonConvert.SerializeObject(payloadJObject))
.WithAtLeastOnceQoS()
.Build();
Debug.WriteLine(" {0:HH:mm:ss} MQTT Client PublishAsync start", DateTime.UtcNow);
await MqttClient.PublishAsync(message);
Debug.WriteLine(" {0:HH:mm:ss} MQTT Client PublishAsync finish", DateTime.UtcNow);
this.Logging.LogEvent("PublishAsync Adafruit payload", processReceiveLoggingFields, LoggingLevel.Information);
}
catch (Exception ex)
{
processReceiveLoggingFields.AddString("Exception", ex.ToString());
this.Logging.LogEvent("PublishAsync Adafruit payload", processReceiveLoggingFields, LoggingLevel.Error);
}
}
The casing of User names (Must match exactly) and Group/Feed names (must be lower case) tripped me up yet again. The “automagic” provisioning of feeds does make setting up small scale systems easier, though I’m not certain how well it would scale.
