.NET Core web API + Dapper – QueryMultiple

Returning multiple recordsets

My current “day job” is building applications for managing portfolios of foreign currency instruments. A portfolio can contain many different types of instrument (Forwards, Options, Swaps etc.). One of the “optimisations” we use is retrieving all the different types of instruments in a portfolio with one stored procedure call.

SQL Server Management Studio Dependency viewer

The closest scenario I could come up with using the World Wide Importers database was retrieving a summary of all the information associated with an Invoice for display on a single screen.

CREATE PROCEDURE [Sales].[InvoiceSummaryGetV1](@InvoiceID as int)
AS
BEGIN

SELECT [InvoiceID]
--        ,[CustomerID]
--        ,[BillToCustomerID]
		,[OrderID]
		,[Invoices].[DeliveryMethodID]
		,[DeliveryMethodName]
--        ,[ContactPersonID]
--        ,[AccountsPersonID]
		,[SalespersonPersonID] as SalesPersonID
		,[SalesPerson].[PreferredName] as SalesPersonName
--        ,[PackedByPersonID]
		,[InvoiceDate]
		,[CustomerPurchaseOrderNumber]
		,[IsCreditNote]
		,[CreditNoteReason]
		,[Comments]
		,[DeliveryInstructions]
--        ,[InternalComments]
--        ,[TotalDryItems]
--        ,[TotalChillerItems]
		,[DeliveryRun]
		,[RunPosition] as DeliveryRunPosition
		,[ReturnedDeliveryData] as DeliveryData
		,[ConfirmedDeliveryTime] as DeliveredAt
		,[ConfirmedReceivedBy] as DeliveredTo
--        ,[LastEditedBy]
--        ,[LastEditedWhen]
	FROM [Sales].[Invoices]
	INNER JOIN [Application].[People] as SalesPerson ON (Invoices.[SalespersonPersonID] = [SalesPerson].[PersonID])
	INNER JOIN [Application].[DeliveryMethods] as DeliveryMethod ON (Invoices.[DeliveryMethodID] = DeliveryMethod.[DeliveryMethodID])
WHERE ([Invoices].[InvoiceID] = @InvoiceID)

SELECT [InvoiceLineID]
      ,[InvoiceID]
      ,[StockItemID]
      ,[Description] as StockItemDescription
      ,[InvoiceLines].[PackageTypeID]
	  ,[PackageType].[PackageTypeName]
      ,[Quantity]
      ,[UnitPrice]
      ,[TaxRate]
      ,[TaxAmount]
--      ,[LineProfit]
      ,[ExtendedPrice]
--      ,[LastEditedBy]
--      ,[LastEditedWhen]
	FROM [Sales].[InvoiceLines]
		INNER JOIN [Warehouse].[PackageTypes] as PackageType ON ([PackageType].[PackageTypeID] = [InvoiceLines].[PackageTypeID])
WHERE ([InvoiceLines].[InvoiceID] = @InvoiceID)

SELECT [StockItemTransactionID]
      ,[StockItemTransactions].[StockItemID]
      ,StockItem.[StockItemName] as StockItemName
      ,[StockItemTransactions].[TransactionTypeID]
      ,[TransactionType].[TransactionTypeName]
--      ,[CustomerID]
--      ,[InvoiceID]
--      ,[SupplierID]
--      ,[PurchaseOrderID]
      ,[TransactionOccurredWhen] as TransactionAt
      ,[Quantity]
--      ,[LastEditedBy]
--      ,[LastEditedWhen]
	FROM [Warehouse].[StockItemTransactions]
	INNER JOIN [Warehouse].[StockItems] as StockItem ON ([StockItemTransactions].StockItemID = [StockItem].StockItemID)
	INNER JOIN [Application].[TransactionTypes] as TransactionType ON ([StockItemTransactions].[TransactionTypeID] = TransactionType.[TransactionTypeID])
	WHERE ([StockItemTransactions].[InvoiceID] = @InvoiceID)

END

The stored procedure returns 3 recordsets, a “summary” of the Order, a summary of the associated OrderLines and a summary of the associated StockItemTransactions.

