YoloV8 ONNX – Nvidia Jetson Orin Nano™ GPU CUDA Inferencing

The Seeedstudio reComputer J3011 has two processors an ARM64 CPU and an Nividia Jetson Orin 8G. To speed up inferencing with the Nividia Jetson Orin 8G with Compute Unified Device Architecture (CUDA) I built an Open Neural Network Exchange(ONNX) CUDA Execution Provider.

The Open Neural Network Exchange(ONNX) model used was trained on Roboflow Universe by Ugur ozdemir dataset which has 23696 images.

// load the app settings into configuration
var configuration = new ConfigurationBuilder()
      .AddJsonFile("appsettings.json", false, true)
.Build();

_applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>();

Console.WriteLine($" {DateTime.UtcNow:yy-MM-dd HH:mm:ss.fff} YoloV8 Model load: {_applicationSettings.ModelPath}");

YoloV8Builder builder = new YoloV8Builder();

builder.UseOnnxModel(_applicationSettings.ModelPath);

if (_applicationSettings.UseCuda)
{
   builder.UseCuda(_applicationSettings.DeviceId) ;
}

if (_applicationSettings.UseTensorrt)
{
   builder.UseTensorrt(_applicationSettings.DeviceId);
}

/*
builder.WithConfiguration(c =>
{
});
*/

/*
builder.WithSessionOptions(new Microsoft.ML.OnnxRuntime.SessionOptions()
{

});
*/

using (var image = await SixLabors.ImageSharp.Image.LoadAsync<Rgba32>(_applicationSettings.ImageInputPath))
using (var predictor = builder.Build())
{
   var result = await predictor.DetectAsync(image);

   Console.WriteLine();
   Console.WriteLine($"Speed: {result.Speed}");
   Console.WriteLine();

   foreach (var prediction in result.Boxes)
   {
      Console.WriteLine($" Class {prediction.Class} {(prediction.Confidence * 100.0):f1}% X:{prediction.Bounds.X} Y:{prediction.Bounds.Y} Width:{prediction.Bounds.Width} Height:{prediction.Bounds.Height}");
   }

   Console.WriteLine();

   Console.WriteLine($" {DateTime.UtcNow:yy-MM-dd HH:mm:ss.fff} Plot and save : {_applicationSettings.ImageOutputPath}");

   using (var imageOutput = await result.PlotImageAsync(image))
   {
      await imageOutput.SaveAsJpegAsync(_applicationSettings.ImageOutputPath);
   }
}

When configured to run the YoloV8.Coprocessor.Detect.Image on the ARM64 CPU the average inference time was 729 mSec.

The first time ran the YoloV8.Coprocessor.Detect.Image application configured to use CUDA for inferencing it failed badly.

The YoloV8.Coprocessor.Detect.Image application was then configured to use CUDA and the average inferencing time was 85mSec.

It took a couple of weeks to get the YoloV8.Coprocessor.Detect.Image application inferencing on the Nividia Jetson Orin 8G coprocessor and this will be covered in detail in another posts.

YoloV8 ONNX – Nvidia Jetson Orin Nano™ ARM64 CPU Inferencing

I configured the demonstration Ultralytics YoloV8 object detection(yolov8s.onnx) console application to process a 1920×1080 image from a security camera on my desktop development box (13th Gen Intel(R) Core(TM) i7-13700 2.10 GHz with 32.0 GB)

Object Detection sample application running on my development box

A Seeedstudio reComputer J3011 uses a Nividia Jetson Orin 8G and looked like a cost-effective platform to explore how a dedicated Artificial Intelligence (AI) co-processor could reduce inferencing times.

To establish a “baseline” I “published” the demonstration application on my development box which created a folder with all the files required to run the application on the Seeedstudio reComputer J3011 ARM64 CPU. I had to manually merge the “User Secrets” and appsettings.json files so the camera connection configuration was correct.

The runtimes folder contained a number of folders with the native runtime files for the supported Open Neural Network Exchange(ONNX) platforms

Object Detection application publish runtimes folder

This Nividia Jetson Orin ARM64 CPU requires the linux-arm64 ONNX runtime which was “automagically” detected. (in previous versions of ML.Net the native runtime had to be copied to the execution directory)

Linux ONNX ARM64 runtime

The final step was to use the demonstration Ultralytics YoloV8 object detection(yolov8s.onnx) console application to process a 1920×1080 image from a security camera on the reComputer J3011 (6-core Arm® Cortex®64-bit CPU 1.5Ghz processor)

Object Detection sample application running on my Seeedstudio reComputer J3011

When I averaged the pre-processing, inferencing and post-processing times for both devices over 20 executions my development box was much faster which was not a surprise. Though the reComputer J3011 post processing times were a bit faster than I was expecting

ARM64 CPU Preprocess 0.05s Inference 0.31s Postprocess 0.05

Myriota Connector – Azure IoT Central Downlink Methods

This post is about Azure IoT Central downlink methods and should be read in conjunction with the Myriota Connector – Azure IoT Central Downlink Methods post. My Myriota Sense and Locate template has 4 commands and in this post, I have focused on the fan speed command.

Sense and Locate Azure IoT Central Template

The Myriota Connector only supports Direct Methods which provide immediate confirmation of the result being queued by the Myriota Cloud API. The Myriota (API) control message send method responds with 400 Bad Request if there is already a message being sent to a device.

Myriota Azure Function Environment Variable configuration

The fan speed downlink payload formatter is specified in the Azure Function Environment Variables.

Sense and Locate Azure IoT Central Template Fan Speed Enumeration

The fan speed value in the message payload is configured in the fan speed enumeration.

Sense and Locate Azure IoT Central Command Fan Speed Selection

The FanSpeed.cs payload formatter extracts the FanSpeed value from the Javascript Object Notation(JSON) payload and returns a two-byte array containing the message type and speed of the fan.

using System;
using System.Collections.Generic;

using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

public class FormatterDownlink : PayloadFormatter.IFormatterDownlink
{
   public byte[] Evaluate(string terminalId, string methodName, JObject payloadJson, byte[] payloadBytes)
   {
      byte? status = payloadJson.Value<byte?>("FanSpeed");

      if (!status.HasValue)
      {
         return new byte[] { };
      }

      return new byte[] { 1, status.Value };
   }
}

Sense and Locate Azure IoT Central Command Fan Speed History

Each Azure Application Insights log entry starts with the TerminalID (to simplify searching for all the messages related to device) and the requestId a Globally Unique Identifier (GUID) to simplify searching for all the “steps” associated with sending/receiving a message) with the rest of the logging message containing “step” specific diagnostic information.

Sense and Locate Azure IoT Central Command Fan Speed Application Insights

In the Myriota Device Manager the status of Control Messages can be tracked and they can be cancelled if in the “pending” state.

Myriota Control Message status Pending

A Control Message can take up to 24hrs to be delivered and confirmation of delivery has to be implemented by the application developer.

Azure Event Grid MQTT-With HiveMQ & MQTTnet Clients

Most of the examples of connecting to Azure Event Grid’s MQTT broker use MQTTnet so for a bit of variety I started with a hivemq-mqtt-client-dotnet based client. (A customer had been evaluating HiveMQ for a project which was later cancelled)

BEWARE – ClientID parameter is case sensitive.

The HiveMQ client was “inspired” by the How to Guides > Custom Client Certificates documentation.

class Program
{
   private static Model.ApplicationSettings _applicationSettings;
   private static HiveMQClient _client;
   private static bool _publisherBusy = false;

   static async Task Main()
   {
      Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss} Hive MQ client starting");

