public class IoTHubApplicationSetting
{
public string DtdlModelId { get; set; }
}
public class IoTHubSettings
{
public string IoTHubConnectionString { get; set; } = string.Empty;
public Dictionary<string, IoTHubApplicationSetting> Applications { get; set; }
}
public class DeviceProvisiongServiceApplicationSetting
{
public string DtdlModelId { get; set; } = string.Empty;
public string GroupEnrollmentKey { get; set; } = string.Empty;
}
public class DeviceProvisiongServiceSettings
{
public string IdScope { get; set; } = string.Empty;
public Dictionary<string, DeviceProvisiongServiceApplicationSetting> Applications { get; set; }
}
public class IoTCentralMethodSetting
{
public byte Port { get; set; } = 0;
public bool Confirmed { get; set; } = false;
public Models.DownlinkPriority Priority { get; set; } = Models.DownlinkPriority.Normal;
public Models.DownlinkQueue Queue { get; set; } = Models.DownlinkQueue.Replace;
}
public class IoTCentralSetting
{
public Dictionary<string, IoTCentralMethodSetting> Methods { get; set; }
}
public class AzureIoTSettings
{
public IoTHubSettings IoTHub { get; set; }
public DeviceProvisiongServiceSettings DeviceProvisioningService { get; set; }
public IoTCentralSetting IoTCentral { get; set; }
}
Azure IoT Central appears to have no support for setting message properties so the LoRaWAN port, confirmed flag, priority, and queuing so these a retrieved from configuration.
Azure Function Configuration
Models.Downlink downlink;
Models.DownlinkQueue queue;
string payloadText = Encoding.UTF8.GetString(message.GetBytes()).Trim();
if (message.Properties.ContainsKey("method-name"))
{
#region Azure IoT Central C2D message processing
string methodName = message.Properties["method-name"];
if (string.IsNullOrWhiteSpace(methodName))
{
_logger.LogWarning("Downlink-DeviceID:{0} MessagedID:{1} LockToken:{2} method-name property empty", receiveMessageHandlerContext.DeviceId, message.MessageId, message.LockToken);
await deviceClient.RejectAsync(message);
return;
}
// Look up the method settings to get confirmed, port, priority, and queue
if ((_azureIoTSettings == null) || (_azureIoTSettings.IoTCentral == null) || !_azureIoTSettings.IoTCentral.Methods.TryGetValue(methodName, out IoTCentralMethodSetting methodSetting))
{
_logger.LogWarning("Downlink-DeviceID:{0} MessagedID:{1} LockToken:{2} method-name:{3} has no settings", receiveMessageHandlerContext.DeviceId, message.MessageId, message.LockToken, methodName);
await deviceClient.RejectAsync(message);
return;
}
downlink = new Models.Downlink()
{
Confirmed = methodSetting.Confirmed,
Priority = methodSetting.Priority,
Port = methodSetting.Port,
CorrelationIds = AzureLockToken.Add(message.LockToken),
};
queue = methodSetting.Queue;
// Check to see if special case for Azure IoT central command with no request payload
if (payloadText.IsPayloadEmpty())
{
downlink.PayloadRaw = "";
}
if (!payloadText.IsPayloadEmpty())
{
if (payloadText.IsPayloadValidJson())
{
downlink.PayloadDecoded = JToken.Parse(payloadText);
}
else
{
downlink.PayloadDecoded = new JObject(new JProperty(methodName, payloadText));
}
}
logger.LogInformation("Downlink-IoT Central DeviceID:{0} Method:{1} MessageID:{2} LockToken:{3} Port:{4} Confirmed:{5} Priority:{6} Queue:{7}",
receiveMessageHandlerContext.DeviceId,
methodName,
message.MessageId,
message.LockToken,
downlink.Port,
downlink.Confirmed,
downlink.Priority,
queue);
#endregion
}
The reboot command payload only contains an “@” so the TTTI payload will be empty, the minimum and maximum command payloads will contain only a numeric value which is added to the decoded payload with the method name, the combined minimum and maximum command has a JSON payload which is “grafted” into the decoded payload.
The Azure IoT Central Location Telemetry messages have a slightly different format to the output of the TTI LPP Payload formatter so the payload has to be “post processed”.
private 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)
{
// TODO Need to check if similar approach necessary accelerometer and gyro LPP payloads
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);
}
}
}
I may have to extend this method for other LPP datatypes
“Post processed” TTI JSON GPS Position data suitable for Azure IoT Central
Azure IoT Central Device Template with Location Capability
For initial development and testing I ran the function application in the desktop emulator and simulated TTI webhook calls with Telerik Fiddler and modified sample payloads. After some issues with iothub-creation-time-utc decoded telemetry messages were displayed in the Device Raw Data tab
Azure IoT Central Device Raw Data tab with successfully decoded GPS location payloads
Azure IoT Central map displaying with device location highlighted
The TTI V3 Connector Minimalist Cloud to Device only required a port number, and there was no way to specify whether delivery of message had to be confirmed, the way the message was queued, or the priority of message delivery. Like the port number these optional settings can be specified in message properties.
If any of these properties are incorrect DeviceClient.RejectAsync is called which deletes the message from the device queue and indicates to the server that the message could not be processed.
private async Task AzureIoTHubClientReceiveMessageHandler(Message message, object userContext)
{
try
{
Models.AzureIoTHubReceiveMessageHandlerContext receiveMessageHandlerContext = (Models.AzureIoTHubReceiveMessageHandlerContext)userContext;
if (!_DeviceClients.TryGetValue(receiveMessageHandlerContext.DeviceId, out DeviceClient deviceClient))
{
_logger.LogWarning("Downlink-DeviceID:{0} unknown", receiveMessageHandlerContext.DeviceId);
return;
}
using (message)
{
string payloadText = Encoding.UTF8.GetString(message.GetBytes()).Trim();
if (!AzureDownlinkMessage.PortTryGet(message.Properties, out byte port))
{
_logger.LogWarning("Downlink-Port property is invalid");
await deviceClient.RejectAsync(message);
return;
}
if (!AzureDownlinkMessage.ConfirmedTryGet(message.Properties, out bool confirmed))
{
_logger.LogWarning("Downlink-Confirmed flag is invalid");
await deviceClient.RejectAsync(message);
return;
}
if (!AzureDownlinkMessage.PriorityTryGet(message.Properties, out Models.DownlinkPriority priority))
{
_logger.LogWarning("Downlink-Priority value is invalid");
await deviceClient.RejectAsync(message);
return;
}
if (!AzureDownlinkMessage.QueueTryGet(message.Properties, out Models.DownlinkQueue queue))
{
_logger.LogWarning("Downlink-Queue value is invalid");
await deviceClient.RejectAsync(message.LockToken);
return;
}
Models.Downlink downlink = new Models.Downlink()
{
Confirmed = confirmed,
Priority = priority,
Port = port,
CorrelationIds = AzureLockToken.Add(message.LockToken),
};
// Split over multiple lines in an attempt to improve readability. In this scenario a valid JSON string should start/end with {/} for an object or [/] for an array
if ((payloadText.StartsWith("{") && payloadText.EndsWith("}"))
||
((payloadText.StartsWith("[") && payloadText.EndsWith("]"))))
{
try
{
downlink.PayloadDecoded = JToken.Parse(payloadText);
}
catch (JsonReaderException)
{
downlink.PayloadRaw = payloadText;
}
}
else
{
downlink.PayloadRaw = payloadText;
}
_logger.LogInformation("Downlink-IoT Hub DeviceID:{0} MessageID:{2} LockToken:{3} Port:{4} Confirmed:{5} Priority:{6} Queue:{7}",
receiveMessageHandlerContext.DeviceId,
message.MessageId,
message.LockToken,
downlink.Port,
downlink.Confirmed,
downlink.Priority,
queue);
Models.DownlinkPayload Payload = new Models.DownlinkPayload()
{
Downlinks = new List<Models.Downlink>()
{
downlink
}
};
string url = $"{receiveMessageHandlerContext.WebhookBaseURL}/{receiveMessageHandlerContext.ApplicationId}/webhooks/{receiveMessageHandlerContext.WebhookId}/devices/{receiveMessageHandlerContext.DeviceId}/down/{queue}".ToLower();
using (var client = new WebClient())
{
client.Headers.Add("Authorization", $"Bearer {receiveMessageHandlerContext.ApiKey}");
client.UploadString(new Uri(url), JsonConvert.SerializeObject(Payload));
}
_logger.LogInformation("Downlink-DeviceID:{0} LockToken:{1} success", receiveMessageHandlerContext.DeviceId, message.LockToken);
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Downlink-ReceiveMessge processing failed");
}
}
A correlation identifier containing the Message LockToken is added to the downlink payload.
Unconfirmed Downlink Messages
Azure IoT Explorer Cloud to Device sending an unconfirmed downlink message
For unconfirmed messages The TTI Connector calls the DeviceClient.CompletedAsync method (with the LockToken from the CorrelationIDs list) which deletes the message from the device queue.
[Function("Queued")]
public async Task<HttpResponseData> Queued([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext executionContext)
{
var logger = executionContext.GetLogger("Queued");
// Wrap all the processing in a try\catch so if anything blows up we have logged it.
try
{
string payloadText = await req.ReadAsStringAsync();
Models.DownlinkQueuedPayload payload = JsonConvert.DeserializeObject<Models.DownlinkQueuedPayload>(payloadText);
if (payload == null)
{
logger.LogInformation("Queued-Payload {0} invalid", payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
string deviceId = payload.EndDeviceIds.DeviceId;
logger.LogInformation("Queued-ApplicationID:{0} DeviceID:{1} ", applicationId, deviceId);
if (!_DeviceClients.TryGetValue(deviceId, out DeviceClient deviceClient))
{
logger.LogInformation("Queued-Unknown device for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
return req.CreateResponse(HttpStatusCode.Conflict);
}
// If the message is not confirmed "complete" it as soon as with network
if (!payload.DownlinkQueued.Confirmed)
{
if (!AzureLockToken.TryGet(payload.DownlinkQueued.CorrelationIds, out string lockToken))
{
logger.LogWarning("Queued-DeviceID:{0} LockToken missing from payload:{1}", payload.EndDeviceIds.DeviceId, payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
try
{
await deviceClient.CompleteAsync(lockToken);
}
catch (DeviceMessageLockLostException)
{
logger.LogWarning("Queued-CompleteAsync DeviceID:{0} LockToken:{1} timeout", payload.EndDeviceIds.DeviceId, lockToken);
return req.CreateResponse(HttpStatusCode.Conflict);
}
logger.LogInformation("Queued-DeviceID:{0} LockToken:{1} success", payload.EndDeviceIds.DeviceId, lockToken);
}
}
catch (Exception ex)
{
logger.LogError(ex, "Queued message processing failed");
return req.CreateResponse(HttpStatusCode.InternalServerError);
}
return req.CreateResponse(HttpStatusCode.OK);
}
The Things Industries Live Data tab for an unconfirmed message-Queued
Azure Application Insights for an unconfirmed message
The Things Industries Live Data tab for an unconfirmed message-Sent
Confirmed Downlink Messages
Azure IoT Explorer Cloud to Device sending a confirmed downlink message
Azure Application Insights for a confirmed message
The Things Industries Live Data tab for a confirmed message-Sent
The Things Industries Live Data tab for a confirmed message-Ack
If message delivery succeeds the deviceClient.CompleteAsync method (with the LockToken from the CorrelationIDs list) is called which removes the message from the device queue.
[Function("Ack")]
public async Task<HttpResponseData> Ack([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext executionContext)
{
var logger = executionContext.GetLogger("Queued");
// Wrap all the processing in a try\catch so if anything blows up we have logged it.
try
{
string payloadText = await req.ReadAsStringAsync();
Models.DownlinkAckPayload payload = JsonConvert.DeserializeObject<Models.DownlinkAckPayload>(payloadText);
if (payload == null)
{
logger.LogInformation("Ack-Payload {0} invalid", payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
string deviceId = payload.EndDeviceIds.DeviceId;
logger.LogInformation("Ack-ApplicationID:{0} DeviceID:{1} ", applicationId, deviceId);
if (!_DeviceClients.TryGetValue(deviceId, out DeviceClient deviceClient))
{
logger.LogInformation("Ack-Unknown device for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
return req.CreateResponse(HttpStatusCode.Conflict);
}
if (!AzureLockToken.TryGet(payload.DownlinkAck.CorrelationIds, out string lockToken))
{
logger.LogWarning("Ack-DeviceID:{0} LockToken missing from payload:{1}", payload.EndDeviceIds.DeviceId, payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
try
{
await deviceClient.CompleteAsync(lockToken);
}
catch (DeviceMessageLockLostException)
{
logger.LogWarning("Ack-CompleteAsync DeviceID:{0} LockToken:{1} timeout", payload.EndDeviceIds.DeviceId, lockToken);
return req.CreateResponse(HttpStatusCode.Conflict);
}
logger.LogInformation("Ack-DeviceID:{0} LockToken:{1} success", payload.EndDeviceIds.DeviceId, lockToken);
}
catch (Exception ex)
{
logger.LogError(ex, "Ack message processing failed");
return req.CreateResponse(HttpStatusCode.InternalServerError);
}
return req.CreateResponse(HttpStatusCode.OK);
}
Azure Application Insights for an confirmed message Ack
If message delivery fails the deviceClient.AbandonAsync method (with the LockToken from the CorrelationIDs list) is called which puts the downlink message back onto the device queue.
[Function("Failed")]
public async Task<HttpResponseData> Failed([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext executionContext)
{
var logger = executionContext.GetLogger("Queued");
// Wrap all the processing in a try\catch so if anything blows up we have logged it.
try
{
string payloadText = await req.ReadAsStringAsync();
Models.DownlinkFailedPayload payload = JsonConvert.DeserializeObject<Models.DownlinkFailedPayload>(payloadText);
if (payload == null)
{
logger.LogInformation("Failed-Payload {0} invalid", payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
string deviceId = payload.EndDeviceIds.DeviceId;
logger.LogInformation("Failed-ApplicationID:{0} DeviceID:{1} ", applicationId, deviceId);
if (!_DeviceClients.TryGetValue(deviceId, out DeviceClient deviceClient))
{
logger.LogInformation("Failed-Unknown device for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
return req.CreateResponse(HttpStatusCode.Conflict);
}
if (!AzureLockToken.TryGet(payload.DownlinkFailed.CorrelationIds, out string lockToken))
{
logger.LogWarning("Failed-DeviceID:{0} LockToken missing from payload:{1}", payload.EndDeviceIds.DeviceId, payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
try
{
await deviceClient.RejectAsync(lockToken);
}
catch (DeviceMessageLockLostException)
{
logger.LogWarning("Failed-RejectAsync DeviceID:{0} LockToken:{1} timeout", payload.EndDeviceIds.DeviceId, lockToken);
return req.CreateResponse(HttpStatusCode.Conflict);
}
logger.LogInformation("Failed-DeviceID:{0} LockToken:{1} success", payload.EndDeviceIds.DeviceId, lockToken);
}
catch (Exception ex)
{
logger.LogError(ex, "Failed message processing failed");
return req.CreateResponse(HttpStatusCode.InternalServerError);
}
return req.CreateResponse(HttpStatusCode.OK);
}
If message delivery is unsuccessful the deviceClient.RejectAsync method (with the LockToken from the CorrelationIDs list) is called which deletes the message from the device queue and indicates to the server that the message could not be processed.
[Function("Nack")]
public async Task<HttpResponseData> Nack([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext executionContext)
{
var logger = executionContext.GetLogger("Queued");
// Wrap all the processing in a try\catch so if anything blows up we have logged it.
try
{
string payloadText = await req.ReadAsStringAsync();
Models.DownlinkNackPayload payload = JsonConvert.DeserializeObject<Models.DownlinkNackPayload>(payloadText);
if (payload == null)
{
logger.LogInformation("Nack-Payload {0} invalid", payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
string deviceId = payload.EndDeviceIds.DeviceId;
logger.LogInformation("Nack-ApplicationID:{0} DeviceID:{1} ", applicationId, deviceId);
if (!_DeviceClients.TryGetValue(deviceId, out DeviceClient deviceClient))
{
logger.LogInformation("Nack-Unknown device for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
return req.CreateResponse(HttpStatusCode.Conflict);
}
if (!AzureLockToken.TryGet(payload.DownlinkNack.CorrelationIds, out string lockToken))
{
logger.LogWarning("Nack-DeviceID:{0} LockToken missing from payload:{1}", payload.EndDeviceIds.DeviceId, payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
try
{
await deviceClient.RejectAsync(lockToken);
}
catch (DeviceMessageLockLostException)
{
logger.LogWarning("Nack-RejectAsync DeviceID:{0} LockToken:{1} timeout", payload.EndDeviceIds.DeviceId, lockToken);
return req.CreateResponse(HttpStatusCode.Conflict);
}
logger.LogInformation("Nack-DeviceID:{0} LockToken:{1} success", payload.EndDeviceIds.DeviceId, lockToken);
}
catch (Exception ex)
{
logger.LogError(ex, "Nack message processing failed");
return req.CreateResponse(HttpStatusCode.InternalServerError);
}
return req.CreateResponse(HttpStatusCode.OK);
}
The way message Failed(Abandon), Ack(CompleteAsync) and Nack(RejectAsync) are handled needs some more testing to confirm my understanding of the sequencing of TTI confirmed message delivery.
BEWARE
The use of Confirmed messaging with devices that send uplink messages irregularly can cause weird problems if the Azure IoT hub downlink message times out.
In the Azure Portal I configured the DPS ID Scope (AzureSettings:DeviceProvisioningServiceSettings:IdScope) and the Group Enrollment Key(AzureSettings:DeviceProvisioningServiceSettings:GroupEnrollmentKey) then saved the configuration which restarted the AppService.
Azure Portal AppService configration
The first time a device sent an uplink message the cache query fails and the RegisterAsync method of the ProvisioningDeviceClient is called to get a device connection string.
logger.LogInformation("Uplink-ApplicationID:{0} DeviceID:{1} Port:{2} Payload Raw:{3}", applicationId, deviceId, port, payload.UplinkMessage.PayloadRaw);
if (!_DeviceClients.TryGetValue(deviceId, out DeviceClient deviceClient))
{
logger.LogInformation("Uplink-Unknown device for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
// Check that only one of Azure Connection string or DPS is configured
if (string.IsNullOrEmpty(_azureSettings.IoTHubConnectionString) && (_azureSettings.DeviceProvisioningServiceSettings == null))
{
logger.LogError("Uplink-Neither Azure IoT Hub connection string or Device Provisioning Service configured");
return req.CreateResponse(HttpStatusCode.UnprocessableEntity);
}
// Check that only one of Azure Connection string or DPS is configured
if (!string.IsNullOrEmpty(_azureSettings.IoTHubConnectionString) && (_azureSettings.DeviceProvisioningServiceSettings != null))
{
logger.LogError("Uplink-Both Azure IoT Hub connection string and Device Provisioning Service configured");
return req.CreateResponse(HttpStatusCode.UnprocessableEntity);
}
// User Azure IoT Connection string if configured and Device Provisioning Service isn't
if (!string.IsNullOrEmpty(_azureSettings.IoTHubConnectionString))
{
deviceClient = DeviceClient.CreateFromConnectionString(_azureSettings.IoTHubConnectionString, deviceId, transportSettings);
try
{
await deviceClient.OpenAsync();
}
catch (DeviceNotFoundException)
{
logger.LogWarning("Uplink-Unknown DeviceID:{0}", deviceId);
return req.CreateResponse(HttpStatusCode.NotFound);
}
}
// Azure IoT Hub Device provisioning service if configured
if (_azureSettings.DeviceProvisioningServiceSettings != null)
{
string deviceKey;
if ( string.IsNullOrEmpty(_azureSettings.DeviceProvisioningServiceSettings.IdScope) || string.IsNullOrEmpty(_azureSettings.DeviceProvisioningServiceSettings.GroupEnrollmentKey))
{
logger.LogError("Uplink-Device Provisioning Service requires ID Scope and Group Enrollment Key configured");
return req.CreateResponse(HttpStatusCode.UnprocessableEntity);
}
using (var hmac = new HMACSHA256(Convert.FromBase64String(_azureSettings.DeviceProvisioningServiceSettings.GroupEnrollmentKey)))
{
deviceKey = Convert.ToBase64String(hmac.ComputeHash(Encoding.UTF8.GetBytes(deviceId)));
}
using (var securityProvider = new SecurityProviderSymmetricKey(deviceId, deviceKey, null))
{
using (var transport = new ProvisioningTransportHandlerAmqp(TransportFallbackType.TcpOnly))
{
ProvisioningDeviceClient provClient = ProvisioningDeviceClient.Create(
Constants.AzureDpsGlobalDeviceEndpoint,
_azureSettings.DeviceProvisioningServiceSettings.IdScope,
securityProvider,
transport);
DeviceRegistrationResult result = await provClient.RegisterAsync();
if (result.Status != ProvisioningRegistrationStatusType.Assigned)
{
_logger.LogError("Config-DeviceID:{0} Status:{1} RegisterAsync failed ", deviceId, result.Status);
return req.CreateResponse(HttpStatusCode.FailedDependency);
}
IAuthenticationMethod authentication = new DeviceAuthenticationWithRegistrySymmetricKey(result.DeviceId, (securityProvider as SecurityProviderSymmetricKey).GetPrimaryKey());
deviceClient = DeviceClient.Create(result.AssignedHub, authentication, transportSettings);
await deviceClient.OpenAsync();
}
}
}
if (!_DeviceClients.TryAdd(deviceId, deviceClient))
{
logger.LogWarning("Uplink-TryAdd failed for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
return req.CreateResponse(HttpStatusCode.Conflict);
}
Models.AzureIoTHubReceiveMessageHandlerContext context = new Models.AzureIoTHubReceiveMessageHandlerContext()
{
DeviceId = deviceId,
ApplicationId = applicationId,
WebhookId = _theThingsIndustriesSettings.WebhookId,
WebhookBaseURL = _theThingsIndustriesSettings.WebhookBaseURL,
ApiKey = _theThingsIndustriesSettings.ApiKey
};
await deviceClient.SetReceiveMessageHandlerAsync(AzureIoTHubClientReceiveMessageHandler, context);
await deviceClient.SetMethodDefaultHandlerAsync(AzureIoTHubClientDefaultMethodHandler, context);
}
JObject telemetryEvent = new JObject
{
{ "ApplicationID", applicationId },
{ "DeviceID", deviceId },
{ "Port", port },
{ "Simulated", payload.Simulated },
{ "ReceivedAtUtc", payload.UplinkMessage.ReceivedAtUtc.ToString("s", CultureInfo.InvariantCulture) },
{ "PayloadRaw", payload.UplinkMessage.PayloadRaw }
};
// If the payload has been decoded by payload formatter, put it in the message body.
if (payload.UplinkMessage.PayloadDecoded != null)
{
telemetryEvent.Add("PayloadDecoded", payload.UplinkMessage.PayloadDecoded);
}
// Send the message to Azure IoT Hub
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", payload.UplinkMessage.ReceivedAtUtc.ToString("s", CultureInfo.InvariantCulture));
ioTHubmessage.Properties.Add("ApplicationId", applicationId);
ioTHubmessage.Properties.Add("DeviceEUI", payload.EndDeviceIds.DeviceEui);
ioTHubmessage.Properties.Add("DeviceId", deviceId);
ioTHubmessage.Properties.Add("port", port.ToString());
ioTHubmessage.Properties.Add("Simulated", payload.Simulated.ToString());
await deviceClient.SendEventAsync(ioTHubmessage);
logger.LogInformation("Uplink-DeviceID:{0} SendEventAsync success", payload.EndDeviceIds.DeviceId);
}
}
catch (Exception ex)
{
logger.LogError(ex, "Uplink-Message processing failed");
return req.CreateResponse(HttpStatusCode.InternalServerError);
}
I used Telerik Fiddler and some sample payloads copied from my Azure Storage Queue sample to simulate many devices and the registrations were spread across my five Azure IoT Hubs.
DPS Device Registrations tab showing distribution of LoRaWAN Devices
I need to review the HTTP Error codes returned for different errors and ensure failures are handled robustly.
In this version a downlink message can be sent to a device only after an uplink message. I’m looking at adding an Azure Function which initiates a connection to the configured Azure IoT Hub for the specified device to mitigate with this issue.
To send a TTN downlink message to a device the minimum required info is the LoRaWAN port number (specified in a Custom Property on the Azure IoT Hub cloud to device message), the device Id (from uplink message payload, which has been validated by a successful Azure IoT Hub connection) web hook id, web hook base URL, and an API Key (The Web Hook parameters are stored in the Connector configuration).
After some experimentation in previous TTN Connectors I found the synchronous nature of DirectMethods didn’t work well with LoRAWAN “irregular” connectivity so currently they are ignored.
public partial class Integration
{
private async Task AzureIoTHubClientReceiveMessageHandler(Message message, object userContext)
{
try
{
Models.AzureIoTHubReceiveMessageHandlerContext receiveMessageHandlerContext = (Models.AzureIoTHubReceiveMessageHandlerContext)userContext;
if (!_DeviceClients.TryGetValue(receiveMessageHandlerContext.DeviceId, out DeviceClient deviceClient))
{
_logger.LogWarning("Downlink-DeviceID:{0} unknown", receiveMessageHandlerContext.DeviceId);
return;
}
using (message)
{
string payloadText = Encoding.UTF8.GetString(message.GetBytes()).Trim();
if (!AzureDownlinkMessage.PortTryGet(message.Properties, out byte port))
{
_logger.LogWarning("Downlink-Port property is invalid");
await deviceClient.RejectAsync(message);
return;
}
// Split over multiple lines in an attempt to improve readability. In this scenario a valid JSON string should start/end with {/} for an object or [/] for an array
if ((payloadText.StartsWith("{") && payloadText.EndsWith("}"))
||
((payloadText.StartsWith("[") && payloadText.EndsWith("]"))))
{
try
{
downlink.PayloadDecoded = JToken.Parse(payloadText);
}
catch (JsonReaderException)
{
downlink.PayloadRaw = payloadText;
}
}
else
{
downlink.PayloadRaw = payloadText;
}
_logger.LogInformation("Downlink-IoT Hub DeviceID:{0} MessageID:{1} LockToken :{2} Port{3}",
receiveMessageHandlerContext.DeviceId,
message.MessageId,
message.LockToken,
downlink.Port);
Models.DownlinkPayload Payload = new Models.DownlinkPayload()
{
Downlinks = new List<Models.Downlink>()
{
downlink
}
};
string url = $"{receiveMessageHandlerContext.WebhookBaseURL}/{receiveMessageHandlerContext.ApplicationId}/webhooks/{receiveMessageHandlerContext.WebhookId}/devices/{receiveMessageHandlerContext.DeviceId}/down/replace");
using (var client = new WebClient())
{
client.Headers.Add("Authorization", $"Bearer {receiveMessageHandlerContext.ApiKey}");
client.UploadString(new Uri(url), JsonConvert.SerializeObject(Payload));
}
_logger.LogInformation("Downlink-DeviceID:{0} LockToken:{1} success", receiveMessageHandlerContext.DeviceId, message.LockToken);
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Downlink-ReceiveMessge processing failed");
}
}
}
public partial class Webhooks
{
[Function("Uplink")]
public async Task<HttpResponseData> Uplink([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext executionContext)
{
var logger = executionContext.GetLogger("Uplink");
// Wrap all the processing in a try\catch so if anything blows up we have logged it. Will need to specialise for connectivity failues etc.
try
{
Models.PayloadUplink payload = JsonConvert.DeserializeObject<Models.PayloadUplink>(await req.ReadAsStringAsync());
if (payload == null)
{
logger.LogInformation("Uplink: Payload {0} invalid", await req.ReadAsStringAsync());
return req.CreateResponse(HttpStatusCode.BadRequest);
}
string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
string deviceId = payload.EndDeviceIds.DeviceId;
if ((payload.UplinkMessage.Port == null ) || (!payload.UplinkMessage.Port.HasValue) || (payload.UplinkMessage.Port.Value == 0))
{
logger.LogInformation("Uplink-ApplicationID:{0} DeviceID:{1} Payload Raw:{2} Control nessage", applicationId, deviceId, payload.UplinkMessage.PayloadRaw);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
int port = payload.UplinkMessage.Port.Value;
logger.LogInformation("Uplink-ApplicationID:{0} DeviceID:{1} Port:{2} Payload Raw:{3}", applicationId, deviceId, port, payload.UplinkMessage.PayloadRaw);
if (!_DeviceClients.TryGetValue(deviceId, out DeviceClient deviceClient))
{
logger.LogInformation("Uplink-Unknown device for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
deviceClient = DeviceClient.CreateFromConnectionString(_configuration.GetConnectionString("AzureIoTHub"), deviceId);
try
{
await deviceClient.OpenAsync();
}
catch (DeviceNotFoundException)
{
logger.LogWarning("Uplink-Unknown DeviceID:{0}", deviceId);
return req.CreateResponse(HttpStatusCode.NotFound);
}
if (!_DeviceClients.TryAdd(deviceId, deviceClient))
{
logger.LogWarning("Uplink-TryAdd failed for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
return req.CreateResponse(HttpStatusCode.Conflict);
}
}
JObject telemetryEvent = new JObject
{
{ "ApplicationID", applicationId },
{ "DeviceID", deviceId },
{ "Port", port },
{ "PayloadRaw", payload.UplinkMessage.PayloadRaw }
};
// If the payload has been decoded by payload formatter, put it in the message body.
if (payload.UplinkMessage.PayloadDecoded != null)
{
telemetryEvent.Add("PayloadDecoded", payload.UplinkMessage.PayloadDecoded);
}
// Send the message to Azure IoT Hub
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", payload.UplinkMessage.ReceivedAtUtc.ToString("s", CultureInfo.InvariantCulture));
ioTHubmessage.Properties.Add("ApplicationId", applicationId);
ioTHubmessage.Properties.Add("DeviceEUI", payload.EndDeviceIds.DeviceEui);
ioTHubmessage.Properties.Add("DeviceId", deviceId);
ioTHubmessage.Properties.Add("port", port.ToString());
await deviceClient.SendEventAsync(ioTHubmessage);
}
}
catch (Exception ex)
{
logger.LogError(ex, "Uplink message processing failed");
return req.CreateResponse(HttpStatusCode.InternalServerError);
}
return req.CreateResponse(HttpStatusCode.OK);
}
}
For initial development and testing I ran the function application in the desktop emulator and simulated TTI webhook calls with Telerik Fiddler and modified TTI sample payloads.
I then used Azure IoT Explorer to configure devices, view uplink traffic etc. When I connected to my Azure IoT Hub shortly after starting the application all the devices were disconnected.
Azure IoT Explorer – no connected devices
The SeeeduinoLoRaWAN devices report roughly every 15 minutes so it took a while for them all to connect. (the SeeeduinoLoRaWAN4 & SeeeduinoLoRaWAN6 need to be repaired) .
After some research I found references to the underlying problem in TTN and OpenAPI forums. The Dev_addr and Dev_eui fields are Base16(Hexidecimal) encoded binary but are being processed as if they were Base64(mime) encoded.
The TTI connector only displays the Device EUI so I changed the Dev_eui property to a string
Now the DeviceEUI values are displayed correctly and searching for EndDevices in Azure Application Insights is easier
TTI V3 Connector application running as a console application showing correct DeviceEUIs
Modifying the nSwag generated classes is a really nasty way of fixing the problem but I think this approach is okay as it’s only one field and any other solution I could find was significantly more complex.
Adding devMobile as a collaborator on the new application
TTI Application API Key configuration
The new Application API Key used by the MQTTnetmanaged client only needs to have write downlink and read uplink traffic enabled.
FTDI Adapter and modified LHT64 cable
So I could reliably connect to my LHT65 devices to configure them I modified a programming cable so I could use it with a spare FTDI adaptor without jumper wires. Todo this I used a small jewelers screwdriver to “pop” out the VCC cable and move the transmit data line.
After entering the device password and checking the firmware version I used the AT+CFG command to display the device settings
AT+CFG: Print all configurations
[334428]***** UpLinkCounter= 0 *****
[334430]TX on freq 923200000 Hz at DR 2
[334804]txDone
[339807]RX on freq 923200000 Hz at DR 2
[339868]rxTimeOut
[340807]RX on freq 923200000 Hz at DR 2
[340868]rxTimeOut
Correct Password
Stop Tx events,Please wait for all configurations to print
Printf all config...
AT+DEUI=a8 .. .. .. .. .. .. d6
AT+DADDR=01......D6
AT+APPKEY=9d .. .. .. .. .. .. .. .. .. .. .. .. .. .. 2e
AT+NWKSKEY=f6 .. .. .. .. .. .. .. .. .. .. .. .. .. .. 69
AT+APPSKEY=4c 35 .. .. .. .. .. .. .. .. .. .. .. .. .. 3d
AT+APPEUI=a0 .. .. .. .. .. .. 00
AT+ADR=1
AT+TXP=0
AT+DR=0
AT+DCS=0
AT+PNM=1
AT+RX2FQ=923200000
AT+RX2DR=2
AT+RX1DL=1000
AT+RX2DL=2000
AT+JN1DL=5000
AT+JN2DL=6000
AT+NJM=1
AT+NWKID=00 00 00 00
AT+FCU=0
AT+FCD=0
AT+CLASS=A
AT+NJS=0
AT+RECVB=0:
AT+RECV=0:
AT+VER=v1.7 AS923
AT+CFM=0
AT+CFS=0
AT+SNR=0
AT+RSSI=0
AT+TDC=1200000
AT+PORT=2
AT+PWORD=123456
AT+CHS=0
AT+DATE=21/3/26 07:49:15
AT+SLEEP=0
AT+EXT=4,2
AT+RTP=20
AT+BAT=3120
AT+WMOD=0
AT+ARTEMP=-40,125
AT+CITEMP=1
Start Tx events OK
[399287]***** UpLinkCounter= 0 *****
[399289]TX on freq 923400000 Hz at DR 2
[399663]txDone
[404666]RX on freq 923400000 Hz at DR 2
[404726]rxTimeOut
[405666]RX on freq 923200000 Hz at DR 2
[405726]rxTimeOut
The Dragino LHT65 uses the DeviceEUI as the DeviceID which meant I had todo more redaction in my TTI/TTN and Azure Application Insights screen captures. The rules around the re-use of EndDevice ID were a pain in the arse(PITA) in my development focused tenant.
Dragino LHT 65 Device uplink payload formatter
The connector supports both uplink and downlink messages with JSON encoded payloads. The Dragino LHT65 has a vendor supplied formatter which is automatically configured when an EndDevice is created. The EndDevice formatter configuration can also be overridden at the Application level in the app.settings.json file.
Device Live Data Uplink Data Payload
Once an EndDevice is configured in TTI/TTN I usually use the “Live data Uplink Payload” to work out the decoded payload JSON property names and data types.
LHT65 Uplink only Azure IoT Central Device Template
LHT65 Device Template View Identity
For Azure IoT Central “automagic” provisioning the DTDLModelId has to be copied from the Azure IoT Central Template into the TTI/TTN EndDevice or app.settings.json file application configuration.
LHT65 Device Template copy DTDL @ID
TTI EndDevice configuring the DTDLV2 @ID at the device level
Configuring the DTDLV2 @ID at the TTI application level in the app.settings.json file
The Azure Device Provisioning Service(DPS) is configured at the TTI application level in the app.settings.json file. The IDScope and one of the Primary or Secondary Shared Access Signature(SAS) keys should be copied into DeviceProvisioningServiceSettings of an Application in the app.settings.json file. I usually set the “Automatically connect devices in this group” flag as part of the “automagic” provisioning process.
Azure IoT Central Group Enrollment Key
Then device templates need to be mapped to an Enrollment Group then Device Group.
For testing the connector application can be run locally with diagnostic information displayed in the application console window as it “automagically’ provisions devices and uploads telemetry data.
Connector application Diagnostics
Azure IoT Central Device list before my LHT65 device is “automagically” provisioned
Azure IoT Central Device list after my LHT65 device is “automagically” provisioned
One a device has been provisioned I check on the raw data display that all the fields I configured have been mapped correctly.
Azure IoT Central raw data display
I then created a dashboard to display the telemetry data from the LHT65 sensors.
Azure IoT Central dashboard displaying LHT65 temperature, humidity and battery voltage graphs.
The dashboard also has a few KPI displays which highlighted an issue which occurs a couple of times a month with the LHT65 onboard temperature sensor values (327.7°). I have connected Dragino technical support and have also been unable to find a way to remove the current an/or filter out future aberrant values.
Azure Application Insights logging
I also noticed that the formatting of the DeviceEUI values in the Application Insights logging was incorrect after trying to search for one of my Seeedstudio LoRaWAN device with its DeviceEUI.
ProvisioningRegistrationAdditionalData provisioningRegistrationAdditionalData = new ProvisioningRegistrationAdditionalData()
{
JsonData = $"{{\"modelId\": \"{modelId}\"}}"
};
result = await provClient.RegisterAsync(provisioningRegistrationAdditionalData);
I remembered seeing a sample where the DTDLV2 methodId was formatted by a library function and after a surprising amount of searching I found what I was looking for in Azure-Samples repository.
// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
using Microsoft.Azure.Devices.Provisioning.Client.Extensions;
namespace Microsoft.Azure.Devices.Provisioning.Client.PlugAndPlay
{
/// <summary>
/// A helper class for formatting the DPS device registration payload, per plug and play convention.
/// </summary>
public static class PnpConvention
{
/// <summary>
/// Create the DPS payload to provision a device as plug and play.
/// </summary>
/// <remarks>
/// For more information on device provisioning service and plug and play compatibility,
/// and PnP device certification, see <see href="https://docs.microsoft.com/en-us/azure/iot-pnp/howto-certify-device"/>.
/// The DPS payload should be in the format:
/// <code>
/// {
/// "modelId": "dtmi:com:example:modelName;1"
/// }
/// </code>
/// For information on DTDL, see <see href="https://github.com/Azure/opendigitaltwins-dtdl/blob/master/DTDL/v2/dtdlv2.md"/>
/// </remarks>
/// <param name="modelId">The Id of the model the device adheres to for properties, telemetry, and commands.</param>
/// <returns>The DPS payload to provision a device as plug and play.</returns>
public static string CreateDpsPayload(string modelId)
{
modelId.ThrowIfNullOrWhiteSpace(nameof(modelId));
return $"{{\"modelId\":\"{modelId}\"}}";
}
}
}
With a couple of changes my code now uses the CreateDpsPayload method
using Microsoft.Azure.Devices.Provisioning.Client.PlugAndPlay;
...
using (var securityProvider = new SecurityProviderSymmetricKey(deviceId, deviceKey, null))
{
using (var transport = new ProvisioningTransportHandlerAmqp(TransportFallbackType.TcpOnly))
{
ProvisioningDeviceClient provClient = ProvisioningDeviceClient.Create(
Constants.AzureDpsGlobalDeviceEndpoint,
deviceProvisiongServiceSettings.IdScope,
securityProvider,
transport);
DeviceRegistrationResult result;
if (!string.IsNullOrEmpty(modelId))
{
ProvisioningRegistrationAdditionalData provisioningRegistrationAdditionalData = new ProvisioningRegistrationAdditionalData()
{
JsonData = PnpConvention.CreateDpsPayload(modelId)
};
result = await provClient.RegisterAsync(provisioningRegistrationAdditionalData, stoppingToken);
}
else
{
result = await provClient.RegisterAsync(stoppingToken);
}
if (result.Status != ProvisioningRegistrationStatusType.Assigned)
{
_logger.LogError("Config-DeviceID:{0} Status:{1} RegisterAsync failed ", deviceId, result.Status);
return false;
}
IAuthenticationMethod authentication = new DeviceAuthenticationWithRegistrySymmetricKey(result.DeviceId, (securityProvider as SecurityProviderSymmetricKey).GetPrimaryKey());
deviceClient = DeviceClient.Create(result.AssignedHub, authentication, transportSettings);
}
}
To reduce the impact of the RegisterAsync call duration this Proof of Concept(PoC) code uses the System.Tasks.Threading library to execute each request in its own thread and then wait for all the requests to finish.
The connector application paginates the retrieval of device configuration from TTI API and a Task is created for each device returned in a page. In the Application Insights Trace logging the duration of a single page of device registrations was approximately the duration of the longest call.
There will be a tradeoff between device page size (resource utilisation by many threads) and startup duration (to many sequential page operations) which will need to be explored.