public async Task<ActionResult<Model.InvoiceSummaryGetDtoV1>>Get([Range(1, int.MaxValue, ErrorMessage = "Invoice id must greater than 0")] int id)
{
	Model.InvoiceSummaryGetDtoV1 response = null;

	try
	{
		using (SqlConnection db = new SqlConnection(this.connectionString))
		{
			var invoiceSummary = await db.QueryMultipleWithRetryAsync("[Sales].[InvoiceSummaryGetV1]", param: new { InvoiceId = id }, commandType: CommandType.StoredProcedure);

			response = await invoiceSummary.ReadSingleOrDefaultWithRetryAsync<Model.InvoiceSummaryGetDtoV1>();
			if (response == default)
			{
				logger.LogInformation("Invoice:{0} not found", id);

				return this.NotFound($"Invoice:{id} not found");
			}

			response.InvoiceLines = (await invoiceSummary.ReadWithRetryAsync<Model.InvoiceLineSummaryListDtoV1>()).ToArray();

			response.StockItemTransactions = (await invoiceSummary.ReadWithRetryAsync<Model.StockItemTransactionSummaryListDtoV1>()).ToArray();
		}
	}
	catch (SqlException ex)
	{
		logger.LogError(ex, "Retrieving Invoice, Invoice Lines or Stock Item Transactions");

		return this.StatusCode(StatusCodes.Status500InternalServerError);
	}

	return this.Ok(response);
}

I use Google Chrome, Mozilla Firefox, Microsoft Edgeium, and Opera but the screen capture was done with FireFox mainly because it formats the Java Script Object Notation(JSON) response payloads nicely.

FireFox displaying Invoice Summary response

I had to extend the DapperTransient module to add SqlMapper extension (plus all the different overloads) retry methods.

NOTE : Error Handling approach has been updated

Smartish Edge Camera – Azure Storage basics

This project is another reworked version of on my ML.Net YoloV5 + Camera + GPIO on ARM64 Raspberry PI which supports only the uploading of camera and marked up images to Azure Storage.

My backyard test-rig consists of a Unv IPC675LFW Pan Tilt Zoom(PTZ) Security Camera, Power over Ethernet(PoE) module, and a Raspberry Pi 4B 8G.

Raspberry PI 4 B backyard test rig

The application can be compiled with Raspberry PI V2 Camera or Unv Security Camera (The security camera configuration may work for other cameras/vendors).

The appsetings.json file has configuration options for the Azure Storage Account, DeviceID (Used for the Azure Blob storage container name), the list of object classes of interest (based on the YoloV5 image classes) , and the image blob storage file names (used to “bucket” images).

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft": "Warning",
      "Microsoft.Hosting.Lifetime": "Information"
    }
  },

  "Application": {
    "DeviceId": "edgecamera",

    "ImageTimerDue": "0.00:00:15",
    "ImageTimerPeriod": "0.00:00:30",

    "ImageCameraFilepath": "ImageCamera.jpg",
    "ImageMarkedUpFilepath": "ImageMarkedup.jpg",

    "ImageCameraUpload": true,
    "ImageMarkedupUpload": true,

    "YoloV5ModelPath": "YoloV5/yolov5s.onnx",

    "PredicitionScoreThreshold": 0.7,
    "PredictionLabelsOfInterest": [
      "bicycle",
      "person",
      "car"
    ],
    "OutputImageMarkup": true
  },

  "SecurityCamera": {
    "CameraUrl": "",
    "CameraUserName": "",
    "CameraUserPassword": ""
  },

  "RaspberryPICamera": {
    "ProcessWaitForExit": 1000,
    "Rotation": 180
  },

  "AzureStorage": {
    "ConnectionString": "FhisIsNotTheConnectionStringYouAreLookingFor",
    "ImageCameraFilenameFormat": "{0:yyyyMMdd}/camera/{0:HHmmss}.jpg",
    "ImageMarkedUpFilenameFormat": "{0:yyyyMMdd}/markedup/{0:HHmmss}.jpg"
  }
}

Part of this refactor was injecting(DI) the logging and configuration dependencies.

public class Program
{
	public static void Main(string[] args)
	{
		CreateHostBuilder(args).Build().Run();
	}