      try
      {
         // load the app settings into configuration
         var configuration = new ConfigurationBuilder()
               .AddJsonFile("appsettings.json", false, true)
               .AddUserSecrets<Program>()
         .Build();

         _applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>();

         var optionsBuilder = new HiveMQClientOptionsBuilder();

         optionsBuilder
            .WithClientId(_applicationSettings.ClientId)
            .WithBroker(_applicationSettings.Host)
            .WithPort(_applicationSettings.Port)
            .WithUserName(_applicationSettings.UserName)
            .WithCleanStart(_applicationSettings.CleanStart)
            .WithClientCertificate(_applicationSettings.ClientCertificateFileName, _applicationSettings.ClientCertificatePassword)
            .WithUseTls(true);

         using (_client = new HiveMQClient(optionsBuilder.Build()))
         {
            _client.OnMessageReceived += OnMessageReceived;

            var connectResult = await _client.ConnectAsync();
            if (connectResult.ReasonCode != ConnAckReasonCode.Success)
            {
               throw new Exception($"Failed to connect: {connectResult.ReasonString}");
            }

            Console.WriteLine($"Subscribed to Topic");
            foreach (string topic in _applicationSettings.SubscribeTopics.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries))
            {
               var subscribeResult = await _client.SubscribeAsync(topic, _applicationSettings.SubscribeQualityOfService);

               Console.WriteLine($" Topic:{topic} Result:{subscribeResult.Subscriptions[0].SubscribeReasonCode}");
            }
   }
//...
}
HiveMQ Client console application output

The MQTTnet client was “inspired” by the Azure MQTT .NET Application sample

class Program
{
   private static Model.ApplicationSettings _applicationSettings;
   private static IMqttClient _client;
   private static bool _publisherBusy = false;

   static async Task Main()
   {
      Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss} MQTTNet client starting");

      try
      {
         // load the app settings into configuration
         var configuration = new ConfigurationBuilder()
               .AddJsonFile("appsettings.json", false, true)
               .AddUserSecrets<Program>()
         .Build();

         _applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>();

         var mqttFactory = new MqttFactory();

         using (_client = mqttFactory.CreateMqttClient())
         {
            // Certificate based authentication
            List<X509Certificate2> certificates = new List<X509Certificate2>
            {
               new X509Certificate2(_applicationSettings.ClientCertificateFileName, _applicationSettings.ClientCertificatePassword)
            };

            var tlsOptions = new MqttClientTlsOptionsBuilder()
                  .WithClientCertificates(certificates)
                  .WithSslProtocols(System.Security.Authentication.SslProtocols.Tls12)
                  .UseTls(true)
                  .Build();

            MqttClientOptions mqttClientOptions = new MqttClientOptionsBuilder()
                     .WithClientId(_applicationSettings.ClientId)
                     .WithTcpServer(_applicationSettings.Host, _applicationSettings.Port)
                     .WithCredentials(_applicationSettings.UserName, _applicationSettings.Password)
                     .WithCleanStart(_applicationSettings.CleanStart)
                     .WithTlsOptions(tlsOptions)
                     .Build();

            var connectResult = await _client.ConnectAsync(mqttClientOptions);
            if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
            {
               throw new Exception($"Failed to connect: {connectResult.ReasonString}");
            }

            _client.ApplicationMessageReceivedAsync += OnApplicationMessageReceivedAsync;

            Console.WriteLine($"Subscribed to Topic");
            foreach (string topic in _applicationSettings.SubscribeTopics.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries))
            {
               var subscribeResult = await _client.SubscribeAsync(topic, _applicationSettings.SubscribeQualityOfService);

               Console.WriteLine($" {topic} Result:{subscribeResult.Items.First().ResultCode}");
            }
      }
//...
}
MQTTnet client console application output

The design of the MQTT protocol means that the hivemq-mqtt-client-dotnet and MQTTnet implementations are similar. Having used both I personally prefer the HiveMQ client library.

Azure Event Grid MQTT-Certificates

When configuring Root, Intermediate and Device certificates for my Azure Event Grid Devices using the smallstep step-ca or OpenSSL I made mistakes/typos which broke my deployment and in the end I was copying and pasting commands from Windows Notepad.

For one test deployment it took me an hour to generate the Root, Intermediate and a number of Devices certificates which was a waste of time. At this point I decided investigate writing some applications to simplify the process.

After some searching I stumbled across CREATING CERTIFICATES FOR X.509 SECURITY IN AZURE IOT HUB USING .NET CORE by Damien Bod which showed how to generate certificates for Azure IoT Hub solutions using his NuGet package Certificate Manager.

The AzureIoTHubDps repository had a sample showing how to generate the certificate chain for Azure IoT Hub devices. It worked really well but I accidentally overwrote my Root and Intermediate certificates, there were some “magic numbers” and hard coded passwords (it was a sample) so I decided to “chop” the sample up into three command line applications.

static void Main(string[] args)
{
   var serviceProvider = new ServiceCollection()
         .AddCertificateManager()
         .BuildServiceProvider();

   // load the app settings into configuration
   var configuration = new ConfigurationBuilder()
         .AddJsonFile("appsettings.json", false, true)
         .AddUserSecrets<Program>()
   .Build();

   _applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>();
//------
   Console.WriteLine($"validFrom:{validFrom} ValidTo:{validTo}");

   var serverRootCertificate = serviceProvider.GetService<CreateCertificatesClientServerAuth>();

   var root = serverRootCertificate.NewRootCertificate(
         new DistinguishedName { 
               CommonName = _applicationSettings.CommonName,
               Organisation = _applicationSettings.Organisation,
               OrganisationUnit = _applicationSettings.OrganisationUnit,
               Locality = _applicationSettings.Locality,
               StateProvince  = _applicationSettings.StateProvince,
               Country = _applicationSettings.Country
         },
         new ValidityPeriod { 
         ValidFrom = validFrom,
         ValidTo = validTo,
         },
         _applicationSettings.PathLengthConstraint,
         _applicationSettings.DnsName);
   root.FriendlyName = _applicationSettings.FriendlyName;

   Console.Write("PFX Password:");
   string password = Console.ReadLine();
   if ( String.IsNullOrEmpty(password))
   {
      Console.WriteLine("PFX Password invalid");
      return;
   }

   var exportCertificate = serviceProvider.GetService<ImportExportCertificate>();

   var rootCertificatePfxBytes = exportCertificate.ExportRootPfx(password, root);
   File.WriteAllBytes(_applicationSettings.RootCertificateFilePath, rootCertificatePfxBytes);

   Console.WriteLine($"Root certificate file:{_applicationSettings.RootCertificateFilePath}");
   Console.WriteLine("press enter to exit");
   Console.ReadLine();
}

The application’s configuration was split between application settings file(certificate file paths, validity periods, Organisation etc.) or entered at runtime ( certificate filenames, passwords etc.) The first application generates a Root Certificate using the distinguished name information from the application settings, plus file names and passwords entered by the user.

Root Certificate generation application output

The second application generates an Intermediate Certificate using the Root Certificate, the distinguished name information from the application settings, plus file names and passwords entered by the user.

