Abstract data services behind an interface and some code cleanup.

Author: barnstee

Services/ADXDataService.cs DataServices/ADXDataService.csServices/ADXDataService.cs renamed to DataServices/ADXDataService.cs

@@ -3,17 +3,17 @@
 using Kusto.Data;
 using Kusto.Data.Common;
 using Kusto.Data.Net.Client;
-using System.Collections.Concurrent;
 using System.Data;
 using System.Diagnostics;
+using UA_DataProcessor.Interfaces;
 
 namespace Opc.Ua.Data.Processor
 {
-    public class ADXDataService : IDisposable
+    public class ADXDataService : IDataService
     {
         private ICslQueryProvider _queryProvider = null;
 
-        public ADXDataService()
+        public void Connect()
         {
             // connect to ADX cluster
             string adxClusterName = Environment.GetEnvironmentVariable("ADX_HOST");
@@ -47,8 +47,11 @@ public void Dispose()
             }
         }
 
-        public void RunADXQuery(string query, ConcurrentDictionary<string, object> values, bool allowMultiRow = false)
+        public Dictionary<string, object> RunQuery(string query)
         {
+            bool allowMultiRow = false;
+            Dictionary<string, object> values = new();
+
             ClientRequestProperties clientRequestProperties = new ClientRequestProperties()
             {
                 ClientRequestId = Guid.NewGuid().ToString()
@@ -107,6 +110,8 @@ public void RunADXQuery(string query, ConcurrentDictionary<string, object> value
             {
                 Console.WriteLine("RunADXQuery: " + ex.Message);
             }
+
+            return values;
         }
     }
 }

Services/DynamicsDataService.cs DataServices/DynamicsDataService.csServices/DynamicsDataService.cs renamed to DataServices/DynamicsDataService.cs

@@ -4,10 +4,11 @@
 using System.Net;
 using System.Net.Http.Headers;
 using System.Text;
+using UA_DataProcessor.Interfaces;
 
 namespace Opc.Ua.Data.Processor
 {
-    public class DynamicsDataService : IDisposable
+    public class DynamicsDataService : IDataService
     {
         private HttpClient _client = null;
 
@@ -17,7 +18,7 @@ public class DynamicsDataService : IDisposable
         private string _tenantId = string.Empty;
         private string _environmentId = string.Empty;
 
-        public DynamicsDataService()
+        public void Connect()
         {
             _client = new();
 
@@ -83,8 +84,21 @@ private async Task Authorize()
             }
         }
 
-        public async Task<DynamicsQueryResponse> RunDynamicsQuery(DynamicsQuery query)
+        public Dictionary<string, object> RunQuery(string query)
         {
+            Dictionary<string, object> results = new();
+
+            string[] queryLines = query.Split(['\r', '\n'], StringSplitOptions.RemoveEmptyEntries);
+            DynamicsQuery queryRequest = new()
+            {
+                tracingDirection = queryLines.Length > 0 ? queryLines[0] : string.Empty,
+                company = queryLines.Length > 1 ? queryLines[1] : string.Empty,
+                itemNumber = queryLines.Length > 2 ? queryLines[2] : string.Empty,
+                batchNumber = queryLines.Length > 3 ? queryLines[3] : string.Empty,
+                serialNumber = queryLines.Length > 4 ? queryLines[4] : string.Empty,
+                shouldIncludeEvents = true
+            };
+
             if (!string.IsNullOrEmpty(_instanceEndpoint))
             {
                 try
@@ -109,7 +123,7 @@ public async Task<DynamicsQueryResponse> RunDynamicsQuery(DynamicsQuery query)
                         Debug.WriteLine("Bearer Token expired! Attempting to retrieve a new barer token.");
 
                         // re-authorize
-                        await Authorize().ConfigureAwait(false);
+                        Authorize().GetAwaiter().GetResult();
 
                         // re-try our data request, using the updated bearer token
                         response = _client.Send(
@@ -131,18 +145,15 @@ public async Task<DynamicsQueryResponse> RunDynamicsQuery(DynamicsQuery query)
                         throw new Exception(response.StatusCode.ToString());
                     }
 
-                    return JsonConvert.DeserializeObject<DynamicsQueryResponse>(await response.Content.ReadAsStringAsync().ConfigureAwait(false));
+                    results.Add(query, JsonConvert.DeserializeObject<DynamicsQueryResponse>(response.Content.ReadAsStringAsync().GetAwaiter().GetResult()));
                 }
                 catch (Exception ex)
                 {
                     Console.WriteLine("RunDynamicsQuery: " + ex.Message);
-                    return null;
-                }
-            }
-            else
-            {
-                return null;
-            }
-        }
+                }
+            }
+
+            return results;
+        }
     }
 }