	public static IHostBuilder CreateHostBuilder(string[] args) =>
		 Host.CreateDefaultBuilder(args)
			.ConfigureServices((hostContext, services) =>
			{
				services.Configure<ApplicationSettings>(hostContext.Configuration.GetSection("Application"));
				services.Configure<SecurityCameraSettings>(hostContext.Configuration.GetSection("SecurityCamera"));
				services.Configure<RaspberryPICameraSettings>(hostContext.Configuration.GetSection("RaspberryPICamera"));
				services.Configure<AzureStorageSettings>(hostContext.Configuration.GetSection("AzureStorage"));
			})
			.ConfigureLogging(logging =>
			{
				logging.ClearProviders();
				logging.AddSimpleConsole(c => c.TimestampFormat = "[HH:mm:ss.ff]");
			})
			.UseSystemd()
			.ConfigureServices((hostContext, services) =>
			{
			  services.AddHostedService<Worker>();
			});
		}
	}
}

After the You Only Look Once(YOLOV5)+ML.Net+Open Neural Network Exchange(ONNX) plumbing has loaded a timer with a configurable due time and period is started.

private async void ImageUpdateTimerCallback(object state)
{
	DateTime requestAtUtc = DateTime.UtcNow;

	// Just incase - stop code being called while photo already in progress
	if (_cameraBusy)
	{
		return;
	}
	_cameraBusy = true;

	_logger.LogInformation("Image processing start");

	try
	{
#if CAMERA_RASPBERRY_PI
		RaspberryPIImageCapture();
#endif
#if CAMERA_SECURITY
		SecurityCameraImageCapture();
#endif
		if (_applicationSettings.ImageCameraUpload)
		{
					await AzureStorageImageUpload(requestAtUtc, _applicationSettings.ImageCameraFilepath, 
 azureStorageSettings.ImageCameraFilenameFormat);
		}

		List<YoloPrediction> predictions;

		using (Image image = Image.FromFile(_applicationSettings.ImageCameraFilepath))
		{
			_logger.LogTrace("Prediction start");
			predictions = _scorer.Predict(image);
			_logger.LogTrace("Prediction done");

			OutputImageMarkup(image, predictions, _applicationSettings.ImageMarkedUpFilepath);
		}

		if (_logger.IsEnabled(LogLevel.Trace))
		{
			_logger.LogTrace("Predictions {0}", predictions.Select(p => new { p.Label.Name, p.Score }));
		}

		var predictionsOfInterest = predictions.Where(p => p.Score > _applicationSettings.PredicitionScoreThreshold).Select(c => c.Label.Name).Intersect(_applicationSettings.PredictionLabelsOfInterest, StringComparer.OrdinalIgnoreCase);
		if (_logger.IsEnabled(LogLevel.Trace))
		{
			_logger.LogTrace("Predictions of interest {0}", predictionsOfInterest.ToList());
		}

		if (_applicationSettings.ImageMarkedupUpload && predictionsOfInterest.Any())
		{
			await AzureStorageImageUpload(requestAtUtc, _applicationSettings.ImageMarkedUpFilepath, _azureStorageSettings.ImageMarkedUpFilenameFormat);
		}

		var predictionsTally = predictions.Where(p => p.Score >= _applicationSettings.PredicitionScoreThreshold)
									.GroupBy(p => p.Label.Name)
									.Select(p => new
									{
										Label = p.Key,
										Count = p.Count()
									});

		if (_logger.IsEnabled(LogLevel.Information))
		{
			_logger.LogInformation("Predictions tally {0}", predictionsTally.ToList());
		}
	}
	catch (Exception ex)
	{
		_logger.LogError(ex, "Camera image download, post procesing, image upload, or telemetry failed");
	}
	finally
	{
		_cameraBusy = false;
	}

	TimeSpan duration = DateTime.UtcNow - requestAtUtc;

	_logger.LogInformation("Image processing done {0:f2} sec", duration.TotalSeconds);
}

In the ImageUpdateTimerCallback method a camera image is captured (by my Raspberry Pi Camera Module 2 or IPC675LFW Security Camera) and written to the local file system.

Raspberry PI4B console displaying image processing and uploading

The MentalStack YoloV5 model ML.Net support library processes the camera image on the local filesystem. The prediction output (can be inspected with Netron) is parsed generating list of objects that have been detected, their Minimum Bounding Rectangle(MBR) and class.

Image from security camera
Azure IoT Storage Explorer displaying list of camera images

The list of predictions is post processed with a Language Integrated Query(LINQ) which filters out predictions with a score below a configurable threshold(PredicitionScoreThreshold) and returns a count of each class. If this list intersects with the configurable PredictionLabelsOfInterest a marked up image is uploaded to Azure Storage.

