RAK7258 Local server and Message Queuing Telemetry Transport(MQTT)

This post was originally about getting the built in Network Server of my RAKWireless RAK7258 WisGate Edge Lite to connect to an Azure IoT Hub or Azure IoT Central. The RAK7258 had been connected to The Things Industries(TTI) network so I updated the firmware and checked the “mode” in the LoRaWAN Network settings.

RAK 7258 LoRaWAN Network settings

Azure IoT Hub is not a fully featured MQTT broker so I initially looked at running Eclipse Mosquitto or HiveMQ locally but this seemed like a lot of effort for a Proof of Concept(PoC).

RAK 7258 Network Server Global Integration settings

I have used MQTTNet in a few other projects (The Things Network(TTN) V3 Azure IoT Connector, The Things Network V2 MQTT SQL Connector, Windows 10 IoT Core MQTT Field gateway etc.) and there was a sample application which showed ho to build a simple server so that became my preferred approach.

I then started exploring how applications and devices are provisioned in the RAK Network Server.

RAK 7258 Network Server applications list

The network server software has “unified” and “separate” “Device authentication mode”s and will “auto Add LoRa Device”s if enabled.

RAK 7258 Network Server Separate Application basic setup
RAK 7258 Network Server Separate Application device basic setup
RAK 7258 Network Server Unified Application device basic setup

Applications also have configurable payload formats(raw & LPP) and integrations (uplink messages plus join, ack, and device notifications etc.)

RAK7258 live device data display

In the sample server I could see how ValidatingConnectionAsync was used to check the clientID, username and password when a device connected. I just wanted to display messages and payloads without having to use an MQTT client and it looked like InterceptingPublishAsync was a possible solution.

But the search results were a bit sparse…

InterceptingPublishAsync + MQTTNet search results

After some reading the MQTTNet documentation and some experimentation I could display the message payload (same as in the live device data display) in a “nasty” console application.

namespace devMobile.IoT.RAKWisgate.ServerBasic
{
   using System;
	using System.Threading.Tasks;

   using MQTTnet;
   using MQTTnet.Protocol;
   using MQTTnet.Server;

   public static class Program
   {
      static async Task Main(string[] args)
      {
         var mqttFactory = new MqttFactory();

         var mqttServerOptions = new MqttServerOptionsBuilder()
             .WithDefaultEndpoint()
             .Build();

         using (var mqttServer = mqttFactory.CreateMqttServer(mqttServerOptions))
         {
            mqttServer.InterceptingPublishAsync += e =>
            {
               Console.WriteLine($"Client:{e.ClientId} Topic:{e.ApplicationMessage.Topic} {e.ApplicationMessage.ConvertPayloadToString()}");

               return Task.CompletedTask;
            };

            mqttServer.ValidatingConnectionAsync += e =>
            {
               if (e.ClientId != "RAK Wisgate7258")
               {
                  e.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid;
               }

               if (e.Username != "ValidUser")
               {
                  e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
               }

               if (e.Password != "TopSecretPassword")
               {
                  e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
               }

               return Task.CompletedTask;
            };

            await mqttServer.StartAsync();

            Console.WriteLine("Press Enter to exit.");
            Console.ReadLine();

            await mqttServer.StopAsync();
         }
      }
   }
}
MQTTNet based console application displaying device payloads

The process of provisioning Applications and Devices is quite different (The use of the AppEUI/JoinEUI is odd) to The Things Network(TTN) and other platforms I have used so I will explore this some more in future post(s).

.NET Core web API + Dapper – MultiMapping

Shaping recordsets with SplitOn

Sometimes there is no easy way to build a “list of lists” using the contents of multiple database tables. I have run into this problem a few times especially when building webby services which query the database of a “legacy” (aka. production) system.

Retrieving a list of StockGroups and their StockItems from the World Wide Importers database was one of the better “real world” examples I could come up with.

SQL Server Management Studio Diagram showing relationships of tables