Interfaces/IDataService.cs

@@ -0,0 +1,9 @@
+namespace UA_DataProcessor.Interfaces
+{
+    public interface IDataService : IDisposable
+    {
+        public void Connect();
+
+        public Dictionary<string, object> RunQuery(string query);
+    }
+}

CarbonIntensity.cs Models/CarbonIntensity.csCarbonIntensity.cs renamed to Models/CarbonIntensity.cs

@@ -2,13 +2,12 @@
 using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
 using Opc.Ua.Cloud.Client.Models;
-using System.Collections.Concurrent;
 using System.Net.Http.Headers;
 using System.Text;
 
 namespace Opc.Ua.Data.Processor
 {
-    public class ProductCarbonFootprintService
+    public class PCFProcessor
     {
         private readonly ADXDataService _adxDataService = new ADXDataService();
         private readonly DynamicsDataService _dynamicsDataService = new DynamicsDataService();
@@ -21,15 +20,21 @@ public class ProductCarbonFootprintService
             }
         };
 
-        public void GeneratePCFs()
+        public PCFProcessor()
+        {
+            _adxDataService.Connect();
+            _dynamicsDataService.Connect();
+        }
+
+        public void Process()
         {
             // we have two production lines in the manufacturing ontologies production line simulation and they are connected like so:
             // assembly -> test -> packaging
-            GeneratePCFForProductionLine("Munich", "48.1375", "11.575", 6);
-            GeneratePCFForProductionLine("Seattle", "47.609722", "-122.333056", 10);
+            CalcPCFForProductionLine("Munich", "48.1375", "11.575", 6);
+            CalcPCFForProductionLine("Seattle", "47.609722", "-122.333056", 10);
         }
 