static void Main(string[] args)
{
   var serviceProvider = new ServiceCollection()
         .AddCertificateManager()
         .BuildServiceProvider();

   // load the app settings into configuration
   var configuration = new ConfigurationBuilder()
         .AddJsonFile("appsettings.json", false, true)
         .AddUserSecrets<Program>()
   .Build();

   _applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>();
//------
   Console.WriteLine($"validFrom:{validFrom} be after ValidTo:{validTo}");

   Console.WriteLine($"Root Certificate file:{_applicationSettings.RootCertificateFilePath}");

   Console.Write("Root Certificate Password:");
   string rootPassword = Console.ReadLine();
   if (String.IsNullOrEmpty(rootPassword))
   {
      Console.WriteLine("Fail");
      return;
   }
   var rootCertificate = new X509Certificate2(_applicationSettings.RootCertificateFilePath, rootPassword);

   var intermediateCertificateCreate = serviceProvider.GetService<CreateCertificatesClientServerAuth>();

   var intermediateCertificate = intermediateCertificateCreate.NewIntermediateChainedCertificate(
         new DistinguishedName
         {
            CommonName = _applicationSettings.CommonName,
            Organisation = _applicationSettings.Organisation,
            OrganisationUnit = _applicationSettings.OrganisationUnit,
            Locality = _applicationSettings.Locality,
            StateProvince = _applicationSettings.StateProvince,
            Country = _applicationSettings.Country
         },
      new ValidityPeriod
      {
         ValidFrom = validFrom,
         ValidTo = validTo,
      },
            _applicationSettings.PathLengthConstraint,
            _applicationSettings.DnsName, rootCertificate);
      intermediateCertificate.FriendlyName = _applicationSettings.FriendlyName;

   Console.Write("Intermediate certificate Password:");
   string intermediatePassword = Console.ReadLine();
   if (String.IsNullOrEmpty(intermediatePassword))
   {
      Console.WriteLine("Fail");
      return;
   }

   var importExportCertificate = serviceProvider.GetService<ImportExportCertificate>();

   Console.WriteLine($"Intermediate PFX file:{_applicationSettings.IntermediateCertificatePfxFilePath}");
   var intermediateCertificatePfxBtyes = importExportCertificate.ExportChainedCertificatePfx(intermediatePassword, intermediateCertificate, rootCertificate);
   File.WriteAllBytes(_applicationSettings.IntermediateCertificatePfxFilePath, intermediateCertificatePfxBtyes);

   Console.WriteLine($"Intermediate CER file:{_applicationSettings.IntermediateCertificateCerFilePath}");
   var intermediateCertificatePemText = importExportCertificate.PemExportPublicKeyCertificate(intermediateCertificate);
   File.WriteAllText(_applicationSettings.IntermediateCertificateCerFilePath, intermediateCertificatePemText);

   Console.WriteLine("press enter to exit");
   Console.ReadLine();
}
Intermediate Certificate generation application output
Uploading the Intermediate certificate to Azure Event Grid

The third application generates Device Certificates using the Intermediate Certificate, distinguished name information from the application settings, plus device id, file names and passwords entered by the user.

static void Main(string[] args)
{
   var serviceProvider = new ServiceCollection()
         .AddCertificateManager()
         .BuildServiceProvider();

   // load the app settings into configuration
   var configuration = new ConfigurationBuilder()
         .AddJsonFile("appsettings.json", false, true)
         .AddUserSecrets<Program>()
   .Build();

   _applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>();
//------
   Console.WriteLine($"validFrom:{validFrom} ValidTo:{validTo}");

   Console.WriteLine($"Intermediate PFX file:{_applicationSettings.IntermediateCertificateFilePath}");

   Console.Write("Intermediate PFX Password:");
   string intermediatePassword = Console.ReadLine();
   if (String.IsNullOrEmpty(intermediatePassword))
   {
      Console.WriteLine("Intermediate PFX Password invalid");
      return;
   }
   var intermediate = new X509Certificate2(_applicationSettings.IntermediateCertificateFilePath, intermediatePassword);

   Console.Write("Device ID:");
   string deviceId = Console.ReadLine();
   if (String.IsNullOrEmpty(deviceId))
   {
      Console.WriteLine("Device ID invalid");
      return;
   }

   var createClientServerAuthCerts = serviceProvider.GetService<CreateCertificatesClientServerAuth>();

   var device = createClientServerAuthCerts.NewDeviceChainedCertificate(
         new DistinguishedName
         {
            CommonName = deviceId,
            Organisation = _applicationSettings.Organisation,
            OrganisationUnit = _applicationSettings.OrganisationUnit,
            Locality = _applicationSettings.Locality,
            StateProvince = _applicationSettings.StateProvince,
            Country = _applicationSettings.Country
         },
      new ValidityPeriod
      {
         ValidFrom = validFrom,
         ValidTo = validTo,
      },
      deviceId, intermediate);
   device.FriendlyName = deviceId;

   Console.Write("Device PFX Password:");
   string devicePassword = Console.ReadLine();
   if (String.IsNullOrEmpty(devicePassword))
   {
      Console.WriteLine("Fail");
      return;
   }

   var importExportCertificate = serviceProvider.GetService<ImportExportCertificate>();

   string devicePfxPath = string.Format(_applicationSettings.DeviceCertificatePfxFilePath, deviceId);

   Console.WriteLine($"Device PFX file:{devicePfxPath}");
   var deviceCertificatePath = importExportCertificate.ExportChainedCertificatePfx(devicePassword, device, intermediate);
   File.WriteAllBytes(devicePfxPath,  deviceCertificatePath);

   Console.WriteLine("press enter to exit");
   Console.ReadLine();
}
Device Certificate generation application output
Uploading the Intermediate certificate to Azure Event Grid

These applications wouldn’t have been possible without Damien Bod’s CREATING CERTIFICATES FOR X.509 SECURITY IN AZURE IOT HUB USING .NET CORE blog post, and his Certificate Manager NuGet package.

Airbnb Dataset – JSON Long Integer Issue

The Inside Airbnb London dataset has 87946 listings and the id column (which is the primary key) has a minimum value of 13913 and maximum of 973895808066047620 in the database.

Back in the early 90’s I used to live next to the Ealing Lawn Tennis Club in London

I used “Ealing” as the SearchText for my initial testing and tried different page numbers and sizes

Testing the search functionality with SwaggerUI

The listings search results JSON looked good but I missed one important detail…

[
  {
    "id": 32458423,
    "name": "Bed and breakfast in Ealing  · ★5.0 · 1 bedroom · 1 bed · 1 private bath",
    "listingURL": "https://www.airbnb.com/rooms/32458423"
  },
  {
    "id": 7905935,
    "name": "Guest suite in Ealing Broadway · ★4.93 · 1 bedroom · 1 bed · 1 private bath",
    "listingURL": "https://www.airbnb.com/rooms/7905935"
  },
  {
    "id": 5262733,
    "name": "Home in Ealing · ★4.97 · 1 bedroom · 1 shared bath",
    "listingURL": "https://www.airbnb.com/rooms/5262733"
  },
  {
    "id": 685148830257321200,
    "name": "Home in Ealing London · ★5.0 · 4 bedrooms · 8 beds · 2.5 baths",
    "listingURL": "https://www.airbnb.com/rooms/685148830257321258"
  },
  {
    "id": 10704599,
    "name": "Home in ealing, london · ★4.47 · 3 bedrooms · 4 beds · 1 bath",
    "listingURL": "https://www.airbnb.com/rooms/10704599"
  }
]

To “smoke test” to the lookup functionality I tried a couple of the listing ids

Swagger user interface successful lookup listing 7905935
Swagger user interface unsuccessful lookup of listing 685148830257321200

The HTTP GET method routing parameter and the response Data Transfer Object(DTO) the Airbnb listing Id properties are long values (ulong might have been a better choice) which should have sufficient range