Image from security camera marked up with Minimum Bounding Boxes(MBRs)
Azure IoT Storage Explorer displaying list of marked up camera images

The current implementation is quite limited, the camera image upload, object detection and image upload if there are objects of interest is implemented in a single timer callback. I’m considering implementing two timers one for the uploading of camera images (time lapse camera) and the other for running the object detection process and uploading marked up images.

Marked up images are uploaded if any of the objects detected (with a score greater than PredicitionScoreThreshold) is in the PredictionLabelsOfInterest. I’m considering adding a PredicitionScoreThreshold and minimum count for individual prediction classes, and optionally marked up image upload only when the list of objects detected has changed.

.NET Core web API + Dapper – Caching

Response Cache

In the beginning this was long long post about In-memory caching, distributed caching, Response caching, Response caching with middleware and Object reuse with ObjectPool. As I was re-reading the post before publishing it I came to the realisation that these different caching approaches didn’t require Dapper.

I started again, but kept the first section as it covers one of the simplest possible approaches to caching using the [ResponseCache] attribute and VaryByQueryKeys.

[HttpGet("Response")]
[ResponseCache(Duration = StockItemsListResponseCacheDuration)]
public async Task<ActionResult<IEnumerable<Model.StockItemListDtoV1>>> GetResponse()
{
	IEnumerable<Model.StockItemListDtoV1> response = null;

	logger.LogInformation("Response cache load");

	try
	{
		response = await dapper.QueryAsync<Model.StockItemListDtoV1>(sql: @"SELECT [StockItemID] as ""ID"", [StockItemName] as ""Name"", [RecommendedRetailPrice], [TaxRate] FROM [Warehouse].[StockItems]", commandType: CommandType.Text);
	}
	catch (SqlException ex)
	{
		logger.LogError(ex, "Retrieving list of StockItems");

		return this.StatusCode(StatusCodes.Status500InternalServerError);
	}

	return this.Ok(response);
}

[HttpGet("ResponseVarying")]
[ResponseCache(Duration = StockItemsListResponseCacheDuration, VaryByQueryKeys = new string[] { "id" })]
public async Task<ActionResult<Model.StockItemGetDtoV1>> Get([FromQuery(Name = "id"), Range(1, int.MaxValue, ErrorMessage = "Stock item id must greater than 0")] int id)
{
	Model.StockItemGetDtoV1 response = null;

	logger.LogInformation("Response cache varying load id:{0}", id);

	try
	{
		response = await dapper.QuerySingleOrDefaultAsync<Model.StockItemGetDtoV1>(sql: "[Warehouse].[StockItemsStockItemLookupV1]", param: new { stockItemId = id }, commandType: CommandType.StoredProcedure);
		if (response == default)
		{
			logger.LogInformation("StockItem:{0} not found", id);

			return this.NotFound($"StockItem:{id} not found");
		}
	}
	catch (SqlException ex)
	{
		logger.LogError(ex, "Looking up StockItem with Id:{0}", id);

		return this.StatusCode(StatusCodes.Status500InternalServerError);
	}

	return this.Ok(response);
}

I use Google Chrome, Mozilla Firefox, Microsoft Edgeium, and Opera but the screen captures have been done with FireFox mainly because it formats the Java Script Object Notation(JSON) response payloads nicely.

All the browsers appeared to respect the cache control headers but Firefox was the only one which did not initiate a new request when I pressed return in the Uniform Resource Locator(URL) field.

Firefox displaying list of stock items

I used Telerik Fiddler and FiddlerFox to capture the HTTP GET method request and response payloads.

Fiddler Fox extension details
Response payload for a list of StockItems with cache control headers highlighted
Firefox displaying single stock item
Response payload for a single StockItem with cache control headers highlighted

Dapper Cache

The Dapper Extensions Library has built in support for In-memory and Redis caching. The Dapper.Extensions Library extends Dapper’s functionality. It requires minimal configuration but I was tripped up by the default connection string requirement because I was using Dependency Injection

Dapper.Extensions NuGet package configuration

The configuration code in the application startup.cs supports in-memory and Redis caches.