-        private void GeneratePCFForProductionLine(string productionLineName, string latitude, string longitude, int idealCycleTime)
+        private void CalcPCFForProductionLine(string productionLineName, string latitude, string longitude, int idealCycleTime)
         {
             try
             {
@@ -39,21 +44,21 @@ private void GeneratePCFForProductionLine(string productionLineName, string lati
                 {
                     // check if a new product was produced (last machine in the production line, i.e. packaging, is in state 2 ("done") with a passed QA)
                     // and get the products serial number and energy consumption at that time
-                    ConcurrentDictionary<string, object> latestProductProduced = ADXQueryForSpecificValue("packaging", productionLineName, "Status", 2);
+                    Dictionary<string, object> latestProductProduced = ADXQueryForSpecificValue("packaging", productionLineName, "Status", 2);
                     if ((latestProductProduced != null) && (latestProductProduced.Count > 0))
                     {
-                        ConcurrentDictionary<string, object> serialNumberResult = ADXQueryForSpecificTime("packaging", productionLineName, "ProductSerialNumber", ((DateTime)latestProductProduced["Timestamp"]).ToString("yyyy-MM-dd HH:mm:ss"), idealCycleTime);
+                        Dictionary<string, object> serialNumberResult = ADXQueryForSpecificTime("packaging", productionLineName, "ProductSerialNumber", ((DateTime)latestProductProduced["Timestamp"]).ToString("yyyy-MM-dd HH:mm:ss"), idealCycleTime);
                         double serialNumber = (double)serialNumberResult["OPCUANodeValue"];
 
-                        ConcurrentDictionary<string, object> timeItWasProducedPackaging = ADXQueryForSpecificValue("packaging", productionLineName, "ProductSerialNumber", serialNumber);
-                        ConcurrentDictionary<string, object> energyPackaging = ADXQueryForSpecificTime("packaging", productionLineName, "EnergyConsumption", ((DateTime)timeItWasProducedPackaging["Timestamp"]).ToString("yyyy-MM-dd HH:mm:ss"), idealCycleTime);
+                        Dictionary<string, object> timeItWasProducedPackaging = ADXQueryForSpecificValue("packaging", productionLineName, "ProductSerialNumber", serialNumber);
+                        Dictionary<string, object> energyPackaging = ADXQueryForSpecificTime("packaging", productionLineName, "EnergyConsumption", ((DateTime)timeItWasProducedPackaging["Timestamp"]).ToString("yyyy-MM-dd HH:mm:ss"), idealCycleTime);
 
                         // check each other machine for the time when the product with this serial number was in the machine and get its energy comsumption at that time
-                        ConcurrentDictionary<string, object> timeItWasProducedTest = ADXQueryForSpecificValue("test", productionLineName, "ProductSerialNumber", serialNumber);
-                        ConcurrentDictionary<string, object> energyTest = ADXQueryForSpecificTime("test", productionLineName, "EnergyConsumption", ((DateTime)timeItWasProducedTest["Timestamp"]).ToString("yyyy-MM-dd HH:mm:ss"), idealCycleTime);
+                        Dictionary<string, object> timeItWasProducedTest = ADXQueryForSpecificValue("test", productionLineName, "ProductSerialNumber", serialNumber);
+                        Dictionary<string, object> energyTest = ADXQueryForSpecificTime("test", productionLineName, "EnergyConsumption", ((DateTime)timeItWasProducedTest["Timestamp"]).ToString("yyyy-MM-dd HH:mm:ss"), idealCycleTime);
 
-                        ConcurrentDictionary<string, object> timeItWasProducedAssembly = ADXQueryForSpecificValue("assembly", productionLineName, "ProductSerialNumber", serialNumber);
-                        ConcurrentDictionary<string, object> energyAssembly = ADXQueryForSpecificTime("assembly", productionLineName, "EnergyConsumption", ((DateTime)timeItWasProducedAssembly["Timestamp"]).ToString("yyyy-MM-dd HH:mm:ss"), idealCycleTime);
+                        Dictionary<string, object> timeItWasProducedAssembly = ADXQueryForSpecificValue("assembly", productionLineName, "ProductSerialNumber", serialNumber);
+                        Dictionary<string, object> energyAssembly = ADXQueryForSpecificTime("assembly", productionLineName, "EnergyConsumption", ((DateTime)timeItWasProducedAssembly["Timestamp"]).ToString("yyyy-MM-dd HH:mm:ss"), idealCycleTime);
 
                         // calculate the total energy consumption for the product by summing up all the machines' energy consumptions (in Ws), divide by 3600 to get seconds and multiply by the ideal cycle time (which is in seconds)
                         double energyTotal = ((double)energyAssembly["OPCUANodeValue"] + (double)energyTest["OPCUANodeValue"] + (double)energyPackaging["OPCUANodeValue"]) / 3600 * idealCycleTime;
@@ -77,7 +82,7 @@ private void GeneratePCFForProductionLine(string productionLineName, string lati
             }
             catch (Exception ex)
             {
-                Console.WriteLine("GeneratePCFForProductionLine: " + ex.Message);
+                Console.WriteLine("CalcPCFForProductionLine: " + ex.Message);
             }
         }
 
@@ -132,17 +137,15 @@ private float RetrieveScope3Emissions()
         {
             try
             {
-                DynamicsQueryResponse response = _dynamicsDataService.RunDynamicsQuery(new DynamicsQuery() {
-                    tracingDirection = "Backward",
-                    company = Environment.GetEnvironmentVariable("DYNAMICS_COMPANY_NAME"),
-                    itemNumber = Environment.GetEnvironmentVariable("DYNAMICS_PRODUCT_NAME"),
-                    serialNumber = Environment.GetEnvironmentVariable("DYNAMICS_BATCH_NAME"),
-                    shouldIncludeEvents = true
-                }).GetAwaiter().GetResult();
-
-                if (response != null)
+                string query = "Backward" + "\r\n"
+                    + Environment.GetEnvironmentVariable("DYNAMICS_COMPANY_NAME") + "\r\n"
+                    + Environment.GetEnvironmentVariable("DYNAMICS_PRODUCT_NAME") + "\r\n"
+                    + Environment.GetEnvironmentVariable("DYNAMICS_BATCH_NAME") + "\r\n";
+
+                Dictionary<string, object> response = _dynamicsDataService.RunQuery(query);
+                if (response.ContainsKey(query) && (response[query] != null) && response[query] is DynamicsQueryResponse dynamicsResponse)
                 {
-                    return FindPcf(response.root);
+                    return FindPcf(dynamicsResponse.root);
                 }
                 else
                 {
@@ -191,7 +194,7 @@ private float FindPcf(ErpNode node)
             return 0.0f;
         }
 
-        private ConcurrentDictionary<string, object> ADXQueryForSpecificValue(string stationName, string productionLineName, string valueToQuery, double desiredValue)
+        private Dictionary<string, object> ADXQueryForSpecificValue(string stationName, string productionLineName, string valueToQuery, double desiredValue)
         {
             string query = "opcua_metadata_lkv\r\n"
                          + "| where Name contains \"" + stationName + "\"\r\n"
@@ -203,13 +206,10 @@ private ConcurrentDictionary<string, object> ADXQueryForSpecificValue(string sta
                          + "| distinct Timestamp, OPCUANodeValue = todouble(Value)\r\n"
                          + "| sort by Timestamp desc";
 
-            ConcurrentDictionary<string, object> values = new ConcurrentDictionary<string, object>();
-            _adxDataService.RunADXQuery(query, values);
-
-            return values;
+            return _adxDataService.RunQuery(query);
         }
 
-        private ConcurrentDictionary<string, object> ADXQueryForSpecificTime(string stationName, string productionLineName, string valueToQuery, string timeToQuery, int idealCycleTime)
+        private Dictionary<string, object> ADXQueryForSpecificTime(string stationName, string productionLineName, string valueToQuery, string timeToQuery, int idealCycleTime)
         {
             string query = "opcua_metadata_lkv\r\n"
                          + "| where Name contains \"" + stationName + "\"\r\n"
@@ -222,10 +222,7 @@ private ConcurrentDictionary<string, object> ADXQueryForSpecificTime(string stat
                          + "| where around(Timestamp, datetime(" + timeToQuery + "), " + idealCycleTime.ToString() + "s)\r\n"
                          + "| sort by Timestamp desc";
 
-            ConcurrentDictionary<string, object> values = new ConcurrentDictionary<string, object>();
-            _adxDataService.RunADXQuery(query, values);
-
-            return values;
+            return _adxDataService.RunQuery(query);
         }
     }
 }

Dynamics.cs Models/Dynamics.csDynamics.cs renamed to Models/Dynamics.cs

@@ -27,24 +27,19 @@
  * http://opcfoundation.org/License/MIT/1.00/
  * ======================================================================*/
 
-using Newtonsoft.Json;
-using Opc.Ua.Cloud.Client.Models;
-using System.Globalization;
-using System.Text;
-
 [assembly: CLSCompliant(false)]
 namespace Opc.Ua.Data.Processor
 {
     public class Program
     {
         static async Task Main()
         {
-            ProductCarbonFootprintService pcfService = new ProductCarbonFootprintService();
+            PCFProcessor pcfProcessor = new PCFProcessor();
             while (true)
             {
                 try
                 {
-                    pcfService.GeneratePCFs();
+                    pcfProcessor.Process();
                     await Task.Delay(5000).ConfigureAwait(false); // Wait 5s before next processing
                 }
                 catch (Exception ex)