string LookupByIdSql = @"SELECT Id, [Name], Listing_URL AS ListingURL
                     FROM ListingsHosts
                     WHERE id = @Id";

public record ListingLookupDto
{
   public long Id { get; set; }
   public string? Name { get; set; }
   public string? ListingURL { get; set; }
};

//...

app.MapGet("/Listing/Results/{id:long}", async (long id, IDapperContext dappperContext) =>
{
   using (var connection = dappperContext.ConnectionCreate())
   {
      ListingLookupDto result = await connection.QuerySingleOrDefaultWithRetryAsync<ListingLookupDto>(LookupByIdSql, new { id });
      if (result is null)
      {
         return Results.Problem($"Listing {id} not found", statusCode: StatusCodes.Status404NotFound);
      }

      return Results.Ok(result);
   }
})
.Produces<ListingLookupDto>(StatusCodes.Status200OK)
.Produces<ProblemDetails>(StatusCodes.Status404NotFound)
.WithOpenApi();

The id values in the search response and lookup DTOs were correct

Visual Studio 2022 Debugger inspecting listing id value

I had missed the clue in the search response JSON the listing id and the listingURL id didn’t match.

{
 "id": 685148830257321200,
 "name": "Home in Ealing London · ★5.0 · 4 bedrooms · 8 beds · 2.5 baths",
 "listingURL": "https://www.airbnb.com/rooms/685148830257321258"
},

The JavaScript Object Notation (JSON) Data Interchange Format(RFC7159) specification for numbers explains the discrepancy.

This specification allows implementations to set limits on the range
and precision of numbers accepted. Since software that implements
IEEE 754-2008 binary64 (double precision) numbers [IEEE754] is
generally available and widely used, good interoperability can be
achieved by implementations that expect no more precision or range
than these provide, in the sense that implementations will
approximate JSON numbers within the expected precision.
Airbnb listing from listingURL

Airbnb Dataset – Querying the Raw Listings

My initial ASP.NET Core Minimal AP exploration uses the Inside Airbnb London dataset which has 87946 listings. The data is pretty “nasty” with lots of nullable and wide columns so it took several attempts to import.

CREATE TABLE [dbo].[listingsRaw](
	[id] [bigint] NOT NULL,
	[listing_url] [nvarchar](50) NOT NULL,
	[scrape_id] [datetime2](7) NOT NULL,
	[last_scraped] [date] NOT NULL,
	[source] [nvarchar](50) NOT NULL,
	[name] [nvarchar](max) NOT NULL,
	[description] [nvarchar](max) NULL,
	[neighborhood_overview] [nvarchar](1050) NULL,
	[picture_url] [nvarchar](150) NULL,
	[host_id] [int] NOT NULL,
	[host_url] [nvarchar](50) NOT NULL,
	[host_name] [nvarchar](50) NULL,
	[host_since] [date] NULL,
	[host_location] [nvarchar](100) NULL,
	[host_about] [nvarchar](max) NULL,
	[host_response_time] [nvarchar](50) NULL,
	[host_response_rate] [nvarchar](50) NULL,
	[host_acceptance_rate] [nvarchar](50) NULL,
	[host_is_superhost] [bit] NULL,
	[host_thumbnail_url] [nvarchar](150) NULL,
	[host_picture_url] [nvarchar](150) NULL,
	[host_neighbourhood] [nvarchar](50) NULL,
	[host_listings_count] [int] NULL,
	[host_total_listings_count] [int] NULL,
	[host_verifications] [nvarchar](50) NOT NULL,
	[host_has_profile_pic] [bit] NULL,
	[host_identity_verified] [bit] NULL,
	[neighbourhood] [nvarchar](100) NULL,
	[neighbourhood_cleansed] [nvarchar](50) NOT NULL,
	[neighbourhood_group_cleansed] [nvarchar](1) NULL,
	[latitude] [float] NOT NULL,
	[longitude] [float] NOT NULL,
	[property_type] [nvarchar](50) NOT NULL,
	[room_type] [nvarchar](50) NOT NULL,
	[accommodates] [tinyint] NOT NULL,
	[bathrooms] [nvarchar](1) NULL,
	[bathrooms_text] [nvarchar](50) NULL,
	[bedrooms] [tinyint] NULL,
	[beds] [tinyint] NULL,
	[amenities] [nvarchar](max) NOT NULL,
	[price] [money] NOT NULL,
	[minimum_nights] [smallint] NOT NULL,
	[maximum_nights] [int] NOT NULL,
	[minimum_minimum_nights] [smallint] NULL,
	[maximum_minimum_nights] [int] NULL,
	[minimum_maximum_nights] [int] NULL,
	[maximum_maximum_nights] [int] NULL,
	[minimum_nights_avg_ntm] [float] NULL,
	[maximum_nights_avg_ntm] [float] NULL,
	[calendar_updated] [nvarchar](1) NULL,
	[has_availability] [bit] NOT NULL,
	[availability_30] [tinyint] NOT NULL,
	[availability_60] [tinyint] NOT NULL,
	[availability_90] [tinyint] NOT NULL,
	[availability_365] [smallint] NOT NULL,
	[calendar_last_scraped] [date] NOT NULL,
	[number_of_reviews] [smallint] NOT NULL,
	[number_of_reviews_ltm] [int] NOT NULL,
	[number_of_reviews_l30d] [tinyint] NOT NULL,
	[first_review] [date] NULL,
	[last_review] [date] NULL,
	[review_scores_rating] [float] NULL,
	[review_scores_accuracy] [float] NULL,
	[review_scores_cleanliness] [float] NULL,
	[review_scores_checkin] [float] NULL,
	[review_scores_communication] [float] NULL,
	[review_scores_location] [float] NULL,
	[review_scores_value] [float] NULL,
	[license] [nvarchar](max) NULL,
	[instant_bookable] [bit] NOT NULL,
	[calculated_host_listings_count] [int] NULL,
	[calculated_host_listings_count_entire_homes] [int] NOT NULL,
	[calculated_host_listings_count_private_rooms] [int] NOT NULL,
	[calculated_host_listings_count_shared_rooms] [int] NOT NULL,
	[reviews_per_month] [float] NULL
) ON [PRIMARY] TEXTIMAGE_ON [PRIMARY]

There are other data quality issues e.g. the host information is duplicated in each of their Listings e.g. host_id, host_name, host_since, host_* etc. which will need to be tidied up.

Swagger user interface for Raw Listings search functionality.

I have implemented basic (“incomplete”) OpenAPI support for functionality and stress testing.

Swagger user interface parameterised search functionality.

The search results are paginated and individual listings can be retrieved using the Airbnb listing “id”.

const string SearchPaginatedSql = @"SELECT Uid,Id,[Name], neighbourhood
                     FROM listings
                     WHERE[Name] LIKE N'%' + @SearchText + N'%'
                     ORDER By[Name] 
                     OFFSET @PageSize *(@PageNumber - 1) ROWS FETCH NEXT @PageSize ROWS ONLY";

public record ListingListDto
{
   public long Id { get; set; }
   public string? Name { get; set; }
   public string? Neighbourhood { get; set; }
};
Swagger user interface search functionality with untyped response.

The first HTTP GET implementation returns an untyped result-set which was not very helpful.

app.MapGet("/Listing/Search", async (string searchText, int pageNumber, int pageSize, [FromServices] IDapperContext dappperContext) =>
{
   using (var connection = dappperContext.ConnectionCreate())
   {
      return await connection.QueryWithRetryAsync(SearchPaginatedSql, new { searchText, pageNumber, pageSize });
   }
})
.WithOpenApi();
Swagger user interface search functionality with typed response