There is a fair bit of duplication (StockGroupID, StockGroupName) in the results set

SQL Server Management Studio StockItems-StockItemStockGroups-StockGroups query and results

There were 442 rows in the results set and 227 StockItems in the database so I ordered the query results by StockItemID and confirmed that there were many StockItems in several StockGroups.

public class StockItemListDtoV1
{
	public int Id { get; set; }

	public string Name { get; set; }

	public decimal RecommendedRetailPrice { get; set; }

	public decimal TaxRate { get; set; }
}

public class StockGroupStockItemsListDto
{
	StockGroupStockItemsListDto()
	{
		StockItems = new List<StockItemListDto>();
	}

	public int StockGroupID { get; set; }

	public string StockGroupName { get; set; }

	public List<StockItemListDto> StockItems { get; set; }
}

My initial version uses a Generic List for a StockGroup’s StockItems which is most probably not a good idea.

[Route("api/[controller]")]
[ApiController]
public class InvoiceQuerySplitOnController : ControllerBase
{
	private readonly string connectionString;
	private readonly ILogger<InvoiceQuerySplitOnController> logger;

	public InvoiceQuerySplitOnController(IConfiguration configuration, ILogger<InvoiceQuerySplitOnController> logger)
	{
		this.connectionString = configuration.GetConnectionString("WorldWideImportersDatabase");

		this.logger = logger;
	}

	[HttpGet]
	public async Task<ActionResult<IEnumerable<StockGroupStockItemsListDto>>> Get()
	{
		IEnumerable<StockGroupStockItemsListDto> response = null;

		try
		{
			using (SqlConnection db = new SqlConnection(this.connectionString))
			{
				var stockGroups = await db.QueryAsync<StockGroupStockItemsListDto, StockItemListDto, StockGroupStockItemsListDto>(
					sql: @"SELECT [StockGroups].[StockGroupID] as 'StockGroupID'" +
								",[StockGroups].[StockGroupName]" +
								",[StockItems].StockItemID as 'ID'" +
								",[StockItems].StockItemName as 'Name'" +
								",[StockItems].TaxRate" +
								",[StockItems].RecommendedRetailPrice " +
							"FROM [Warehouse].[StockGroups] " +
							"INNER JOIN[Warehouse].[StockItemStockGroups] ON ([StockGroups].[StockGroupID] = [StockItemStockGroups].[StockGroupID])" +
							"INNER JOIN[Warehouse].[StockItems] ON ([Warehouse].[StockItemStockGroups].[StockItemID] = [StockItems].[StockItemID])",
					(stockGroup, stockItem) =>
					{
						// Not certain I think using a List<> here is a good idea...
						stockGroup.StockItems.Add(stockItem);
						return stockGroup;
					},
				splitOn: "ID",
				commandType: CommandType.Text);

			response = stockGroups.GroupBy(p => p.StockGroupID).Select(g =>
			{
				var groupedStockGroup = g.First();
				groupedStockGroup.StockItems = g.Select(p => p.StockItems.Single()).ToList();
				return groupedStockGroup;
			});
		}
	}
	catch (SqlException ex)
	{
		logger.LogError(ex, "Retrieving S, Invoice Lines or Stock Item Transactions");

		return this.StatusCode(StatusCodes.Status500InternalServerError);
	}

	return this.Ok(response);
}

The MultiMapper syntax always trips me up and it usually takes a couple of attempts to get it to work.

List of StockGroups with StockItems

I have extended my DapperTransient module adding WithRetry versions of the 14 MultiMapper methods.

NOTE : Error Handling approach has changed

.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

Azure Smartish Edge Camera – Background Service

This is a “note to self” post about deploying a .NET Core Worker Service to a Raspberry PI 4B 8G running Raspberry PI OS (Bullseye). After reading many posts, then a lot of trial and error this approach appeared to work reliably for my system configuration.(Though YMMV with other distros etc.)

The first step was to create a new Worker Service project in Visual Studio 2019

VS 2019 Add new Worker Service project wizard
VS 2019 Add new Worker Service project name
Visual Studio 2019 NuGet management

I intentionally did not update the Microsoft.Extensions.Hosting and Microsoft.Extensions.Hosting.Systemd (for UseSystemd) NuGet packages for my initial development.

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

namespace devMobile.IoT.MachineLearning.AzureIoTSmartEdgeCameraService
{
	public class Program
	{
		public static void Main(string[] args)
		{
			CreateHostBuilder(args).Build().Run();
		}