// This method gets called by the runtime. Use this method to add services to the container.
public void ConfigureServices(IServiceCollection services)
{
	services.AddControllers();

	services.AddResponseCaching();

	services.AddDapperForMSSQL();

#if DAPPER_EXTENSIONS_CACHE_MEMORY
	services.AddDapperCachingInMemory(new MemoryConfiguration
	{
		AllMethodsEnableCache = false
	});
#endif
#if DAPPER_EXTENSIONS_CACHE_REDIS
	services.AddDapperCachingInRedis(new RedisConfiguration
	{
		AllMethodsEnableCache = false,
		KeyPrefix = Configuration.GetValue<string>("RedisKeyPrefix"),
		ConnectionString = Configuration.GetConnectionString("RedisConnection")
	}); 
#endif
	services.AddApplicationInsightsTelemetry();
}

The StockItemsCachingController was rewritten with the Dapper.Extensions QueryAsync and QuerySingleOrDefaultAsync methods.

[HttpGet("DapperMemory")]
public async Task<ActionResult<IEnumerable<Model.StockItemListDtoV1>>> GetDapper()
{
	List<Model.StockItemListDtoV1> response;

	logger.LogInformation("Dapper cache load");

	try
	{
		response = await dapper.QueryAsync<Model.StockItemListDtoV1>(
							sql: @"SELECT [StockItemID] as ""ID"", [StockItemName] as ""Name"", [RecommendedRetailPrice], [TaxRate] FROM [Warehouse].[StockItems]",
							commandType: CommandType.Text,
							enableCache: true,
							cacheExpire: TimeSpan.Parse(this.Configuration.GetValue<string>("DapperCachingDuration"))
					);

	}
	catch (SqlException ex)
	{
		logger.LogError(ex, "Retrieving list of StockItems");

		return this.StatusCode(StatusCodes.Status500InternalServerError);
	}

	return this.Ok(response);
}

[HttpGet("DapperMemoryVarying")]
public async Task<ActionResult<Model.StockItemGetDtoV1>> GetDapperVarying([FromQuery(Name = "id"), Range(1, int.MaxValue, ErrorMessage = "Stock item id must greater than 0")] int id)
{
	Model.StockItemGetDtoV1 response = null;

	logger.LogInformation("Dapper cache varying load id:{0}", id);

	try
	{
		response = await dapper.QuerySingleOrDefaultAsync<Model.StockItemGetDtoV1>(
					sql: "[Warehouse].[StockItemsStockItemLookupV1]",
					param: new { stockItemId = id },
					commandType: CommandType.StoredProcedure,
					cacheKey: $"StockItem:{id}",
					enableCache: true,
					cacheExpire: TimeSpan.Parse(this.Configuration.GetValue<string>("DapperCachingDuration"))
							);
		if (response == default)
		{
			logger.LogInformation("StockItem:{0} not found", id);

			return this.NotFound($"StockItem:{id} not found");
		}
	}
	catch (SqlException ex)
	{
		logger.LogError(ex, "Looking up StockItem with Id:{0}", id);

		return this.StatusCode(StatusCodes.Status500InternalServerError);
	}

	return this.Ok(response);
}

Both the Dapper.Extensions In-Memory and Redis cache reduced the number of database requests to the bare minimum. In a larger application the formatting of the cacheKey (cacheKey: “StockItems” & cacheKey: $”StockItem:{id}”) would be important to stop database query result collisions.

SQL Server Profiler displaying the list and single record requests.

I used Memurai which is a Microsoft Windows version of Redis for testing on my development machine before deploying to Microsoft Azure and using Azure Cache for Redis. Memurai runs as a Windows Service and supports master, replica, cluster node or sentinel roles.

Memurai running as a Windows Service on my development machine

When the Web API project was restarted the contents in-memory cache were lost. The Redis cache contents survive a restart and can be access from multiple clients.

The Dapper.Extensions Query, QueryAsync, QueryFirstOrDefaultAsync, QuerySingleOrDefault, QuerySingleOrDefaultAsync, QueryMultiple, QueryMultipleAsync, ExecuteReader, ExecuteReaderAsync, QueryPageAsync, QueryPageAsync, QueryPlainPage, QueryPlainPageAsync, Execute, ExecuteAsync, ExecuteScalar, ExecuteScalarAsync, BeginTransaction, CommitTransactionm and RollbackTransaction do not appear to a versions which “Retry” actions when there is a “Transient” failure. If there is no solution available I will build one using the approach in my DapperTransient module.

NOTE : Error Handling approach has been updated

Azure Percept Pay Attention to the Warnings