The second HTTP GET implementation returns a typed result-set which improved the “usability” of clients generated from the OpenAPI definition file.

app.MapGet("/Listing/Search/Typed", async (string searchText, int pageNumber, int pageSize, [FromServices] IDapperContext dappperContext) =>
{
   using (var connection = dappperContext.ConnectionCreate())
   {
      return await connection.QueryWithRetryAsync<ListingListDto>(SearchPaginatedSql, new { searchText, pageNumber, pageSize });
   }
})
.Produces<IList<ListingListDto>>(StatusCodes.Status200OK)
.WithOpenApi();

The third HTTP GET method uses the Listing id specified in a routing parameter to lookup a Listing

string LookupByIdSql = @"SELECT Id,[Name], neighbourhood
FROM ListingsHosts
WHERE id = @Id";

public record ListingLookupDto
{
public long Id { get; set; }
public string? Name { get; set; }
public string? Neighbourhood { get; set; }
};
Swagger user interface Listing lookup functionality
app.MapGet("/Listing/{id:long}", async (long id, IDapperContext dappperContext) =>
{
   using (var connection = dappperContext.ConnectionCreate())
   {
      ListingLookupDto result = await connection.QuerySingleOrDefaultWithRetryAsync<ListingLookupDto>(LookupByIdSql, new { id });
      if (result is null)
      {
         return Results.Problem($"Listing {id} not found", statusCode: StatusCodes.Status404NotFound);
      }

      return Results.Ok(result);
   }
})
.Produces<ListingLookupDto>(StatusCodes.Status200OK)
.Produces<ProblemDetails>(StatusCodes.Status404NotFound)
.WithOpenApi();

The lack of validation of the SearchText, PageSize and PageNumber parameters allow uses to enter invalid values which caused searches to fail.

Swagger user interface search functionality with an invalid page number

My initial approach was to decorate the parameters of the ValidatedQuery method with DataAnnotations to ensure only valid values were accepted.

const byte SearchTextMinimumLength = 3;
const byte SearchTextMaximumLength = 20;
const byte PageNumberMinimum = 1;
const byte PageNumberMaximum = 100;
const byte PageSizeMinimum = 5;
const byte PageSizeMaximum = 50;

app.MapGet("/Listing/Search/ValidatedQuery", async (
   [FromQuery,Required, MinLength(SearchTextMinimumLength, ErrorMessage = "SearchTextMaximumLength"), MaxLength(SearchTextMaximumLength, ErrorMessage = "SearchTextMaximumLegth")]
   string searchText,
   [FromQuery, Range(PageNumberMinimum, PageNumberMaximum, ErrorMessage = "PageNumberMinimum PageNumberMaximum")]
   int pageNumber,
   [FromQuery, Range(PageSizeMinimum, PageSizeMaximum, ErrorMessage = "PageSizeMinimum PageSizeMaximum")]
   int pageSize,
   [FromServices] IDapperContext dappperContext) =>
{
   using (var connection = dappperContext.ConnectionCreate())
   {
      return await connection.QueryWithRetryAsync<ListingListDto>(SearchPaginatedSql, new { searchText, pageNumber, pageSize });
   }
})
.Produces<IList<ListingListDto>>(StatusCodes.Status200OK)
.Produces<ProblemDetails>(StatusCodes.Status400BadRequest)
.WithOpenApi();

This wasn’t a great solution because the validation of the parameters was declared as part of the user interface and would have to be repeated everywhere listing search functionality was provided.

Swagger user interface search functionality with parameter validation

The final HTTP GET method uses DataAnnotations on the SearchParameter(DTO) and AsParameters to bind the query string values.

app.MapGet("/Listing/Search/Parameters", async ([AsParameters] SearchParameters searchParameters,
   [FromServices] IDapperContext dappperContext) =>
{
   using (var connection = dappperContext.ConnectionCreate())
   {
      return await connection.QueryWithRetryAsync<ListingListDto>(SearchPaginatedSql, new { searchText = searchParameters.SearchText, searchParameters.PageNumber, searchParameters.PageSize });
   }
})
.Produces<IList<ListingListDto>>(StatusCodes.Status200OK)
.Produces<ProblemDetails>(StatusCodes.Status400BadRequest)
.WithOpenApi();

public record SearchParameters
{
   // https://github.com/domaindrivendev/Swashbuckle.AspNetCore/issues/2658 possibly related?
   public const byte SearchTextMinimumLength = 3;
   public const byte SearchTextMaximumLength = 15;
   public const int PageNumberMinimum = 1;
   public const int PageNumberMaximum = 100;
   public const byte PageSizeMinimum = 5;
   public const byte PageSizeMaximum = 50;

   //[FromQuery, Required, MinLength(SearchTextMinimumLength, ErrorMessage = "SearchTextMinimumLegth"), MaxLength(SearchTextMaximumLength, ErrorMessage = "SearchTextMaximumLegth")]
   //[Required, MinLength(SearchTextMinimumLength, ErrorMessage = "SearchTextMinimumLegth"), MaxLength(SearchTextMaximumLength, ErrorMessage = "SearchTextMaximumLegth")]
   [MinLength(SearchTextMinimumLength, ErrorMessage = "SearchTextMinimumLegth"), MaxLength(SearchTextMaximumLength, ErrorMessage = "SearchTextMaximumLegth")]
   public string SearchText { get; set; }

   //[FromQuery, Range(PageNumberMinimum, PageNumberMaximum, ErrorMessage = "PageNumberMinimum PageNumberMaximum")]
   //[Required, Range(PageNumberMinimum, PageNumberMaximum, ErrorMessage = "PageNumberMinimum PageNumberMaximum")]
   [Range(PageNumberMinimum, PageNumberMaximum, ErrorMessage = "PageNumberMinimum PageNumberMaximum")]
   public int PageNumber { get; set; }

   [Range(PageSizeMinimum, PageSizeMaximum, ErrorMessage = "PageSizeMinimum PageSizeMaximum")]
   public int PageSize { get; set; }
}
Swagger user interface search functionality with parameter validation

This last two implementations worked though the error messages I had embedded in the code were not displayed I think this is related to this Swashbuckle Issue.

There is also an issue looking up some listings with larger listing ids which I will need some investigation.

Myriota Connector – Azure IoT Hub Downlink Methods

The Azure IoT Hub Cloud to Device(C2D) messaging approach didn’t work at all well with the Myriota Application Programming Interface(API). A downlink message can take up to 24hrs to be delivered and the Myriota (API) doesn’t currently queue control messages. Also, the Myriota (API) control message send method responds with 400 Bad Request if there is already a message being sent to a device. Azure IoT Hubs also support Direct Methods which provide immediate confirmation of the result of a request.

The Method Callback Delegate has different parameters, so I had to update the downlink formatter interface and update all of the sample downlink payload formatters.

public interface IFormatterDownlink
{
   public byte[] Evaluate(string terminalId, string methodName, JObject? payloadJson, byte[] payloadBytes);
}

How direct methods will be processed is configured in the application settings. For each direct method name the downlink payload formatter to be invoked and an optional Javascript Object Notation(JSON) payload can be configured.