		public static IHostBuilder CreateHostBuilder(string[] args) =>
			 Host.CreateDefaultBuilder(args)
					.UseSystemd()
				  .ConfigureServices((hostContext, services) =>
				  {
					  services.AddHostedService<Worker>();
				  });
	}
}

program.cs

using System;
using System.Threading;
using System.Threading.Tasks;

using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace devMobile.IoT.MachineLearning.AzureIoTSmartEdgeCameraService
{
	public class Worker : BackgroundService
	{
		private readonly ILogger<Worker> _logger;

		public Worker(ILogger<Worker> logger)
		{
			_logger = logger;
		}

		protected override async Task ExecuteAsync(CancellationToken stoppingToken)
		{
			while (!stoppingToken.IsCancellationRequested)
			{
				_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);

				await Task.Delay(1000, stoppingToken);
			}
		}
	}
}

Worker.cs

The first step was to create a new directory (AzureIoTSmartEdgeCameraService) on the device. In Visual Studio 2019 I “published” my application and copied the contents of the “publish” folder to the Raspberry PI with Winscp. (This intermediary folder was to avoid issues with the permissions of the /usr/sbin/ & etc/systemd/system folders)

Using Winscp to copy files to AzureIoTSmartEdgeCameraService folder on my device
Install service

Test in application directory
	/home/pi/.dotnet/dotnet AzureIoTSmartEdgeCameraService.dll

Make service directory
	sudo mkdir /usr/sbin/AzureIoTSmartEdgeCameraService

Copy files to service directory
	sudo cp *.* /usr/sbin/AzureIoTSmartEdgeCameraService

Copy .service file to systemd folderclear
	sudo cp AzureIoTSmartEdgeCameraService.service /etc/systemd/system/AzureIoTSmartEdgeCameraService.service
 
Force reload of systemd configuration
	sudo systemctl daemon-reload

Start the Azure IoT SmartEdge Camera service
	sudo systemctl start AzureIoTSmartEdgeCameraService
