The Things Network MQTT & Azure IoT Part2

Uplink with decoded_payload & frm_payload

The next functionality added to my Proof of Concept(PoC) Azure IoT Hub, The Things Network(TTN) V3 Hypertext Transfer Protocol(HTTP) client API Integration, and Message Queue Telemetry Transport (MQTT) Data API Integration is sending of raw and decoded uplink messages to an Azure IoT Hub.

// At this point all the AzureIoT Hub deviceClients setup and ready to go so can enable MQTT receive
mqttClient.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(e => MqttClientApplicationMessageReceived(e)));

// This may shift to individual device subscriptions
string uplinkTopic = $"v3/{options.MqttApplicationID}/devices/+/up";
await mqttClient.SubscribeAsync(uplinkTopic, MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce);

//string queuedTopic = $"v3/{options.MqttApplicationID}/devices/+/queued";
//await mqttClient.SubscribeAsync(queuedTopic, MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce);

The additional commented out subscriptions are for the processing of downlink messages

The MQTTNet received message handler uses the last segment of the topic to route messages to a method for processing

private static async void MqttClientApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e)
{
	if (e.ApplicationMessage.Topic.EndsWith("/up", StringComparison.InvariantCultureIgnoreCase))
	{
		await UplinkMessageReceived(e);
	}

	/*
	if (e.ApplicationMessage.Topic.EndsWith("/queued", StringComparison.InvariantCultureIgnoreCase))
	{
		await DownlinkMessageQueued(e);
	}
	...			
	*/
}

The UplinkMessageReceived method deserialises the message payload, retrieves device context information from the local ObjectCache, adds relevant uplink messages fields (including the raw payload), then if the message has been unpacked by a TTN Decoder, the message fields are added as well.

static async Task UplinkMessageReceived(MqttApplicationMessageReceivedEventArgs e)
{
	try
	{
		PayloadUplinkV3 payload = JsonConvert.DeserializeObject<PayloadUplinkV3>(e.ApplicationMessage.ConvertPayloadToString());
		string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
		string deviceId = payload.EndDeviceIds.DeviceId;
		int port = payload.UplinkMessage.Port;
...
		DeviceClient deviceClient = (DeviceClient)DeviceClients.Get(deviceId);
		if (deviceClient == null)
		{
			Console.WriteLine($" UplinkMessageReceived unknown DeviceID: {deviceId}");
			return;
		}

		JObject telemetryEvent = new JObject();
		telemetryEvent.Add("DeviceID", deviceId);
		telemetryEvent.Add("ApplicationID", applicationId);
		telemetryEvent.Add("Port", port);
		telemetryEvent.Add("PayloadRaw", payload.UplinkMessage.PayloadRaw);

		// If the payload has been unpacked in TTN backend add fields to telemetry event payload
		if (payload.UplinkMessage.PayloadDecoded != null)
		{
			EnumerateChildren(telemetryEvent, payload.UplinkMessage.PayloadDecoded);
		}

		// Send the message to Azure IoT Hub/Azure IoT Central
		using (Message ioTHubmessage = new Message(Encoding.ASCII.GetBytes(JsonConvert.SerializeObject(telemetryEvent))))
		{
			// Ensure the displayed time is the acquired time rather than the uploaded time. 
			//ioTHubmessage.Properties.Add("iothub-creation-time-utc", payloadObject.Metadata.ReceivedAtUtc.ToString("s", CultureInfo.InvariantCulture));
			ioTHubmessage.Properties.Add("ApplicationId", applicationId);
			ioTHubmessage.Properties.Add("DeviceId", deviceId);
			ioTHubmessage.Properties.Add("port", port.ToString());

			await deviceClient.SendEventAsync(ioTHubmessage);
		}
	}
	catch( Exception ex)
	{
		Debug.WriteLine("UplinkMessageReceived failed: {0}", ex.Message);
	}
}