"IoTHub": {
   ...
   "Methods": {
      "LightsGoOn": {
         "Formatter": "LightsOffOn.cs",
         "Payload": "{\"Light\": true}"
      },
      "LightsGoOff": {
         "Formatter": "LightsOffOn.cs",
         "Payload": "{\"Light\": false}"
      },
      "FanSpeed": {
         "Formatter": "FanSpeed.cs",
         "Payload": ""
      },
...
}

If there is no configuration for the direct method name, the payload formatter specified in Myriota device “DownlinkDefault” Attribute is used, and if that is not configured the default formatter in the payloadFormatters section of the application settings is used.

namespace devMobile.IoT.MyriotaAzureIoTConnector.Connector
{
   internal class IoTHubDownlink(ILogger<IoTHubDownlink> _logger, IOptions<Models.AzureIoT> azureIoTSettings, IPayloadFormatterCache _payloadFormatterCache, IMyriotaModuleAPI _myriotaModuleAPI) : IIoTHubDownlink
   {
      private readonly Models.AzureIoT _azureIoTSettings = azureIoTSettings.Value;

      public async Task<MethodResponse> IotHubMethodHandler(MethodRequest methodRequest, object userContext)
      {
         // DIY request identifier so processing progress can be tracked in Application Insights
         string requestId = Guid.NewGuid().ToString();

         Models.DeviceConnectionContext context = (Models.DeviceConnectionContext)userContext;

         try
         {
            _logger.LogInformation("Downlink- TerminalId:{TerminalId} RequestId:{requestId} Name:{Name}", context.TerminalId, requestId, methodRequest.Name);

            // Lookup payload formatter name, none specified use context one which is from device attributes or the default in configuration
            string payloadFormatterName;
            if (_azureIoTSettings.IoTHub.Methods.TryGetValue(methodRequest.Name, out Models.AzureIoTHubMethod? method) && !string.IsNullOrEmpty(method.Formatter))
            {
               payloadFormatterName = method.Formatter;

               _logger.LogInformation("Downlink- IoT Hub TerminalID:{TermimalId} RequestID:{requestId} Method formatter:{payloadFormatterName} ", context.TerminalId, requestId, payloadFormatterName);
            }
            else
            {
               payloadFormatterName = context.PayloadFormatterDownlink;

               _logger.LogInformation("Downlink- IoT Hub TerminalID:{TermimalId} RequestID:{requestId} Context formatter:{payloadFormatterName} ", context.TerminalId, requestId, payloadFormatterName);
            }

            // Display methodRequest.Data as Hex
            if (methodRequest.Data is not null)
            {
               _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Data:{Data}", context.TerminalId, requestId, BitConverter.ToString(methodRequest.Data));
            }
            else
            {
               _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Data:null", context.TerminalId, requestId);
            }

            JObject? requestJson = null;

            if ((method is not null) && !string.IsNullOrWhiteSpace(method.Payload))
            {
               // There is a matching method with a possible JSON payload
               string payload = method.Payload.Trim();

               if ((payload.StartsWith('{') && payload.EndsWith('}')) || (payload.StartsWith('[') && payload.EndsWith(']')))
               {
                  // The payload is could be JSON
                  try
                  {
                     requestJson = JObject.Parse(payload);
                  }
                  catch (JsonReaderException jex)
                  {
                     _logger.LogWarning(jex, "Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Method Payload is not valid JSON", context.TerminalId, requestId);

                     return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"RequestID:{requestId} Method payload is not valid JSON.\"}}"), (int)HttpStatusCode.UnprocessableEntity);
                  }
               }
               else
               {
                  // The payload couldn't be JSON
                  _logger.LogWarning("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Method Payload is definitely not valid JSON", context.TerminalId, requestId);

                  return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"RequestID:{requestId} Method payload is definitely not valid JSON.\"}}"), (int)HttpStatusCode.UnprocessableEntity);
               }

               _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Method Payload:{requestJson}", context.TerminalId, requestId, JsonConvert.SerializeObject(requestJson, Formatting.Indented));
            }
            else
            {
               // If there was not matching method or the payload was "empty" see if the method request payload is valid
               if (!string.IsNullOrWhiteSpace(methodRequest.DataAsJson))
               {
                  string payload = methodRequest.DataAsJson.Trim();

                  if ((payload.StartsWith('{') && payload.EndsWith('}')) || (payload.StartsWith('[') && payload.EndsWith(']')))
                  {
                     // The payload is could be JSON
                     try
                     {
                        requestJson = JObject.Parse(payload);

                        _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} DataAsJson:{requestJson}", context.TerminalId, requestId, JsonConvert.SerializeObject(requestJson, Formatting.Indented));
                     }
                     catch (JsonReaderException jex)
                     {
                        _logger.LogInformation(jex, "Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} DataAsJson is not valid JSON", context.TerminalId, requestId);
                     }
                  }
               }
            }

            // This "shouldn't" fail, but it could for invalid path to blob, timeout retrieving blob, payload formatter syntax error etc.
            IFormatterDownlink payloadFormatter = await _payloadFormatterCache.DownlinkGetAsync(payloadFormatterName);

            if ( requestJson is null ) 
            { 
               requestJson = new JObject();
            }

            // This also "shouldn't" fail, but the payload formatters can throw runtime exceptions like null reference, divide by zero, index out of range etc.
            byte[] payloadBytes = payloadFormatter.Evaluate(context.TerminalId, methodRequest.Name, requestJson, methodRequest.Data);

            // Validate payload before calling Myriota control message send API method
            if (payloadBytes is null)
            {
               _logger.LogWarning("Downlink- IoT Hub TerminalID:{TerminalId} Request:{requestId} Evaluate returned null", context.TerminalId, requestId);

               return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"RequestID:{requestId} payload evaluate returned null.\"}}"), (int)HttpStatusCode.UnprocessableEntity);
            }

            if ((payloadBytes.Length < Constants.DownlinkPayloadMinimumLength) || (payloadBytes.Length > Constants.DownlinkPayloadMaximumLength))
            {
               _logger.LogWarning("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} PayloadBytes:{payloadBytes} length:{Length} invalid, must be {DownlinkPayloadMinimumLength} to {DownlinkPayloadMaximumLength} bytes", context.TerminalId, requestId, BitConverter.ToString(payloadBytes), payloadBytes.Length, Constants.DownlinkPayloadMinimumLength, Constants.DownlinkPayloadMaximumLength); ;

               return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"RequestID:{requestId} payload evaluation length invalid.\"}}"), (int)HttpStatusCode.UnprocessableEntity);
            }

            // Finally send Control Message to device using the Myriota API
            _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestID} PayloadBytes:{payloadBytes} Length:{Length} sending", context.TerminalId, requestId, BitConverter.ToString(payloadBytes), payloadBytes.Length);

            string messageId = await _myriotaModuleAPI.SendAsync(context.TerminalId, payloadBytes);

            _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Myriota MessageID:{messageId} sent", context.TerminalId, requestId, messageId);
         }
         catch (Exception ex)
         {
            _logger.LogError(ex, "Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} IotHubMethodHandler processing failed", context.TerminalId, requestId);

            return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"TerminalID:{context.TerminalId} RequestID:{requestId} method handler failed.\"}}"), (int)HttpStatusCode.InternalServerError);
         }

         return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"TerminalID:{context.TerminalId} RequestID:{requestId} Message sent successfully.\"}}"), (int)HttpStatusCode.OK);
      }
   }
}

The validation of Direct Method MessageRequest DataAsJson and the method configuration payload is a three-step process, first any leading or trailing whitespace is removed, then the first and last characters are checked as a JSON payload has to enclosed in {}(an object) or [] (an array) characters, then finally JObject.Parse is used to populate a JObject.

If the method configuration payload is “broken” an HTTP 422 Unprocessable content is returned. If the MessageRequest DataAsJson is “broken” only the Direct Method MessageRequest Payload passed to the evaluator.

Azure IoT Explorer Invoking FanSpeed method with correct JSON payload