Installing and starting the AzureIoTSmartEdgeCameraService
Uninstall service
	sudo systemctl stop AzureIoTSmartEdgeCameraService

	sudo rm /etc/systemd/system/AzureIoTSmartEdgeCameraService.service

	sudo systemctl daemon-reload

	sudo rm /usr/sbin/AzureIoTSmartEdgeCameraService/*.*

	sudo rmdir /usr/sbin/AzureIoTSmartEdgeCameraService

	See what is happening
	journalctl -xe

Stopping and removing the AzureIoTSmartEdgeCameraService

It took a lot of attempts to get a clean install then uninstall for the screen captures.

.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

libcamera-jpeg on Raspberry Pi OS Bullseye Duration

The image capture process was taking about 5 seconds which a bit longer than I was expecting.

libcamera-jpeg -o rotated.jpg --rotation 180

The libcamera-jpeg program has a lot of command line parameters.

pi@raspberrypi4a:~ $ libcamera-jpeg --help
Valid options are:
  -h [ --help ] [=arg(=1)] (=0)         Print this help message
  --version [=arg(=1)] (=0)             Displays the build version number
  -v [ --verbose ] [=arg(=1)] (=0)      Output extra debug and diagnostics
  -c [ --config ] [=arg(=config.txt)]   Read the options from a file. If no filename is specified, default to
                                        config.txt. In case of duplicate options, the ones provided on the command line
                                        will be used. Note that the config file must only contain the long form
                                        options.
  --info-text arg (=#%frame (%fps fps) exp %exp ag %ag dg %dg)
                                        Sets the information string on the titlebar. Available values:
                                        %frame (frame number)
                                        %fps (framerate)
                                        %exp (shutter speed)
                                        %ag (analogue gain)
                                        %dg (digital gain)
                                        %rg (red colour gain)
                                        %bg (blue colour gain)
                                        %focus (focus FoM value)
                                        %aelock (AE locked status)
  --width arg (=0)                      Set the output image width (0 = use default value)
  --height arg (=0)                     Set the output image height (0 = use default value)
  -t [ --timeout ] arg (=5000)          Time (in ms) for which program runs
  -o [ --output ] arg                   Set the output file name
  --post-process-file arg               Set the file name for configuring the post-processing
  --rawfull [=arg(=1)] (=0)             Force use of full resolution raw frames
  -n [ --nopreview ] [=arg(=1)] (=0)    Do not show a preview window
  -p [ --preview ] arg (=0,0,0,0)       Set the preview window dimensions, given as x,y,width,height e.g. 0,0,640,480
  -f [ --fullscreen ] [=arg(=1)] (=0)   Use a fullscreen preview window
  --qt-preview [=arg(=1)] (=0)          Use Qt-based preview window (WARNING: causes heavy CPU load, fullscreen not
                                        supported)
  --hflip [=arg(=1)] (=0)               Request a horizontal flip transform
  --vflip [=arg(=1)] (=0)               Request a vertical flip transform
  --rotation arg (=0)                   Request an image rotation, 0 or 180
  --roi arg (=0,0,0,0)                  Set region of interest (digital zoom) e.g. 0.25,0.25,0.5,0.5
  --shutter arg (=0)                    Set a fixed shutter speed
  --analoggain arg (=0)                 Set a fixed gain value (synonym for 'gain' option)
  --gain arg                            Set a fixed gain value
  --metering arg (=centre)              Set the metering mode (centre, spot, average, custom)
  --exposure arg (=normal)              Set the exposure mode (normal, sport)
  --ev arg (=0)                         Set the EV exposure compensation, where 0 = no change
  --awb arg (=auto)                     Set the AWB mode (auto, incandescent, tungsten, fluorescent, indoor, daylight,
                                        cloudy, custom)
  --awbgains arg (=0,0)                 Set explict red and blue gains (disable the automatic AWB algorithm)
  --flush [=arg(=1)] (=0)               Flush output data as soon as possible
  --wrap arg (=0)                       When writing multiple output files, reset the counter when it reaches this
                                        number
  --brightness arg (=0)                 Adjust the brightness of the output images, in the range -1.0 to 1.0
  --contrast arg (=1)                   Adjust the contrast of the output image, where 1.0 = normal contrast
  --saturation arg (=1)                 Adjust the colour saturation of the output, where 1.0 = normal and 0.0 =
                                        greyscale
  --sharpness arg (=1)                  Adjust the sharpness of the output image, where 1.0 = normal sharpening
  --framerate arg (=30)                 Set the fixed framerate for preview and video modes
  --denoise arg (=auto)                 Sets the Denoise operating mode: auto, off, cdn_off, cdn_fast, cdn_hq
  --viewfinder-width arg (=0)           Width of viewfinder frames from the camera (distinct from the preview window
                                        size
  --viewfinder-height arg (=0)          Height of viewfinder frames from the camera (distinct from the preview window
                                        size)
  --tuning-file arg (=-)                Name of camera tuning file to use, omit this option for libcamera default
                                        behaviour
  --lores-width arg (=0)                Width of low resolution frames (use 0 to omit low resolution stream
  --lores-height arg (=0)               Height of low resolution frames (use 0 to omit low resolution stream
  -q [ --quality ] arg (=93)            Set the JPEG quality parameter
  -x [ --exif ] arg                     Add these extra EXIF tags to the output file
  --timelapse arg (=0)                  Time interval (in ms) between timelapse captures
  --framestart arg (=0)                 Initial frame counter value for timelapse captures
  --datetime [=arg(=1)] (=0)            Use date format for output file names
  --timestamp [=arg(=1)] (=0)           Use system timestamps for output file names
  --restart arg (=0)                    Set JPEG restart interval
  -k [ --keypress ] [=arg(=1)] (=0)     Perform capture when ENTER pressed
  -s [ --signal ] [=arg(=1)] (=0)       Perform capture when signal received
  --thumb arg (=320:240:70)             Set thumbnail parameters as width:height:quality
  -e [ --encoding ] arg (=jpg)          Set the desired output encoding, either jpg, png, rgb, bmp or yuv420
  -r [ --raw ] [=arg(=1)] (=0)          Also save raw file in DNG format
  --latest arg                          Create a symbolic link with this name to most recent saved file
  --immediate [=arg(=1)] (=0)           Perform first capture immediately, with no preview phase
pi@raspberrypi4a:~ $

My libcamera-jpeg application is run “headless” so I tried turning off the image preview functionality.

libcamera-jpeg -o rotatednopreview.jpg --nopreview

When I ran libcamera-jpeg in a console windows or my application this didn’t appear to make any noticeable difference.

libcamera-jpeg run from the command line with –nopreview

libcamera-jpeg run by my application with –nopreview

I then had another look at the libcamera-jpeg command line parameters to see if any looked useful for reducing the time that it took to take a save an image and this one caught my attention.

I had assumed the delay was related to how long the preview window was displayed.

libcamera-jpeg run from the command line with –nopreview –t1

I modified the application (V5) then ran it from the command line and the time reduced to less than a second.

private static void ImageUpdateTimerCallback(object state)
{
	try
	{
		Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss} Image update start");

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

		Console.WriteLine($" {DateTime.UtcNow:yy-MM-dd HH:mm:ss} Image capture start");

		using (Process process = new Process())
		{
			process.StartInfo.FileName = @"libcamera-jpeg";
			// V1 it works
			//process.StartInfo.Arguments = $"-o {_applicationSettings.ImageFilenameLocal}";
			// V3a Image right way up
			//process.StartInfo.Arguments = $"-o {_applicationSettings.ImageFilenameLocal} --vflip --hflip";
			// V3b Image right way up
			//process.StartInfo.Arguments = $"-o {_applicationSettings.ImageFilenameLocal} --rotation 180";
			// V4 Image no preview
			//process.StartInfo.Arguments = $"-o {_applicationSettings.ImageFilenameLocal} --rotation 180 --nopreview";
			// V5 Image no preview, no timeout
			process.StartInfo.Arguments = $"-o {_applicationSettings.ImageFilenameLocal} --nopreview -t1 --rotation 180";
			//process.StartInfo.RedirectStandardOutput = true;
			// V2 No diagnostics
			process.StartInfo.RedirectStandardError = true;
			//process.StartInfo.UseShellExecute = false;
			//process.StartInfo.CreateNoWindow = true; 

			process.Start();

			if (!process.WaitForExit(10000) || (process.ExitCode != 0))
			{
				Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss} Image update failure {process.ExitCode}");
			}
		}

		Console.WriteLine($" {DateTime.UtcNow:yy-MM-dd HH:mm:ss} Image capture done");
	}
	catch (Exception ex)
	{
		Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss} Image update error {ex.Message}");
	}
	finally
	{
		_cameraBusy = false;
	}
}
libcamera-jpeg run by my application with –nopreview -t1

The image capture process now takes less that a second which is much better (but not a lot less than retrieving an image from one of my security cameras).

libcamera on Raspberry Pi OS Bullseye

This is a “note to self” post about using libcamera(replacement for raspistill) on my Raspberry PI 4 Model B to capture an image from my Raspberry Pi Camera Module 2 with an application built with .NET Core.

I wanted one of my ML.Net demos to use the Raspberry PI Camera rather than a security camera (so it was more portable) but it took a bit more work than I expected.

Version 1 used Process.Start to launch the libcamera-jpeg application with a command line to store an image to the local file system.

libcamera-jpeg -o latest.jpg
libcamera-jpeg with diagnostic information displayed

There was a lot of diagnostic information which I didn’t want displayed so after reading many stackoverflow posts (lots of different approaches none of which worked in my scenario), then some trial and error I found that I only had to enable RedirectStandardError.

libcamera-jpeg without diagnostic information displayed

At this point there was a lot less noise but the image was upside down.

Inverted picture of my 30th anniversary Mini Cooper in the backyard

I then added a vertical flip to the command line parameters

libcamera-jpeg -o latest.jpg --vflip
My 30th anniversary Mini Cooper in the backyard

The image was backwards so I added a horizontal flip to the commandline parameters

libcamera-jpeg -o latest.jpg --vflip --hflip

or

libcamera-jpeg -o latest.jpg --rotation 180
My 30th anniversary Mini Cooper in the backyard with the correct orientation

The libcamera code is in a Timer callback so I added the _cameraBusy boolean flag to stop reentrancy problems.

private static void ImageUpdateTimerCallback(object state)
{
	try
	{
		Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss} Image update start");

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

		Console.WriteLine($" {DateTime.UtcNow:yy-MM-dd HH:mm:ss} Image capture start");

		using (Process process = new Process())
		{
			process.StartInfo.FileName = @"libcamera-jpeg";
			// V1 it works
			//process.StartInfo.Arguments = $"-o {_applicationSettings.ImageFilenameLocal}";
			// V3 Image right way up
			//process.StartInfo.Arguments = $"-o {_applicationSettings.ImageFilenameLocal} --vflip";
			// V3 Image right way round
			process.StartInfo.Arguments = $"-o {_applicationSettings.ImageFilenameLocal} --vflip --hflip";
			//process.StartInfo.RedirectStandardOutput = true;
			// V2 No diagnostics
			process.StartInfo.RedirectStandardError = true;
			//process.StartInfo.UseShellExecute = false;
			//process.StartInfo.CreateNoWindow = true; 

			process.Start();

			if (!process.WaitForExit(10000) || (process.ExitCode != 0))
			{
				Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss} Image update failure {process.ExitCode}");
			}
		}

		Console.WriteLine($" {DateTime.UtcNow:yy-MM-dd HH:mm:ss} Image capture done");
	}
	catch (Exception ex)
	{
		Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss} Image update error {ex.Message}");
	}
	finally
	{
		_cameraBusy = false;
	}
}

This was the simplest way I could get an image onto the local file system without lots of dependencies on third party libraries. The image capture process takes about 5 seconds which a bit longer than I was expecting.

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.

TTI V3 Connector Minimalist Cloud to Device(C2D)

In a previous version of my Things Network Industries(TTI) The Things Network(TTN) connector I queried the The Things Stack(TTS) Application Programing Interface(API) to get a list of Applications and their Devices. For a large number of Applications and/or Devices this process could take many 10’s of seconds. Application and Device creation and deletion then had to be tracked to keep the AzureDeviceClient connection list current, which added significant complexity.

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

Azure IoT Explorer Clod to Device message with LoRaWAN Port number custom parameter

When a LoRaWAN device sends an Uplink message a session is established using Advanced Message Queuing Protocol(AMQP) so connections can be multiplexed)

I used Azure IoT Explorer to send Cloud to Device messages to the Azure IoT Hub (to initiate the sending of a downlink message to the Device by the Connector) after simulating a TTN uplink message with Telerik Fiddler and a modified TTN sample payload.

BEWARE – TTN URLs and Azure IoT Hub device identifiers are case sensitive

...
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, 
                    new ITransportSettings[]
                    {
                        new AmqpTransportSettings(TransportType.Amqp_Tcp_Only)
                        {
                            AmqpConnectionPoolSettings = new AmqpConnectionPoolSettings()
                            {
                                Pooling = true,
                            }
                        }
                    });

   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);
   }

   Models.AzureIoTHubReceiveMessageHandlerContext context = new Models.AzureIoTHubReceiveMessageHandlerContext()
   { 
      DeviceId = deviceId,
      ApplicationId = applicationId,
      WebhookId = _configuration.GetSection("TheThingsIndustries").GetSection("WebhookId").Value,
      WebhookBaseURL = _configuration.GetSection("TheThingsIndustries").GetSection("WebhookBaseURL").Value,
      ApiKey = _configuration.GetSection("TheThingsIndustries").GetSection("APiKey").Value,
   };      

   await deviceClient.SetReceiveMessageHandlerAsync(AzureIoTHubClientReceiveMessageHandler, context);

   await deviceClient.SetMethodDefaultHandlerAsync(AzureIoTHubClientDefaultMethodHandler, context);
 }

...

An Azure IoT Hub can invoke methods(synchronous) or send messages(asynchronous) to a device for processing. The Azure IoT Hub DeviceClient has two methods SetMethodDefaultHandlerAsync and SetReceiveMessageHandlerAsync which enable the processing of direct methods and messages.

private async Task<MethodResponse> AzureIoTHubClientDefaultMethodHandler(MethodRequest methodRequest, object userContext)
{
	if (methodRequest.DataAsJson != null)
	{
		_logger.LogWarning("AzureIoTHubClientDefaultMethodHandler name:{0} payload:{1}", methodRequest.Name, methodRequest.DataAsJson);
	}
	else
	{
		_logger.LogWarning("AzureIoTHubClientDefaultMethodHandler name:{0} payload:NULL", methodRequest.Name);
	}

	return new MethodResponse(404);
}

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");
		}
	}
}

If the message body contains a valid JavaScript Object Notation(JSON) payload it is “spliced” into the DownLink message decoded_payload field otherwise the Base64 encoded frm_payload is populated.

The Things “Industries Live” data tab downlink message

The SetReceiveMessageHandlerAsync context information is used to construct a TTN downlink message payload and request URL(with default queuing, message priority and confirmation options)

Arduino Serial Monitor displaying Uplink and Downlink messages

TTI V3 Connector Azure Storage Queues Paused

After running my The Things Industries(TTI) V3 HTTPStorageQueueOutput application for a week I think there are some problems with my approach so I have paused development while I build another HTTPTrigger Azure Functions based Proof of Concept(PoC).

The HTTPTrigger and Azure Storage Queue OutputBinding based code which inserts messages into an Azure Storage Queue was minimal

[StorageAccount("AzureWebJobsStorage")]
public static class Webhooks
{
	[Function("Uplink")]
	public static async Task<HttpTriggerUplinkOutputBindingType> Uplink([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext context)
	{
		var logger = context.GetLogger("UplinkMessage");

		logger.LogInformation("Uplink processed");
			
		var response = req.CreateResponse(HttpStatusCode.OK);

		return new HttpTriggerUplinkOutputBindingType()
		{
			Name = await req.ReadAsStringAsync(),
			HttpReponse = response
		};
	}
}

With Azure Storage Explorer I could inspect uplink, queued, sent, and acknowledgment(ACK) messages. It was difficult to generate failed and Negative Acknowledgement (Nack) and failed messages

Azure Storage Explorer displaying Uplink messages
Azure Storage Explorer displaying queued messages
Azure Storage Explorer displaying sent messages
Azure Storage Explorer Displaying Ack messages

After some experimentation I realised that I had forgotten that the order of message processing was important e.g. a TTI Queued message should be processed before the associated Ack. This could (and did happen) because I had a queue for each message type and in addition the Azure Queue Storage trigger binding would use parallel execution to process backlogs of messages. My approach caused issues with both intra and inter queue message ordering