Azure IoT Hub setup “Learnings”

After roughly an hour the telemetry stopped and the Azure Percept displayed a message which wasn’t terribly helpful.

I had manually created the Azure IoT Hub and selected the “Free Tier” (I was trying to keep my monthly billing reasonable) then as I was stepping through the Azure Percept setup wizard I didn’t read the warning message highlighted below.

Azure Percept Azure IoT Hub Warning

The Azure Percept generates a lot of messages and I had quickly hit the 8000 messages per day limit of the “Free Tier”.

Azure IoT Hub Daily Message Quota

I had to create a new Azure IoT Hub, repave the Azure Percept Device (there were some updates and I had made some mistakes in the initial setup) and reconfigure the device.

Azure IoT Hub Minimum Tier configuration

Azure Percept “low code” integration Setup

Introduction

There have been blog posts showing how to build Azure Percept integrations with Power BI, Azure Logic Apps etc. with “zero code”.  But what do you do if your Azure Percept based solution needs some “glue” to connect to other systems?

I work on a SmartAg computer vision based application that uses security cameras to monitor the flow of cattle through stockyards. It has to control some local hardware, display real-time dashboards, and integrate with an existing application so a “zero code” solution wouldn’t work.

Having to connect an Azure Percept to 3rd party applications can’t be a unique problem so this series blog posts will show a couple of “low code” options that I have used to solve this issue. The technologies that will be covered include Azure IoT Hub Message Routing. Azure Storage Queues, Azure Service Bus Queues, Azure Service Bus Topics and Azure Functions.

The Pivot

The initial plan was to take the Azure Percept to a piggery to see if I could build a Proof of Concept(PoC) of a product that the CEO and I had been discussing for a couple of weeks.

But shortly after I started working on this series of blog posts New Zealand went into strict lockdown. Only essential shops like supermarkets and petrol stations were open, our groceries were being delivered, and schools were closed.

I needed a demonstration application which used props I could source from home and the local petrol station. In addition my teenage son’s school was closed so he could be the project “intern”.

While at the local petrol station to buy milk I observed that they had a large selection of confectionary so we decided to build a series of object detection models to count different types of chocolates.

In a retail scenario this could be counting products on shelves, pallets in a cold store, or at the SmartAg start-up I work for counting cattle in a yard.

Configuring The Test Environment

I have not included screen shots of the hardware configuration process as this has been covered by other bloggers. Though, for projects like this I always create a new resource group so I can easily delete all the resources so my Azure invoice doesn’t cause “bill shock”.

Azure Resource Group Creation blade

I also created the Azure IoT Hub before configuring the Percept device rather than via the Device provisioning process.

Azure Percept configuration assigning an Azure IoT Hub

The intern trialed different trays, camera orientations, and lighting as part of building a test rig on the living room floor. After some trial and error, he identified the optimal camera orientation (on top of the packing foam) and lighting (indirect sunlight with no shadows) for reliable inferencing. As this was a proof-of-concept project we limited the number of variables so we didn’t have to collect lots of images which the intern would then have to mark up.

Trialing image capture with M&M’s
Trialling Image capture with Cadbury Favourites

Azure Percept Studio + CustomVision.AI for capturing and marking up images

The intern created two Custom Vision projects, one for M&M’s and the other for Cadbury Favourites.

Azure M&M and Cadbury Favourites Percept Projects

The intern then spent an afternoon drawing minimum bounding rectangles (MBRs) around the different chocolates in the images he had collected.

M&M Size issue

The intern then decided to focus on the chocolate bars after realising they were much easier and faster to markup than the M&Ms.

Cadbury Favourites images before markup

Training

The intern repeatedly trained the model adding additional images and adjusting parameters until the results were “good enough”.

Fine-tuning the Configuration

After using the test rig one evening we found the performance of the model wasn’t great, so the intern collected more images with different lighting, shadows, chocolate bar placements, and orientations to improve the accuracy of the inferencing.

Manual reviewing of object detection results.

Inspecting the Inferencing Results

After several iterations the accuracy of the chocolate bar object detection model was acceptable I wanted to examine the telemetry that was being streamed to my Azure IoT Hub.

In Azure Percept Studio I could view (in a limited way) inferencing telemetry and check the quality and format of the results.

Azure Percept Studio device telemetry