I used Azure IoT Explorer to invoke C2D methods with optional “hand-crafted” JavaScript Object Notation(JSON) payloads.

public class FormatterDownlink : PayloadFormatter.IFormatterDownlink
{
   public byte[] Evaluate(string terminalId, string methodName, JObject payloadJson, byte[] payloadBytes)
   {
      byte? status = payloadJson.GetValue("FanSpeed", StringComparison.OrdinalIgnoreCase)?.Value<byte>();

      if (!status.HasValue)
      {
         return new byte[] { };
      }

      return new byte[] { 1, status.Value };
   }
}

The FanSpeed.cs payload formatter extracts the FanSpeed value from the JSON payload and returns a two byte array containing the message type and speed of the fan.

Azure Function application displaying Diagnostic information for control message

Each logging message starts with the TerminalID (to simplify searching for all the direct methods invoked on a device) and the requestId a Globally Unique Identifier (GUID) to simplify searching for all the “steps” associated with sending a message) with the rest of the logging message containing “step” specific diagnostic information.

Azure Application Insights displaying information diagnostic information
Myriota Device manager control message history displaying pending control message

The Azure IoT Explorer payload for an empty message contained two ” characters which is a bit odd. I will have to build a test application which uses the Azure IoT Hub C2D direct method API to see if this is a “feature”.

Myriota Connector – Azure IoT Hub Downlink logging refactor

After several refactorings the code stabilised and the Azure IoT Hub downlink message handler (configured with SetMethodDefaultHandlerAsync ) was ready for testing. I used Azure IoT Explorer to send some “hand-crafted” JavaScript Object Notation(JSON) Cloud to Device(C2D) messages.

Each logging message starts with the TerminalID (to simplify searching for all the messages sent to a device) and the message LockToken (to simplify searching for all the “steps” associated with sending a message) with the rest of the logging message containing “step” specific diagnostic information.

Successful Azure IoT Explorer C2D JSON Message

If there is no PayloadFormatter attribute the default in the PayloadFormatters section of the function configuration is used.

using System;
using System.Collections.Generic;

using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

public class FormatterDownlink : PayloadFormatter.IFormatterDownlink
{
   public byte[] Evaluate(IDictionary<string, string> properties, string terminalId, JObject payloadJson, byte[] payloadBytes)
   {
      byte? status = payloadJson.Value<byte?>("FanSpeed");

      if (!status.HasValue)
      {
         return new byte[] { };
      }

      return new byte[] { 1, status.Value };
   }
}

The FanSpeed.cs payload formatter extracts the FanSpeed value from the JSON payload and returns a two byte array containing the message type and speed of the fan.

Azure IoT Function running waiting for a C2D message

After re-reading the SetMethodHandlerAync documentation I refactored the code (back to the approach used a couple of branches ago) with the “using” wrapping the try/catch.

public async Task AzureIoTHubMessageHandler(Message message, object userContext)
{
   Models.DeviceConnectionContext context = (Models.DeviceConnectionContext)userContext;

   _logger.LogInformation("Downlink- IoT Hub TerminalId:{TermimalId} LockToken:{LockToken}", context.TerminalId, message.LockToken);

   using (message) // https://learn.microsoft.com/en-us/dotnet/api/microsoft.azure.devices.client.deviceclient.setreceivemessagehandlerasync?view=azure-dotnet
   {
      try
      {
         // Replace default formatter with message specific formatter if configured.
         if (!message.Properties.TryGetValue(Constants.IoTHubDownlinkPayloadFormatterProperty, out string? payloadFormatterName) || string.IsNullOrEmpty(payloadFormatterName))
         {
            _logger.LogInformation("Downlink- IoT Hub TerminalID:{TermimalId} LockToken:{LockToken} Context formatter:{payloadFormatterName} ", context.TerminalId, message.LockToken, payloadFormatterName);

            payloadFormatterName = context.PayloadFormatterDownlink;
         }
         else
         {
            _logger.LogInformation("Downlink- IoT Hub TerminalID:{TermimalId} LockToken:{LockToken} Property formatter:{payloadFormatterName} ", context.TerminalId, message.LockToken, payloadFormatterName);
         }


         // If GetBytes fails payload really badly broken
         byte[] messageBytes = message.GetBytes();

         _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} Message bytes:{messageBytes}", context.TerminalId, message.LockToken, BitConverter.ToString(messageBytes));


         // Try converting the bytes to text then to JSON
         JObject? messageJson = null;
         try
         {
            // These will fail for some messages, payload formatter gets bytes only
            string messageText = Encoding.UTF8.GetString(messageBytes);

            try
            {
               messageJson = JObject.Parse(messageText);

               _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} JSON:{messageJson}", context.TerminalId, message.LockToken, JsonConvert.SerializeObject(messageJson, Formatting.Indented));
            }
            catch (JsonReaderException jex)
            {
               _logger.LogInformation(jex, "Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} not valid JSON", context.TerminalId, message.LockToken);
            }
         }
         catch (ArgumentException aex)
         {
            _logger.LogInformation(aex, "Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} message bytes not valid text", context.TerminalId, message.LockToken);
         }


         // This shouldn't fail, but it could for invalid path to blob, timeout retrieving blob, payload formatter syntax error etc.
         IFormatterDownlink payloadFormatter = await _payloadFormatterCache.DownlinkGetAsync(payloadFormatterName);

         // This will fail if payload formatter throws runtime exceptions like null reference, divide by zero, index out of range etc.
         byte[] payloadBytes = payloadFormatter.Evaluate(message.Properties, context.TerminalId, messageJson, messageBytes);


         // Validate payload before calling Myriota control message send API method
         if (payloadBytes is null)
         {
            _logger.LogWarning("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} payload formatter:{payloadFormatter} Evaluate returned null", context.TerminalId, message.LockToken, payloadFormatterName);

            await context.DeviceClient.RejectAsync(message);

            return;
         }

         if ((payloadBytes.Length < Constants.DownlinkPayloadMinimumLength) || (payloadBytes.Length > Constants.DownlinkPayloadMaximumLength))
         {
            _logger.LogWarning("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} PayloadBytes:{payloadBytes} length:{Length} invalid must be {DownlinkPayloadMinimumLength} to {DownlinkPayloadMaximumLength} bytes", context.TerminalId, message.LockToken, Convert.ToHexString(payloadBytes), payloadBytes.Length, Constants.DownlinkPayloadMinimumLength, Constants.DownlinkPayloadMaximumLength);

            await context.DeviceClient.RejectAsync(message);

            return;
         }


         // Finally send Control Message to device using the Myriota API
         _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} PayloadBytes:{payloadBytes} Length:{Length} sending", context.TerminalId, message.LockToken, BitConverter.ToString(payloadBytes), payloadBytes.Length);

         string messageId = await _myriotaModuleAPI.SendAsync(context.TerminalId, payloadBytes);

         _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} MessageID:{messageId} sent", context.TerminalId, message.LockToken, messageId);

         await context.DeviceClient.CompleteAsync(message);
      }
      catch (Exception ex)
      {
         _logger.LogError(ex, "Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} MessageHandler processing failed", context.TerminalId, message.LockToken);

         await context.DeviceClient.RejectAsync(message);
      }
   }
}

The first time I ran the myriotaAzureIoTConnector Azure function in the Core Tools debugging environment there were no errors and the Microsoft.Azure.Devices.Client.DeviceClient connection cache loaded in the background.

Azure IoT Function failing with a SystemArgumentOutOfRangeException

The first time I sent a downlink message the handler failed spectacularly with a SystemArgumentOutOfRangeException