private static void EnumerateChildren(JObject jobject, JToken token)
{
	if (token is JProperty property)
	{
		if (token.First is JValue)
		{
			// Temporary dirty hack for Azure IoT Central compatibility
			if (token.Parent is JObject possibleGpsProperty)
			{
				if (possibleGpsProperty.Path.StartsWith("GPS_", StringComparison.OrdinalIgnoreCase))
				{
					if (string.Compare(property.Name, "Latitude", true) == 0)
					{
						jobject.Add("lat", property.Value);
					}
					if (string.Compare(property.Name, "Longitude", true) == 0)
					{
						jobject.Add("lon", property.Value);
					}
					if (string.Compare(property.Name, "Altitude", true) == 0)
					{
						jobject.Add("alt", property.Value);
					}
				}
			}
			jobject.Add(property.Name, property.Value);
		}
		else
		{
			JObject parentObject = new JObject();
			foreach (JToken token2 in token.Children())
			{
				EnumerateChildren(parentObject, token2);
				jobject.Add(property.Name, parentObject);
			}
		}
	}
	else
	{
		foreach (JToken token2 in token.Children())
		{
			EnumerateChildren(jobject, token2);
		}
	}
}

There is also some basic reformatting of the messages for Azure IoT Central

TTN Simulate uplink message with GPS location payload.
Nasty console application processing uplink message
Message from LoRaWAN device displayed in Azure IoT Explorer

Currently the code has a lots of diagnostic Console.Writeline statements, doesn’t support Uplink messages, has no Advanced Message Queuing Protocol(AMQP) client connection pooling, can’t run as an Azure Webjob, and a number of other features which I plan on adding in future blog posts.

Azure Blob storage BlobTrigger .Net Webjob

With the Windows 10 IoT Core application now reliably uploading images to Azure Blob Storage I wanted a simple test application to email the images to me as they arrived. So I hacked up an Azure Webjob using the SendGrid extension and a BlobTrigger

PIR Sensor trigger

After a couple of failed attempts (due to NuGet package versioning mismatches) this was the smallest, reliable enough application I could come up with. Beware BlobTriggers are not really intended for solutions requiring high throughput and/or reliability.

/*
    Copyright ® 2019 March devMobile Software, All Rights Reserved
 
    MIT License
...
*/
namespace devMobile.Azure.Storage
{
	using System.IO;
	using System.Configuration;
	using System.Threading.Tasks;
	using Microsoft.Azure.WebJobs;
	using Microsoft.Azure.WebJobs.Host;
	using SendGrid.Helpers.Mail;

	public static class ImageEmailer
	{
		[FunctionName("ImageEmailer")]
		public async static Task Run(
				[BlobTrigger("seeedrpibasehat190321/{name}")]
				Stream inputBlob,
				string name,
				[SendGrid(ApiKey = "")]
				IAsyncCollector<SendGridMessage> messageCollector,
				TraceWriter log)
		{
			log.Info($"C# Blob trigger function Processed blob Name:{name} Size: {inputBlob.Length} Bytes");

			SendGridMessage message = new SendGridMessage();
			message.AddTo(new EmailAddress(ConfigurationManager.AppSettings["EmailAddressTo"]));
			message.From = new EmailAddress(ConfigurationManager.AppSettings["EmailAddressFrom"]);
			message.SetSubject("RPI Web camera Image attached");
			message.AddContent("text/plain", $"{name} {inputBlob.Length} bytes" );

			await message.AddAttachmentAsync(name, inputBlob, "image/jpeg");

			await messageCollector.AddAsync(message);
		}
	}
}
Blob container and naming issues

This application highlighted a number of issues with my Windows 10 IoT Core client. They were

  • Configurable minimum period between images as PIR sensor would trigger multiple times as someone moved across my office.
  • Configurable Azure Blob Storage container for latest image as my BlobTrigger fired twice (for latest and timestamped images).
  • Configurable Azure Blob Storage container for image history as my BlobTrigger fired twice (for latest and timestamped images).
  • Include a unique device identifier (possibly MAC address) with image as I had two machines with the same device name on different networks.
  • Additional Blob metadata would be useful.
  • Additional logging would be useful for diagnosing problems.

I’ll look fix these issues in my next couple of posts