I use Azure IoT Explorer on other projects to configure devices, view telemetry from devices, send messages to devices, view and modify device twin JSON etc. So I used it to inspect the inferencing results streamed to the Azure IoT Hub.

Azure IoT Explorer device telemetry

Summary

In an afternoon the intern had configured and trained a Custom Vision project for me that I could use to to build some “low code” integrations .

Project “Learnings”

If the image capture delay is too short there will be images with hands.

Captured image with interns hands

Though, the untrained model did identify the hands

The intern also discovered that by including images with “not favourites” the robustness of the model improved.

Cadbury Favourites with M&Ms

When I had to collect some more images for a blog post, I found the intern had consumed quite a few of the “props” and left the wrappers in the bottom of the Azure Percept packaging.

Cadbury Favourties wrappers

TTI V3 Connector Azure IoT Central Cloud to Device(C2D)

Handling Cloud to Device(D2C) Azure IoT Central messages (The Things Industries(TTI) downlink) is a bit more complex than Device To Cloud(D2C) messaging. The format of the command messages is reasonably well documented and I have already explored in detail with basic telemetry, basic commands, request commands, and The Things Industries Friendly commands and Digital Twin Definition Language(DTDL) support.

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.

Azure IoT Central Device Template

Azure Device Provisioning Service(DPS) when transient isn’t

After some updates to my Device Provisioning Service(DPS) code the RegisterAsync method was exploding with an odd exception.

TTI Webhook Integration running in desktop emulator

In the Visual Studio 2019 Debugger the exception text was “IsTransient = true” so I went and made a coffee and tried again.

Visual Studio 2019 Quickwatch displaying short from error message

The call was still failing so I dumped out the exception text so I had some key words to search for

Microsoft.Azure.Devices.Provisioning.Client.ProvisioningTransportException: AMQP transport exception
 ---> System.UnauthorizedAccessException: Sys
   at Microsoft.Azure.Amqp.ExceptionDispatcher.Throw(Exception exception)
   at Microsoft.Azure.Amqp.AsyncResult.End[TAsyncResult](IAsyncResult result)
   at Microsoft.Azure.Amqp.AmqpObject.OpenAsyncResult.End(IAsyncResult result)
   at Microsoft.Azure.Amqp.AmqpObject.EndOpen(IAsyncResult result)
   at Microsoft.Azure.Amqp.Transport.AmqpTransportInitiator.HandleTransportOpened(IAsyncResult result)
   at Microsoft.Azure.Amqp.Transport.AmqpTransportInitiator.OnTransportOpenCompete(IAsyncResult result)
--- End of stack trace from previous location ---
   at Microsoft.Azure.Devices.Provisioning.Client.Transport.AmqpClientConnection.OpenAsync(TimeSpan timeout, Boolean useWebSocket, X509Certificate2 clientCert, IWebProxy proxy, RemoteCertificateValidationCallback remoteCerificateValidationCallback)
   at Microsoft.Azure.Devices.Provisioning.Client.Transport.ProvisioningTransportHandlerAmqp.RegisterAsync(ProvisioningTransportRegisterMessage message, TimeSpan timeout, CancellationToken cancellationToken)
   --- End of inner exception stack trace ---
   at Microsoft.Azure.Devices.Provisioning.Client.Transport.ProvisioningTransportHandlerAmqp.RegisterAsync(ProvisioningTransportRegisterMessage message, TimeSpan timeout, CancellationToken cancellationToken)
   at Microsoft.Azure.Devices.Provisioning.Client.Transport.ProvisioningTransportHandlerAmqp.RegisterAsync(ProvisioningTransportRegisterMessage message, CancellationToken cancellationToken)
   at devMobile.IoT.TheThingsIndustries.AzureIoTHub.Integration.Uplink(HttpRequestData req, FunctionContext executionContext) in C:\Users\BrynLewis\source\repos\TTIV3AzureIoTConnector\TTIV3WebHookAzureIoTHubIntegration\TTIUplinkHandler.cs:line 245

I tried a lot of keywords and went and looked at the source code on github

One of the many keyword searches

Another of the many keyword searches

I then tried another program which did used the Device provisioning Service and it worked first time so it was something wrong with the code.