After adding some breakpoints, restarting the application, then single stepping through the code I found that I had accidentally used BitConverter.ToSingle(payloadBytes) instead of BitConverter.ToString(payloadBytes) to get the Hexadecimal representation of the payload bytes.

...
// Finally send Control Message to device using the Myriota API
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} PayloadBytes:{payloadBytes} Length:{Length} sending", context.TerminalId, message.LockToken, BitConverter.ToString(payloadBytes), payloadBytes.Length);

string messageId = await _myriotaModuleAPI.SendAsync(context.TerminalId, payloadBytes);
...
Azure IoT Function successfully sending downlink message.

The Encoding.UTF8.GetString and JObject.Parse are processed in a single Try with a specialised catch for when the payload cannot be converted to text. If the payload cannot be converted to JSON only the payloadBytes parameter of payload formatter is populated.

Myriota Connector – Azure IoT Hub Downlink final refactoring

I often print code and review it away from a computer. I can’t be distracted by “tinkering” with the code and I find that drawing on it helps me visualise what is going on. The payload formatters are retrieved from Azure Storage blob which have a default retry policy, the Azure IoT Hub DeviceClient methods have a default retry policy, the Myriota Cloud API SendMessage has retries (Implemented with Polly) and if the CS-Script compilation fails there is nothing that can be done so the code could be simplified.

The Azure IoT Hub downlink message handler was a partial class and part of implementation of the IDeviceConnectionCache which was a hangover from one of the initial versions.

internal partial class DeviceConnectionCache : IDeviceConnectionCache
{
   public async Task AzureIoTHubMessageHandler(Message message, object userContext)
   {
      Models.DeviceConnectionContext context = (Models.DeviceConnectionContext)userContext;

      _logger.LogInformation("Downlink- IoT Hub TerminalId:{termimalId} LockToken:{LockToken}", context.TerminalId, message.LockToken);

I replaced the IDeviceConnectionCache interface with IIoTHubDownlink which was declare in a new file in the interfaces folder.

namespace devMobile.IoT.MyriotaAzureIoTConnector.Connector
{
   public interface IIoTHubDownlink
   {
      public Task AzureIoTHubMessageHandler(Message message, object userContext);
   }
}

Then had to inject all the required dependencies which had been implemented in one of the other partial class files.

internal class IoTHubDownlink : IIoTHubDownlink
{
   private readonly ILogger<IoTHubDownlink> _logger;
   private readonly IPayloadFormatterCache _payloadFormatterCache;
   private readonly IMyriotaModuleAPI _myriotaModuleAPI;

   public IoTHubDownlink(ILogger<IoTHubDownlink> logger, IPayloadFormatterCache payloadFormatterCache, IMyriotaModuleAPI myriotaModuleAPI)
   {
      _logger = logger;
      _payloadFormatterCache = payloadFormatterCache;
      _myriotaModuleAPI = myriotaModuleAPI;
   }
...
}

The implementation had been extracted to a separate class so it had to be constructed by the Dependency Injection plumbing.

...
services.AddSingleton<IPayloadFormatterCache, PayloadFormatterCache>();
services.AddSingleton<IIoTHubDownlink, IoTHubDownlink>();
services.AddSingleton<IIoTCentralDownlink, IoTCentralDownlink>();
services.AddOptions<Models.MyriotaSettings>().Configure<IConfiguration>((settings, configuration) =>
{
    configuration.GetSection("Myriota").Bind(settings);
 });
 services.AddSingleton<IMyriotaModuleAPI, MyriotaModuleAPI>();
...

The lifetime of the Microsoft.Azure.Devices.Client.Message was being managed manually which seemed a bit odd.

public async Task AzureIoTHubMessageHandler(Message message, object userContext)
{
   Models.DeviceConnectionContext context = (Models.DeviceConnectionContext)userContext;

   _logger.LogInformation("Downlink- IoT Hub TerminalId:{termimalId} LockToken:{LockToken}", context.TerminalId, message.LockToken);

   // Use default formatter and replace with message specific formatter if configured.
   string payloadFormatter;
   if (!message.Properties.TryGetValue(Constants.IoTHubDownlinkPayloadFormatterProperty, out payloadFormatter) || string.IsNullOrEmpty(payloadFormatter))
   {
      payloadFormatter = context.PayloadFormatterDownlink;
   }

   _logger.LogInformation("Downlink- IoT Hub TerminalID:{termimalId} LockToken:{LockToken} Payload formatter:{payloadFormatter} ", context.TerminalId, message.LockToken, payloadFormatter);

   try
   {
   ...
   }
   finally
   {
      // Mop up the non managed resources of message
      message.Dispose();
   }
}

I replaced this with a with a “using” which “automagically” manages the lifetime of any non-managed resources. I also added string Locktoken variable for DeviceClient.RejectAsync and DeviceClient.CompletedAsync so that the “using” could be inside the try/catch (there is scope to reduce the amount of code in the “using”)

public async Task AzureIoTHubMessageHandler(Message message, object userContext)
{
   Models.DeviceConnectionContext context = (Models.DeviceConnectionContext)userContext;

   _logger.LogInformation("Downlink- IoT Hub TerminalId:{TermimalId} LockToken:{lockToken}", context.TerminalId, message.LockToken);

   // broken out so using for message only has to be inside try
   string lockToken = message.LockToken; 

   try
   {
      using (message)
      {
...
      }
      catch (Exception ex)
      {
         _logger.LogError(ex, "Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{lockToken} MessageHandler processing failed", context.TerminalId, lockToken);

         await context.DeviceClient.RejectAsync(lockToken);
      }
   }
}

The handling of the Encoding.UTF8.GetString and JObject.Parse payload was broken. If the Encoding.UTF8.GetString threw an exception there was no point in calling the JObject.Parse

// If this fails payload broken
byte[] messageBytes = message.GetBytes();

// This will fail for some messages, payload formatter gets bytes only
string messageText = string.Empty;
try
{
   messageText = Encoding.UTF8.GetString(messageBytes);
}
catch (ArgumentException aex)
{
   _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} messageBytes:{2} not valid Text", context.TerminalId, message.LockToken, BitConverter.ToString(messageBytes));
}

// This will fail for some messages, payload formatter gets bytes only
JObject? messageJson = null;
try
{
   messageJson = JObject.Parse(messageText);
}
catch ( JsonReaderException jex)
{
   _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} messageText:{2} not valid json", context.TerminalId, message.LockToken, BitConverter.ToString(messageBytes));
}

The Encoding.UTF8.GetString and JObject.Parse are now processed in a single Try with a specialised catch handling.

// These will fail for some messages, then payload formatter gets bytes only
string messageText = string.Empty;
JObject? messageJson = null;
try
{
   messageText = Encoding.UTF8.GetString(messageBytes);

   messageJson = JObject.Parse(messageText);
 }
catch (ArgumentException aex)
{
   _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{lockToken} messageBytes:{messageBytes} not valid text exception:{Message}", context.TerminalId, lockToken, BitConverter.ToString(messageBytes), aex.Message);
}
catch (JsonReaderException jex)
{
   _logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{lockToken} messageText:{messageText} not valid json exception:{Message}", context.TerminalId, lockToken, messageText, jex.Message);
}

// This shouldn't fail, but it could for lots of different reasons, invalid path to blob, syntax error, interface broken etc.
IFormatterDownlink payloadFormatter = await _payloadFormatterCache.DownlinkGetAsync(payloadFormatterName);

This refactored code now looks an awful lot like the “sunny days” code checked in on the 3rd of November.