using (var securityProvider = new SecurityProviderSymmetricKey(deviceId, deviceKey, null))
{
	using (var transport = new ProvisioningTransportHandlerAmqp(TransportFallbackType.TcpOnly))
	{
		DeviceRegistrationResult result;

		ProvisioningDeviceClient provClient = ProvisioningDeviceClient.Create(
			Constants.AzureDpsGlobalDeviceEndpoint,
			 dpsApplicationSetting.GroupEnrollmentKey, <<= Should be _azureIoTSettings.DeviceProvisioningService.IdScope,
			securityProvider,
			transport);

		try
		{
				result = await provClient.RegisterAsync();
		}
		catch (ProvisioningTransportException ex)
		{
			logger.LogInformation(ex, "Uplink-DeviceID:{0} RegisterAsync failed IDScope and/or GroupEnrollmentKey invalid", deviceId);

			return req.CreateResponse(HttpStatusCode.Unauthorized);
		}

		if (result.Status != ProvisioningRegistrationStatusType.Assigned)
		{
			_logger.LogError("Uplink-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();

		logger.LogInformation("Uplink-DeviceID:{0} Azure IoT Hub connected (Device Provisioning Service)", deviceId);
	}
}

I then carefully inspected my source code and worked back through the file history and realised I had accidentally replaced the IDScope with the GroupEnrollment setting so it was never going to work i.e. IsTransient != true. So, for the one or two other people who get this error message check your IDScope and GroupEnrollment key make sure they are the right variables and that values they contain are correct.

TTI V3 Connector Azure IoT Central Device to Cloud(D2C)

This post is largely about adapting the output of The Things Industries(TTI) Low Power Protocol(LPP) payload formatter so that it can be injested by Azure IoT Central. The Azure function for processing TTI Uplink messages first deserialises the JSON payload discarding any LoRaWAN control messages and messages with empty payloads.

[Function("Uplink")]
public async Task<HttpResponseData> Uplink([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext executionContext)
{
	Models.PayloadUplink payload;
	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();

		try
		{
			payload = JsonConvert.DeserializeObject<Models.PayloadUplink>(payloadText);
		}
		catch(JsonException ex)
		{
			logger.LogInformation(ex, "Uplink-Payload Invalid JSON:{0}", payloadText);

			return req.CreateResponse(HttpStatusCode.BadRequest);
		}

		if (payload == null)
		{
			logger.LogInformation("Uplink-Payload invalid:{0}", payloadText);

			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 message", applicationId, deviceId, payload.UplinkMessage.PayloadRaw);

			return req.CreateResponse(HttpStatusCode.UnprocessableEntity);
		}

		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))
		{
...		
		}

		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)
		{
			EnumerateChildren(telemetryEvent, 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);
	}

	return req.CreateResponse(HttpStatusCode.OK);
}

If the message has been successfully decoded by a payload formatter the PayloadDecoded contents will be “grafted” into the Azure IoT Central Telemetry message.

TTI JSON GPS position format

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

To test the telemetry message JSON I created an Azure IoT Central Device Template which had a “capability type” of Location.

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

This post uses a lot of the work done for my The Things Network V2 integration. I also found the first time a device connected to the Azure IoT Central Azure IoT hub (using the Azure IoT Central Device Provisioning Service(DPS) to get the connection string) there was always an exception.

Microsoft.Azure.Devices.Client.Exceptions.IotHubException: error(condition:com.microsoft:connection-closed-on-new-connection,description:Backend initiated disconnection.

TTI V3 Gateway Azure IoT Central first call exception

This exception occurs when the SetMethodDefaultHandlerAsync method is called which is a bit odd. This exception does not occur when I use Device Provisioning Service(DPS) and Azure IoT Hub instances I have provisioned.

TTI V3 Connector Cloud to Device(C2D)

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.

  • Confirmation – True/False
  • Queue – Push/Replace
  • Priority – Lowest/Low/BelowNormal/Normal/AboveNormal/High/Highest

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.

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
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.

TTI V3 Connector Device Provisioning Service(DPS) support

The previous versions of my Things Network Industries(TTI) and The Things Network(TTN) connectors supported the Azure IoT Hub Device Provisioning Service(DPS) with Symmetric Key Attestation(SAS) to “automagically” setup the LoRaWAN devices in a TTI Application.(See my V2 Gateway DPS setup post for more detail).

Azure Device Provisioning Service configuring Azure IoT Hubs

I used an “evenly weighted distribution” to spread the devices across five Azure IoT Hubs.

Azure IoT Hub no registered devices

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.