diff --git a/DataAggregator.sln b/DataAggregator.sln index cddc1c4..7c3b036 100644 --- a/DataAggregator.sln +++ b/DataAggregator.sln @@ -17,8 +17,6 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Collector.Te EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Registration.Tests", "tests\DataAggregator.Registration.Tests\DataAggregator.Registration.Tests.csproj", "{FB0B4F80-5801-407F-8425-54069C93DB60}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Integration.Tests", "tests\DataAggregator.Integration.Tests\DataAggregator.Integration.Tests.csproj", "{12FEEB75-A03A-4389-980B-6B735C550CAC}" -EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{02EA681E-C7D8-13C7-8484-4AC65E1B71E8}" ProjectSection(SolutionItems) = preProject .editorconfig = .editorconfig @@ -38,8 +36,6 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Collector.Op EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Processor", "src\DataAggregator.Processor\DataAggregator.Processor.csproj", "{039EC00D-5EDF-4C48-B449-29CFB6750232}" EndProject -Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "processor", "processor", "{330C7B17-DB90-458C-B630-99F9C1B5EA45}" -EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "processor", "processor", "{EB9A5576-3E00-4007-8C72-2235E0A1546D}" EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Processor.Tests", "tests\DataAggregator.Processor.Tests\DataAggregator.Processor.Tests.csproj", "{EFE47FE6-F41E-CDD6-0991-472080AD88B0}" @@ -114,18 +110,6 @@ Global {FB0B4F80-5801-407F-8425-54069C93DB60}.Release|x64.Build.0 = Release|x64 {FB0B4F80-5801-407F-8425-54069C93DB60}.Release|x86.ActiveCfg = Release|x86 {FB0B4F80-5801-407F-8425-54069C93DB60}.Release|x86.Build.0 = Release|x86 - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|Any CPU.Build.0 = Debug|Any CPU - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|x64.ActiveCfg = Debug|x64 - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|x64.Build.0 = Debug|x64 - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|x86.ActiveCfg = Debug|x86 - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|x86.Build.0 = Debug|x86 - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|Any CPU.ActiveCfg = Release|Any CPU - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|Any CPU.Build.0 = Release|Any CPU - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|x64.ActiveCfg = Release|x64 - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|x64.Build.0 = Release|x64 - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|x86.ActiveCfg = Release|x86 - {12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|x86.Build.0 = Release|x86 {4A5708E5-B5D1-4DA4-A2E4-323D3A21B182}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {4A5708E5-B5D1-4DA4-A2E4-323D3A21B182}.Debug|Any CPU.Build.0 = Debug|Any CPU {4A5708E5-B5D1-4DA4-A2E4-323D3A21B182}.Debug|x64.ActiveCfg = Debug|x64 @@ -184,7 +168,6 @@ Global {E593EA7B-F713-4444-AE40-FDCEF462660A} = {AA521B32-6E75-46BD-ABDF-0450227A94C4} {63D70F2C-9C7D-4397-8150-37F343949530} = {247EF7A2-1DFD-4B51-AC7D-0FD13827CAC2} {FB0B4F80-5801-407F-8425-54069C93DB60} = {247EF7A2-1DFD-4B51-AC7D-0FD13827CAC2} - {12FEEB75-A03A-4389-980B-6B735C550CAC} = {247EF7A2-1DFD-4B51-AC7D-0FD13827CAC2} {95F8787D-6FE2-4173-8450-6762EA6FAE1F} = {E1AD9667-4C40-4CAF-8096-5FA749EBB2B1} {AA521B32-6E75-46BD-ABDF-0450227A94C4} = {E1AD9667-4C40-4CAF-8096-5FA749EBB2B1} {F5250AC4-9CCC-432B-8725-DAC6AE01CCF0} = {95F8787D-6FE2-4173-8450-6762EA6FAE1F} diff --git a/src/DataAggregator.Collector.Shared/Abstraction/CollectorService.cs b/src/DataAggregator.Collector.Shared/Abstraction/CollectorService.cs index a9b5c79..efb68fe 100644 --- a/src/DataAggregator.Collector.Shared/Abstraction/CollectorService.cs +++ b/src/DataAggregator.Collector.Shared/Abstraction/CollectorService.cs @@ -27,7 +27,7 @@ public class CollectorService( private readonly SemaphoreSlim _processingLock = new(1, 1); private readonly ConcurrentQueue _dataQueue = new(); - private readonly TimeSpan _flushInterval = TimeSpan.FromSeconds(1); // Flush every 1 second + private readonly TimeSpan _flushInterval = TimeSpan.FromMilliseconds(configuration.FlushIntervalMilliseconds); // Flush every 1 second private bool _isRunning; private CancellationTokenSource? _cancellationTokenSource; diff --git a/src/DataAggregator.Collector.Shared/Abstraction/Configuration/CollectorConfiguration.cs b/src/DataAggregator.Collector.Shared/Abstraction/Configuration/CollectorConfiguration.cs index ce549a6..ff15dbe 100644 --- a/src/DataAggregator.Collector.Shared/Abstraction/Configuration/CollectorConfiguration.cs +++ b/src/DataAggregator.Collector.Shared/Abstraction/Configuration/CollectorConfiguration.cs @@ -24,4 +24,9 @@ public class CollectorConfiguration /// Gets or sets the list of sensors configured for this collector. /// public List Sensors { get; set; } = []; + + /// + /// Gets or sets the interval in milliseconds for flushing data to the time series database. + /// + public int FlushIntervalMilliseconds { get; set; } = 1000; } diff --git a/src/DataAggregator.Collector/appsettings.json b/src/DataAggregator.Collector/appsettings.json index e7cb0b7..83198c7 100644 --- a/src/DataAggregator.Collector/appsettings.json +++ b/src/DataAggregator.Collector/appsettings.json @@ -20,6 +20,7 @@ "DeviceName": "Micro5", "Location": "Test", "HealthCheckEndpoint": "http://localhost:5000/health", + "FlushIntervalMilliseconds": 1000, "SamplingRate": 100, "CapnProto": { "ServerAddress": "192.168.53.15", diff --git a/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs b/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs index 5f94703..f681ef4 100644 --- a/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs +++ b/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs @@ -1,4 +1,6 @@ -namespace DataAggregator.Processor.Configuration; +using DataAggregator.Processor.Services.Processing.Factory; + +namespace DataAggregator.Processor.Configuration; /// /// Configuration for a specific machine prediction. @@ -16,32 +18,27 @@ public class MachinePredictionConfig public bool Enabled { get; set; } = true; /// - /// Gets or sets the path to the ONNX model file. - /// - public string ModelPath { get; set; } = string.Empty; - - /// - /// Gets or sets the preprocessing strategy name. + /// Gets or sets the list of input sensor names. /// - public string PreprocessingStrategy { get; set; } = string.Empty; + public List InputSensors { get; set; } = []; /// - /// Gets or sets the list of input sensor names. + /// Gets or sets a value indicating whether Window size in seconds if true, otherwise in elements number. /// - public List InputSensors { get; set; } = []; + public bool WindowSizeInSeconds { get; set; } = true; /// - /// Gets or sets the window size in seconds for data collection. + /// Gets or sets the window size unit depends on WindowSizeInSeconds property. /// - public int WindowSizeSeconds { get; set; } = 60; + public int WindowSize { get; set; } = 60; /// /// Gets or sets the cycle interval in seconds for this machine. /// - public int CycleIntervalSeconds { get; set; } = 1; + public double CycleIntervalSeconds { get; set; } = 1; /// - /// Gets or sets the preprocessing configuration for Z-score normalization. + /// Gets or sets the processing pipeline for this machine. /// - public PreprocessingConfig Preprocessing { get; set; } = new(); + public List ProcessingPipeline { get; set; } = new(); } diff --git a/src/DataAggregator.Processor/Program.cs b/src/DataAggregator.Processor/Program.cs index d71ea41..6827537 100644 --- a/src/DataAggregator.Processor/Program.cs +++ b/src/DataAggregator.Processor/Program.cs @@ -1,8 +1,9 @@ +using System.Text.Json; using DataAggregator.Processor.Configuration; using DataAggregator.Processor.Services; using DataAggregator.Processor.Services.DataStorage; using DataAggregator.Processor.Services.Prediction; -using DataAggregator.Processor.Services.PreProcessing; +using DataAggregator.Processor.Services.Processing.Factory; using DataAggregator.Processor.Services.Registration; using Serilog; @@ -25,8 +26,18 @@ // Register health checks builder.Services.AddHealthChecks(); -// Configure prediction service -builder.Services.Configure(builder.Configuration.GetSection("PredictionService")); +// Configuration management +string appSettingsJson = File.ReadAllText("appsettings.json"); +using var doc = JsonDocument.Parse(appSettingsJson); +JsonElement predictionServiceElement = doc.RootElement.GetProperty("PredictionService"); +string json = predictionServiceElement.GetRawText(); + +var options = new JsonSerializerOptions { PropertyNameCaseInsensitive = true }; +options.Converters.Add(new ProcessorDescriptionJsonConverter()); +PredictionServiceConfiguration? predictionConfig = JsonSerializer.Deserialize(json, options); +if (predictionConfig == null) + throw new Exception("Failed to deserialize PredictionServiceConfiguration"); +builder.Services.AddSingleton(predictionConfig); // Configure HTTP clients builder.Services.AddHttpClient("RegistrationClient", client => @@ -38,8 +49,7 @@ // Register services builder.Services.AddScoped(); -builder.Services.AddSingleton(); -builder.Services.AddSingleton(); +builder.Services.AddSingleton(); builder.Services.AddScoped(); // Register background service diff --git a/src/DataAggregator.Processor/Services/DataStorage/IDataRepository.cs b/src/DataAggregator.Processor/Services/DataStorage/IDataRepository.cs index 83a5eaa..833a89c 100644 --- a/src/DataAggregator.Processor/Services/DataStorage/IDataRepository.cs +++ b/src/DataAggregator.Processor/Services/DataStorage/IDataRepository.cs @@ -13,7 +13,7 @@ public interface IDataRepository /// /// The InfluxDB endpoint. /// The authentication token. - public void InitializeAsync(string endpoint, string token); + public void Initialize(string endpoint, string token); /// /// Queries measurements from InfluxDB for a specific time range and sensors with type information. @@ -25,6 +25,15 @@ public interface IDataRepository /// A list of measurement data. public Task> QueryMeasurementsAsync(string table, DateTime startTime, DateTime endTime, List sensors); + /// + /// Queries the last measurements from InfluxDB for a specific table and window size, filtering by sensors. + /// + /// The table name. + /// The size of the window. + /// The list of sensor information with type data. + /// A list of measurements data of windowSize size max. + public Task> QueryLastMeasurements(string table, int windowSize, List sensors); + /// /// Writes a single measurement to InfluxDB. /// diff --git a/src/DataAggregator.Processor/Services/DataStorage/InfluxV3Repository.cs b/src/DataAggregator.Processor/Services/DataStorage/InfluxV3Repository.cs index f8cae55..7269174 100644 --- a/src/DataAggregator.Processor/Services/DataStorage/InfluxV3Repository.cs +++ b/src/DataAggregator.Processor/Services/DataStorage/InfluxV3Repository.cs @@ -17,7 +17,7 @@ public class InfluxV3Repository : IDataRepository, IDisposable private InfluxDBClient? _client; /// - public void InitializeAsync(string endpoint, string token) + public void Initialize(string endpoint, string token) { _client?.Dispose(); @@ -43,97 +43,52 @@ public void InitializeAsync(string endpoint, string token) } /// - public async Task> QueryMeasurementsAsync(string table, DateTime startTime, DateTime endTime, List sensors) + public async Task> QueryMeasurementsAsync( + string table, + DateTime startTime, + DateTime endTime, + List sensors) { if (_client == null) { throw new InvalidOperationException("InfluxDB client is not initialized."); } - try - { - string sensorColumns = string.Join(", ", sensors.Select(s => $"\"{s.SensorName}\"")); - - string query = $""" - SELECT time, {sensorColumns} - FROM "{table}" - WHERE time >= '{startTime:yyyy-MM-ddTHH:mm:ssZ}' - AND time < '{endTime:yyyy-MM-ddTHH:mm:ssZ}' - """; + string sensorColumns = string.Join(", ", sensors.Select(s => $"\"{s.SensorName}\"")); - var measurements = new List(); - var sensorDict = sensors.ToDictionary(s => s.SensorName, s => s); + string query = $""" + SELECT time, {sensorColumns} + FROM "{table}" + WHERE time >= '{startTime:yyyy-MM-ddTHH:mm:ssZ}' + AND time < '{endTime:yyyy-MM-ddTHH:mm:ssZ}' + """; - await foreach (PointDataValues point in _client.QueryPoints(query)) - { - // point contains the original structure with all fields at once - // This matches how we write data: one row per timestamp with multiple sensors - System.Numerics.BigInteger? timestampBigInt = point.GetTimestamp(); - if (timestampBigInt == null) - { - Log.Warning("Skipping point with null timestamp"); - continue; - } - - // Convert BigInteger timestamp to DateTime - DateTime timestamp = DateTimeOffset.FromUnixTimeMilliseconds((long)(timestampBigInt.Value / 1_000_000)).DateTime; + return await ExecuteMeasurementQuery(query, sensors, table, startTime, endTime); + } - string[] fieldsNames = point.GetFieldNames(); - foreach (string fieldName in fieldsNames) - { - string sensorName = fieldName; - object? value = point.GetField(fieldName); + /// + public async Task> QueryLastMeasurements( + string table, + int windowSize, + List sensors) + { + if (_client == null) + { + throw new InvalidOperationException("InfluxDB client is not initialized."); + } - // Only include sensors that were requested - if (sensorDict.TryGetValue(sensorName, out SensorInfoDto? sensorInfo) && value != null) - { - IMeasurementData? measurement = sensorInfo.DataType switch - { - SensorDataType.Boolean when bool.TryParse(value.ToString(), out bool boolValue) => - new MeasurementData(timestamp, sensorName, boolValue), - - SensorDataType.Integer when int.TryParse(value.ToString(), out int intValue) => - new MeasurementData(timestamp, sensorName, intValue), - - SensorDataType.Double or SensorDataType.Float when double.TryParse(value.ToString(), out double doubleValue) => - new MeasurementData(timestamp, sensorName, doubleValue), - - SensorDataType.String => - new MeasurementData(timestamp, sensorName, value.ToString() ?? string.Empty), - - _ => null, - }; - - if (measurement != null) - { - measurements.Add(measurement); - } - else - { - Log.Debug( - "Skipping value for sensor {Sensor} with type {DataType}: {Value}", - sensorName, - sensorInfo.DataType, - value); - } - } - } - } + string sensorColumns = string.Join(", ", sensors.Select(s => $"\"{s.SensorName}\"")); - Log.Debug( - "Queried {Count} measurements for table {Table} from {StartTime} to {EndTime}", - measurements.Count, - table, - startTime, - endTime); + string query = $""" + SELECT time, {sensorColumns} + FROM "{table}" + ORDER BY time DESC + LIMIT {windowSize} + """; - return measurements; - } - catch (Exception ex) - { - Log.Error(ex, "Failed to query measurements from InfluxDB for table {Table}", table); - throw; - } + // Note: sorting back to ascending to preserve chronological order + List reversed = await ExecuteMeasurementQuery(query, sensors, table); + return reversed.OrderBy(m => m.TimeStamp).ToList(); } /// @@ -172,6 +127,104 @@ public async Task WriteMeasurementAsync(string tag, IEnumerable + /// Execute a generic InfluxDB query and convert results to a flat list of IMeasurementData. + /// Assumes the data is stored in a "wide" format (one row per timestamp, multiple sensors as columns). + /// + private async Task> ExecuteMeasurementQuery( + string query, + List sensors, + string table, + DateTime? start = null, + DateTime? end = null) + { + if (_client is null) + { + return []; + } + + var measurements = new List(); + var sensorDict = sensors.ToDictionary(s => s.SensorName, s => s); + + try + { + await foreach (PointDataValues point in _client.QueryPoints(query)) + { + // point contains the original structure with all fields at once + // This matches how we write data: one row per timestamp with multiple sensors + System.Numerics.BigInteger? timestampBigInt = point.GetTimestamp(); + if (timestampBigInt == null) + { + Log.Warning("Skipping point with null timestamp"); + continue; + } + + // Convert BigInteger timestamp to DateTime + DateTime timestamp = DateTimeOffset + .FromUnixTimeMilliseconds((long)(timestampBigInt.Value / 1_000_000)) + .DateTime; + + string[] fieldNames = point.GetFieldNames(); + foreach (string fieldName in fieldNames) + { + string sensorName = fieldName; + + // Only include sensors that were requested + if (!sensorDict.TryGetValue(sensorName, out SensorInfoDto? sensorInfo)) + continue; + + object? value = point.GetField(sensorName); + if (value == null) + continue; + + // Deserialize the value according to the expected data type + IMeasurementData? measurement = sensorInfo.DataType switch + { + SensorDataType.Boolean when bool.TryParse(value.ToString(), out bool b) => + new MeasurementData(timestamp, sensorName, b), + + SensorDataType.Integer when int.TryParse(value.ToString(), out int i) => + new MeasurementData(timestamp, sensorName, i), + + SensorDataType.Double or SensorDataType.Float when double.TryParse(value.ToString(), out double d) => + new MeasurementData(timestamp, sensorName, d), + + SensorDataType.String => + new MeasurementData(timestamp, sensorName, value.ToString() ?? string.Empty), + + _ => null, + }; + + if (measurement != null) + { + measurements.Add(measurement); + } + else + { + Log.Debug( + "Skipping value for sensor {Sensor} with type {DataType}: {Value}", + sensorName, + sensorInfo.DataType, + value); + } + } + } + + Log.Debug( + "Fetched {Count} measurements from {Table} {Range}", + measurements.Count, + table, + start != null && end != null ? $"from {start} to {end}" : "(last N)"); + + return measurements; + } + catch (Exception ex) + { + Log.Error(ex, "Query failed for table {Table}", table); + throw; + } + } + /// /// Disposes the InfluxDB client. /// diff --git a/src/DataAggregator.Processor/Services/PreProcessing/IPreprocessingStrategy.cs b/src/DataAggregator.Processor/Services/PreProcessing/IPreprocessingStrategy.cs deleted file mode 100644 index 1f2bb74..0000000 --- a/src/DataAggregator.Processor/Services/PreProcessing/IPreprocessingStrategy.cs +++ /dev/null @@ -1,18 +0,0 @@ -using DataAggregator.Collector.Shared.Models; -using DataAggregator.Processor.Configuration; - -namespace DataAggregator.Processor.Services.PreProcessing; - -/// -/// Interface for preprocessing strategies that convert raw measurement data into feature vectors for ML models. -/// -public interface IPreprocessingStrategy -{ - /// - /// Preprocesses a list of measurements into a feature vector for a single prediction sample. - /// - /// List of raw measurements from the data window. - /// Configuration for the machine prediction. - /// Feature vector as dictionary mapping input names to values for a single sample. - public IEnumerable PreprocessAsync(IEnumerable measurements, MachinePredictionConfig config); -} diff --git a/src/DataAggregator.Processor/Services/PreProcessing/IPreprocessingStrategyFactory.cs b/src/DataAggregator.Processor/Services/PreProcessing/IPreprocessingStrategyFactory.cs deleted file mode 100644 index 71715c8..0000000 --- a/src/DataAggregator.Processor/Services/PreProcessing/IPreprocessingStrategyFactory.cs +++ /dev/null @@ -1,14 +0,0 @@ -namespace DataAggregator.Processor.Services.PreProcessing; - -/// -/// Factory interface for creating preprocessing strategies based on strategy name. -/// -public interface IPreprocessingStrategyFactory -{ - /// - /// Creates a preprocessing strategy based on the strategy name. - /// - /// Name of the strategy to create. - /// Configured preprocessing strategy. - public IPreprocessingStrategy CreateStrategy(string strategyName); -} diff --git a/src/DataAggregator.Processor/Services/PreProcessing/PreprocessingStrategyFactory.cs b/src/DataAggregator.Processor/Services/PreProcessing/PreprocessingStrategyFactory.cs deleted file mode 100644 index f7c7689..0000000 --- a/src/DataAggregator.Processor/Services/PreProcessing/PreprocessingStrategyFactory.cs +++ /dev/null @@ -1,26 +0,0 @@ -using DataAggregator.Processor.Services.PreProcessing.ActuatorMergingCurrentPreprocessing; -using Serilog; - -namespace DataAggregator.Processor.Services.PreProcessing; - -/// -/// Factory implementation for creating preprocessing strategies. -/// -public class PreprocessingStrategyFactory : IPreprocessingStrategyFactory -{ - /// - /// Creates a preprocessing strategy based on the strategy name. - /// - /// Name of the strategy to create. - /// Configured preprocessing strategy. - public IPreprocessingStrategy CreateStrategy(string strategyName) - { - Log.Information("Creating preprocessing strategy: {StrategyName}", strategyName); - - return strategyName.ToLowerInvariant() switch - { - "actuatorcurrent" => new ActuatorCurrentFeatureExtractor(), - _ => throw new ArgumentException($"Unknown preprocessing strategy: {strategyName}", nameof(strategyName)), - }; - } -} diff --git a/src/DataAggregator.Processor/Services/Prediction/IOnnxPredictionEngine.cs b/src/DataAggregator.Processor/Services/Prediction/IOnnxPredictionEngine.cs deleted file mode 100644 index 4e99b7c..0000000 --- a/src/DataAggregator.Processor/Services/Prediction/IOnnxPredictionEngine.cs +++ /dev/null @@ -1,17 +0,0 @@ -using DataAggregator.Collector.Shared.Models; - -namespace DataAggregator.Processor.Services.Prediction; - -/// -/// Interface for ONNX prediction engine. -/// -public interface IOnnxPredictionEngine -{ - /// - /// Performs prediction using an ONNX model. - /// - /// The path to the ONNX model file. - /// The input data for prediction as a dictionary mapping input names to values. - /// The prediction results as a dictionary mapping output names to values. - public Task> PredictAsync(string modelPath, IEnumerable inputData); -} diff --git a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs index c98b051..b8ff3d7 100644 --- a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs +++ b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs @@ -1,123 +1,53 @@ using DataAggregator.Collector.Shared.Models; using DataAggregator.Processor.Configuration; using DataAggregator.Processor.Services.DataStorage; -using DataAggregator.Processor.Services.PreProcessing; +using DataAggregator.Processor.Services.Processing.Abstraction; +using DataAggregator.Processor.Services.Processing.Factory; using DataAggregator.Processor.Services.Registration; using DataAggregator.Shared.DTOs; using Serilog; namespace DataAggregator.Processor.Services.Prediction; -/// -/// Processor for machine prediction operations. -/// -/// -/// Initializes a new instance of the class. -/// -/// The InfluxDB repository. -/// The registration service client. -/// The ONNX prediction engine. -/// The preprocessing strategy factory. +/// public class MachinePredictionProcessor( IDataRepository influxRepository, IRegistrationServiceClient registrationClient, - IOnnxPredictionEngine predictionEngine, - IPreprocessingStrategyFactory strategyFactory) : IMachinePredictionProcessor + IDataProcessorFactory processorFactory) : IMachinePredictionProcessor { - #region Private fields - - // Track the last endpoint used to avoid unnecessary reinitializations - private string? _lastEndpoint; - - #endregion - - #region Public methods - - /// + /// public async Task ProcessAsync(MachinePredictionConfig config) { try { Log.Debug("Starting prediction process for machine {MachineName}", config.MachineName); - // Get collector info from registration service - CollectorInfoDto? collectorInfo = await registrationClient.GetCollectorInfoAsync(config.MachineName); + CollectorInfoDto? collectorInfo = await FetchCollectorInfo(config.MachineName); + if (collectorInfo == null) return; - if (collectorInfo == null) - { - Log.Warning("Collector info not found for machine {MachineName}", config.MachineName); - return; - } + List? requestedSensors = ValidateAndGetRequestedSensors(config, collectorInfo); + if (requestedSensors == null) return; - // Validate that all required sensors are available - var availableSensors = collectorInfo.Sensors.ToDictionary(s => s.SensorName, s => s); - var requestedSensors = config.InputSensors.Where(s => availableSensors.ContainsKey(s)).ToList(); + InitializeRepositoryIfNeeded(collectorInfo); - if (requestedSensors.Count != config.InputSensors.Count) - { - IEnumerable missingSensors = config.InputSensors.Except(requestedSensors); - Log.Warning( - "Missing sensors for machine {MachineName}: {MissingSensors}", - config.MachineName, - string.Join(", ", missingSensors)); - - if (requestedSensors.Count == 0) - { - Log.Error("No valid sensors found for machine {MachineName}", config.MachineName); - return; - } - } - - // Initialize InfluxDB repository only if endpoint changed - if (_lastEndpoint != collectorInfo.AssignedInfluxEndpoint.Endpoint) - { - influxRepository.InitializeAsync( - collectorInfo.AssignedInfluxEndpoint.Endpoint, - collectorInfo.AssignedInfluxEndpoint.Token); - - _lastEndpoint = collectorInfo.AssignedInfluxEndpoint.Endpoint; - Log.Debug("Reinitialized InfluxDB connection with new endpoint: {Endpoint}", _lastEndpoint); - } - - // Get sensor info for requested sensors - var requestedSensorInfos = requestedSensors - .Select(sensorName => availableSensors[sensorName]) - .ToList(); - - // Fetch data window with sensor type information - List measurements = await FetchDataWindowAsync(config, requestedSensorInfos); - - if (measurements.Count == 0) + IEnumerable> measurements = await FetchDataWindowAsync(config, requestedSensors); + if (!measurements.Any()) { Log.Warning("No measurements found for machine {MachineName} in the specified time window", config.MachineName); return; } - // Preprocess data using strategy - IEnumerable preprocessedData = PreprocessDataAsync(measurements, config); - - if (preprocessedData == null || !preprocessedData.Any()) - { - Log.Warning("Data preprocessing failed for machine {MachineName}", config.MachineName); - return; - } - - // Perform prediction - string fullPath = Path.Combine(AppContext.BaseDirectory, config.ModelPath); - IEnumerable predictions = await predictionEngine.PredictAsync(fullPath, preprocessedData); - - if (predictions == null || !predictions.Any()) + if (!BuildPipelineIfNeeded(config)) { - Log.Warning("No predictions returned for machine {MachineName}", config.MachineName); + Log.Error("Failed to build processing pipeline for machine {MachineName}", config.MachineName); return; } - // Write prediction to InfluxDB - await influxRepository.WriteMeasurementAsync(config.MachineName, predictions); + IEnumerable? processedData = await RunPipelineAsync(measurements, config.MachineName); + if (processedData == null || !processedData.Any()) return; - Log.Information( - "Prediction completed for machine {MachineName}", - config.MachineName); + await influxRepository.WriteMeasurementAsync(config.MachineName, processedData); + Log.Information("Prediction pipeline completed for machine {MachineName}", config.MachineName); } catch (Exception ex) { @@ -126,49 +56,172 @@ public async Task ProcessAsync(MachinePredictionConfig config) } } - #endregion - #region Private methods + private async Task FetchCollectorInfo(string machineName) + { + CollectorInfoDto? collectorInfo = await registrationClient.GetCollectorInfoAsync(machineName); + if (collectorInfo == null) + Log.Warning("Collector info not found for machine {MachineName}", machineName); + return collectorInfo; + } - private async Task> FetchDataWindowAsync(MachinePredictionConfig config, List sensors) + private List? ValidateAndGetRequestedSensors(MachinePredictionConfig config, CollectorInfoDto collectorInfo) { - DateTime endTime = DateTime.UtcNow; - DateTime startTime = endTime.AddSeconds(-config.WindowSizeSeconds); - - return await influxRepository.QueryMeasurementsAsync( - config.MachineName, - startTime, - endTime, - sensors); + var availableSensors = collectorInfo.Sensors.ToDictionary(s => s.SensorName); + var requestedSensors = config.InputSensors.Where(availableSensors.ContainsKey).ToList(); + + if (requestedSensors.Count != config.InputSensors.Count) + { + IEnumerable missing = config.InputSensors.Except(requestedSensors); + Log.Warning("Missing sensors for machine {MachineName}: {MissingSensors}", config.MachineName, string.Join(", ", missing)); + if (requestedSensors.Count == 0) + { + Log.Error("No valid sensors found for machine {MachineName}", config.MachineName); + return null; + } + } + + return requestedSensors.Select(s => availableSensors[s]).ToList(); } - private IEnumerable PreprocessDataAsync(IEnumerable measurements, MachinePredictionConfig config) + private void InitializeRepositoryIfNeeded(CollectorInfoDto collectorInfo) { - try + if (_lastEndpoint != collectorInfo.AssignedInfluxEndpoint.Endpoint) { - if (string.IsNullOrEmpty(config.PreprocessingStrategy)) + influxRepository.Initialize( + collectorInfo.AssignedInfluxEndpoint.Endpoint, + collectorInfo.AssignedInfluxEndpoint.Token); + _lastEndpoint = collectorInfo.AssignedInfluxEndpoint.Endpoint; + Log.Debug("Reinitialized InfluxDB connection with new endpoint: {Endpoint}", _lastEndpoint); + } + } + + private async Task> RunPipelineAsync( + IEnumerable> dataBlocks, + string machineName) + { + if (_pipelineProcessors is null || _pipelineProcessors.Count == 0) + { + Log.Error("No processing pipeline configured for machine {MachineName}", machineName); + return Array.Empty(); + } + + var results = new List(); + + foreach (IReadOnlyList block in dataBlocks) + { + IEnumerable? current = block; + + foreach (IDataProcessor processor in _pipelineProcessors) { - Log.Error("No preprocessing strategy configured for machine {MachineName}", config.MachineName); - return Array.Empty(); + try + { + current = await processor.ProcessAsync(current); + if (current is null || !current.Any()) + { + Log.Warning("Processor {Processor} returned no data for machine {MachineName} (block skipped)", processor.GetType().Name, machineName); + current = null; + break; + } + } + catch (Exception ex) + { + Log.Error(ex, "Error in processor {Processor} for machine {MachineName} (block skipped)", processor.GetType().Name, machineName); + current = null; + break; + } } - IPreprocessingStrategy strategy = strategyFactory.CreateStrategy(config.PreprocessingStrategy); - IEnumerable preprocessedData = strategy.PreprocessAsync(measurements, config); + if (current is not null) + { + results.AddRange(current); + } + } - Log.Debug( - "Preprocessed data for machine {MachineName} using strategy {Strategy}: {InputCount} inputs", + return results; + } + + private bool BuildPipelineIfNeeded(MachinePredictionConfig config) + { + if (_pipelineProcessors == null) + { + if (config.ProcessingPipeline == null) + { + Log.Error("No processing pipeline configured for machine {MachineName}", config.MachineName); + return false; + } + + _pipelineProcessors = processorFactory.CreateProcessors(config.ProcessingPipeline); + } + + return true; + } + + private async Task>> FetchDataWindowAsync(MachinePredictionConfig config, List sensors) + { + List> measurements = []; + + // If the window size is in secondes, simply query the measurements for the last N seconds. + if (config.WindowSizeInSeconds) + { + DateTime endTime = DateTime.UtcNow; + DateTime startTime = endTime.AddSeconds(-config.WindowSize); + + List datas = await influxRepository.QueryMeasurementsAsync( config.MachineName, - config.PreprocessingStrategy, - preprocessedData.Count()); + startTime, + endTime, + sensors); - return preprocessedData; + measurements.Add(datas); } - catch (Exception ex) + else { - Log.Error(ex, "Error preprocessing data for machine {MachineName}", config.MachineName); - return Array.Empty(); + // If we want to fetch the last N measurements, we must take inaccount the number of sensors + int windowSize = config.WindowSize * config.InputSensors.Count; + + // Otherwise, query all the measurements since the last query, and slice it in windows size list + if (_lastQueryTime == DateTime.MinValue) + { + // For the first run, we assume we want all the element for the CycleIntervalSeconds period. + _lastQueryTime = DateTime.UtcNow.AddSeconds(-config.CycleIntervalSeconds); + } + + List datas = await influxRepository.QueryMeasurementsAsync( + config.MachineName, + _lastQueryTime, + DateTime.UtcNow, + sensors); + + if (datas.Count < windowSize) + return measurements; + + datas = [.. datas.OrderBy(d => d.TimeStamp)]; + + int fullBlockCount = datas.Count / windowSize; + + // Slice the data into windows of size windowSize. + for (int i = 0; i < fullBlockCount; i++) + { + List block = datas.GetRange(i * windowSize, windowSize); + measurements.Add(block); + } + + // Set the last query time to the maximum timestamp of the fetched data complete block, + // so that the next query will only fetch new data. + int lastProcessedIndex = (fullBlockCount * windowSize) - 1; + _lastQueryTime = datas[lastProcessedIndex].TimeStamp; } + + Log.Debug("Fetched {Count} data blocks for machine {MachineName}", measurements.Count, config.MachineName); + + return measurements; } + #endregion + #region Private fields + private List? _pipelineProcessors; + private string? _lastEndpoint; + private DateTime _lastQueryTime = DateTime.MinValue; #endregion } diff --git a/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs b/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs index 796456a..0406eb6 100644 --- a/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs +++ b/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs @@ -1,6 +1,5 @@ using DataAggregator.Processor.Configuration; using DataAggregator.Processor.Services.Prediction; -using Microsoft.Extensions.Options; using Serilog; namespace DataAggregator.Processor.Services; @@ -14,7 +13,7 @@ namespace DataAggregator.Processor.Services; /// The prediction service configuration. /// The service provider. public class PredictionBackgroundService( - IOptions configuration, + PredictionServiceConfiguration configuration, IServiceProvider serviceProvider) : BackgroundService { #region Private fields @@ -37,7 +36,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) ValidateConfigurationAsync(); // Schedule machines - foreach (MachinePredictionConfig machineConfig in configuration.Value.Machines) + foreach (MachinePredictionConfig machineConfig in configuration.Machines) { if (machineConfig.Enabled) { @@ -47,7 +46,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) Log.Information( "Prediction background service started with {MachineCount} machines", - configuration.Value.Machines.Count(m => m.Enabled)); + configuration.Machines.Count(m => m.Enabled)); // Keep the service running while (!stoppingToken.IsCancellationRequested) @@ -84,7 +83,7 @@ public override async Task StopAsync(CancellationToken cancellationToken) private void ValidateConfigurationAsync() { - var enabledMachines = configuration.Value.Machines.Where(m => m.Enabled).ToList(); + var enabledMachines = configuration.Machines.Where(m => m.Enabled).ToList(); if (enabledMachines.Count == 0) { @@ -96,18 +95,6 @@ private void ValidateConfigurationAsync() { try { - // Check if ONNX model file exists - string fullPath = Path.Combine(AppContext.BaseDirectory, machineConfig.ModelPath); - if (!File.Exists(fullPath)) - { - Log.Error( - "ONNX model file not found for machine {MachineName}: {ModelPath}", - machineConfig.MachineName, - machineConfig.ModelPath); - - throw new FileNotFoundException($"ONNX model file not found: {machineConfig.ModelPath}"); - } - // Validate configuration if (string.IsNullOrEmpty(machineConfig.MachineName)) { @@ -127,15 +114,6 @@ private void ValidateConfigurationAsync() throw new InvalidOperationException($"No input sensors configured for machine {machineConfig.MachineName}"); } - if (string.IsNullOrEmpty(machineConfig.PreprocessingStrategy)) - { - Log.Error( - "Preprocessing strategy is not configured for machine {MachineName}", - machineConfig.MachineName); - - throw new InvalidOperationException($"Preprocessing strategy is not configured for machine {machineConfig.MachineName}"); - } - Log.Information("Configuration validated for machine {MachineName}", machineConfig.MachineName); } catch (Exception ex) diff --git a/src/DataAggregator.Processor/Services/Processing/Abstraction/IDataProcessor.cs b/src/DataAggregator.Processor/Services/Processing/Abstraction/IDataProcessor.cs new file mode 100644 index 0000000..3669d5a --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/Abstraction/IDataProcessor.cs @@ -0,0 +1,16 @@ +using DataAggregator.Collector.Shared.Models; + +namespace DataAggregator.Processor.Services.Processing.Abstraction; + +/// +/// Base interface for data processors that handle measurement data. +/// +public interface IDataProcessor +{ + /// + /// Processes a collection of measurement data asynchronously and return a new processed list of measurements. + /// + /// The input measurements. + /// A new IEnumerable of IMeasurementData which is processed. + public Task> ProcessAsync(IEnumerable input); +} diff --git a/src/DataAggregator.Processor/Services/Processing/Abstraction/IProcessorConfiguration.cs b/src/DataAggregator.Processor/Services/Processing/Abstraction/IProcessorConfiguration.cs new file mode 100644 index 0000000..bc235a6 --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/Abstraction/IProcessorConfiguration.cs @@ -0,0 +1,8 @@ +namespace DataAggregator.Processor.Services.Processing.Abstraction; + +/// +/// Flag interface for processor configuration. +/// +public interface IProcessorConfiguration +{ +} diff --git a/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs new file mode 100644 index 0000000..4e0a97a --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs @@ -0,0 +1,47 @@ +using DataAggregator.Processor.Services.Prediction; +using DataAggregator.Processor.Services.Processing.Abstraction; +using DataAggregator.Processor.Services.Processing.Onnx; +using DataAggregator.Processor.Services.Processing.PostProcessing.StateDeductionPostProcess; +using DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; + +namespace DataAggregator.Processor.Services.Processing.Factory; + +/// +/// Defines a factory for creating data processors based on a pipeline description. +/// +public class DataProcessorFactory : IDataProcessorFactory +{ + /// + public List CreateProcessors(IEnumerable pipeline) + { + var processors = new List(); + foreach (ProcessorDescription desc in pipeline) + { + switch (desc.Name.ToLowerInvariant()) + { + case "actuatorcurrent": + if (desc.Configuration is PreprocessingConfig preConfig) + processors.Add(new ActuatorCurrentFeatureExtractor(preConfig)); + else + throw new ArgumentException("Invalid config type for actuatorcurrent"); + break; + case "onnxprediction": + if (desc.Configuration is OnnxPredictionConfig onnxConfig) + processors.Add(new OnnxPredictionEngine(onnxConfig)); + else + throw new ArgumentException("Invalid config type for onnxprediction"); + break; + case "statedeductionpostprocessor": + if (desc.Configuration is StateDeductionPostProcessorConfig stateConfig) + processors.Add(new StateDeductionPostProcessor(stateConfig)); + else + throw new ArgumentException("Invalid config type for statedeductionpostprocessor"); + break; + default: + throw new ArgumentException($"Unknown processor name: {desc.Name}"); + } + } + + return processors; + } +} diff --git a/src/DataAggregator.Processor/Services/Processing/Factory/IDataProcessorFactory.cs b/src/DataAggregator.Processor/Services/Processing/Factory/IDataProcessorFactory.cs new file mode 100644 index 0000000..2d2789d --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/Factory/IDataProcessorFactory.cs @@ -0,0 +1,16 @@ +using DataAggregator.Processor.Services.Processing.Abstraction; + +namespace DataAggregator.Processor.Services.Processing.Factory; + +/// +/// Factory interface for creating processing strategies based on strategy name. +/// +public interface IDataProcessorFactory +{ + /// + /// Create multiple data processor which correspond to the pipeline configuration. + /// + /// The pipeline configuration. + /// Configured pipeline strategy. + public List CreateProcessors(IEnumerable pipeline); +} diff --git a/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescription.cs b/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescription.cs new file mode 100644 index 0000000..e61c337 --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescription.cs @@ -0,0 +1,21 @@ +using System.Text.Json.Serialization; +using DataAggregator.Processor.Services.Processing.Abstraction; + +namespace DataAggregator.Processor.Services.Processing.Factory; + +/// +/// Describes a processor with its name and configuration. +/// +[JsonConverter(typeof(ProcessorDescriptionJsonConverter))] +public class ProcessorDescription +{ + /// + /// Gets or sets the name of the processor. + /// + public string Name { get; set; } = string.Empty; + + /// + /// Gets or sets the configuration for the processor. + /// + public IProcessorConfiguration? Configuration { get; set; } +} diff --git a/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs b/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs new file mode 100644 index 0000000..a54c262 --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs @@ -0,0 +1,44 @@ +using System.Text.Json; +using System.Text.Json.Serialization; +using DataAggregator.Processor.Services.Processing.Abstraction; +using DataAggregator.Processor.Services.Processing.Onnx; +using DataAggregator.Processor.Services.Processing.PostProcessing.StateDeductionPostProcess; +using DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; + +namespace DataAggregator.Processor.Services.Processing.Factory; + +/// +/// Coverts to and from JSON. +/// +public class ProcessorDescriptionJsonConverter : JsonConverter +{ + /// + public override ProcessorDescription Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) + { + using var jsonDoc = JsonDocument.ParseValue(ref reader); + JsonElement root = jsonDoc.RootElement; + string name = root.GetProperty("Name").GetString() ?? string.Empty; + JsonElement configElement = root.GetProperty("Configuration"); + IProcessorConfiguration? config = name.ToLowerInvariant() switch + { + "actuatorcurrent" => configElement.Deserialize(options), + "onnxprediction" => configElement.Deserialize(options), + "statedeductionpostprocessor" => configElement.Deserialize(options), + _ => null, + }; + return new ProcessorDescription { Name = name, Configuration = config }; + } + + /// + public override void Write(Utf8JsonWriter writer, ProcessorDescription value, JsonSerializerOptions options) + { + writer.WriteStartObject(); + writer.WriteString("Name", value.Name); + writer.WritePropertyName("Configuration"); + if (value.Configuration != null) + JsonSerializer.Serialize(writer, value.Configuration, value.Configuration.GetType(), options); + else + writer.WriteNullValue(); + writer.WriteEndObject(); + } +} diff --git a/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionConfig.cs b/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionConfig.cs new file mode 100644 index 0000000..d8c1c09 --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionConfig.cs @@ -0,0 +1,14 @@ +using DataAggregator.Processor.Services.Processing.Abstraction; + +namespace DataAggregator.Processor.Services.Processing.Onnx; + +/// +/// Configuration for ONNX prediction. +/// +public class OnnxPredictionConfig : IProcessorConfiguration +{ + /// + /// Gets or sets the path of the model. + /// + public string ModelPath { get; set; } = string.Empty; +} diff --git a/src/DataAggregator.Processor/Services/Prediction/OnnxPredictionEngine.cs b/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs similarity index 88% rename from src/DataAggregator.Processor/Services/Prediction/OnnxPredictionEngine.cs rename to src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs index d19e5e7..ddb36b1 100644 --- a/src/DataAggregator.Processor/Services/Prediction/OnnxPredictionEngine.cs +++ b/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs @@ -1,4 +1,6 @@ using DataAggregator.Collector.Shared.Models; +using DataAggregator.Processor.Services.Processing.Abstraction; +using DataAggregator.Processor.Services.Processing.Onnx; using Microsoft.ML.OnnxRuntime; using Microsoft.ML.OnnxRuntime.Tensors; using Serilog; @@ -8,37 +10,38 @@ namespace DataAggregator.Processor.Services.Prediction; /// /// Implementation of ONNX prediction engine. /// -public class OnnxPredictionEngine : IOnnxPredictionEngine, IDisposable +public class OnnxPredictionEngine(OnnxPredictionConfig config) : IDataProcessor, IDisposable { #region Private fields - private readonly Dictionary _modelCache = []; + private static readonly Dictionary _modelCache = []; #endregion #region Public methods /// - public async Task> PredictAsync( - string modelPath, - IEnumerable inputData) + public async Task> ProcessAsync( + IEnumerable input) { try { // Load the model or get the existing session + string executablePath = AppContext.BaseDirectory; + string modelPath = Path.Combine(executablePath, config.ModelPath); InferenceSession session = LoadOrGetModel(modelPath); - if (!inputData.Any()) + if (!input.Any()) { - Log.Information($"No inputs data provided to model for model {modelPath}"); + Log.Information($"No inputs data provided to model for model {config.ModelPath}"); return Array.Empty(); } - ValidateInputData(session, inputData); + ValidateInputData(session, input); // Prepare the inputs for the inference session var inputs = new List(); - foreach (IMeasurementData data in inputData) + foreach (IMeasurementData data in input) { inputs.Add(CreateNamedOnnxValue(data)); } @@ -59,7 +62,7 @@ public async Task> PredictAsync( // Prepare the list to store output measurements var outputMeasurements = new List(); - DateTime processedDataTime = inputData.First().TimeStamp; + DateTime processedDataTime = input.First().TimeStamp; // Process each filtered result to convert it into measurement data foreach (DisposableNamedOnnxValue? result in filteredResults) @@ -69,13 +72,12 @@ public async Task> PredictAsync( outputMeasurements.AddRange(measurementData); } - // Log the completion of the prediction - Log.Debug("Prediction completed for model {ModelPath} with {OutputCount} outputs", modelPath, outputMeasurements.Count); + Log.Debug("Prediction completed for model {ModelPath} with {OutputCount} outputs", config.ModelPath, outputMeasurements.Count); return await Task.FromResult(outputMeasurements); } catch (Exception ex) { - Log.Error(ex, "Error during prediction with model {ModelPath}", modelPath); + Log.Error(ex, "Error during prediction with model {ModelPath}", config.ModelPath); throw; } } diff --git a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs new file mode 100644 index 0000000..2a2d67e --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs @@ -0,0 +1,104 @@ +using DataAggregator.Collector.Shared.Models; +using DataAggregator.Processor.Services.Processing.Abstraction; + +namespace DataAggregator.Processor.Services.Processing.PostProcessing.StateDeductionPostProcess; + +/// +/// Post processor for state deduction which avoids bad transitions recognition +/// by not allowing state change if the change duration is less than a threshold. +/// +public class StateDeductionPostProcessor(StateDeductionPostProcessorConfig config) : IDataProcessor +{ + #region Public methods + + /// + public Task> ProcessAsync(IEnumerable input) + { + var inputList = input.ToList(); + IMeasurementData? prediction = inputList.FirstOrDefault(x => x.SensorName == _resultOutputName); + string currentPredictedState = prediction?.GetRawValue() as string ?? string.Empty; + + if (_lastState == null) + return InitializeFirstState(inputList, currentPredictedState); + + if (currentPredictedState == _lastState) + return Task.FromResult>(inputList); + + if (currentPredictedState == _pendingState) + { + _stableCount++; + _bufferedBatches.Add(inputList); + + if (_stableCount == config.Threshold) + return ConfirmPendingState(); + + return Task.FromResult(Enumerable.Empty()); + } + else + { + IEnumerable flushed = FlushInvalidTransitionAsLastState(); + + _pendingState = currentPredictedState; + _stableCount = 1; + _bufferedBatches.Clear(); + _bufferedBatches.Add(inputList); + + return Task.FromResult(flushed); + } + } + + #endregion + + #region Private methods + + private Task> InitializeFirstState(List inputList, string state) + { + _lastState = state; + _pendingState = state; + return Task.FromResult>(inputList); + } + + private Task> ConfirmPendingState() + { + _lastState = _pendingState; + + var confirmed = _bufferedBatches + .SelectMany(batch => OverrideStateInBatch(batch, _lastState!)) + .ToList(); + + _bufferedBatches.Clear(); + return Task.FromResult>(confirmed); + } + + private IEnumerable FlushInvalidTransitionAsLastState() + { + if (_bufferedBatches.Count == 0 || _lastState == null) + return Enumerable.Empty(); + + var corrected = _bufferedBatches + .SelectMany(batch => OverrideStateInBatch(batch, _lastState)) + .ToList(); + + _bufferedBatches.Clear(); + return corrected; + } + + private IEnumerable OverrideStateInBatch(List batch, string forcedValue) + => batch.Select( + x => x.SensorName == _resultOutputName + ? new MeasurementData(x.TimeStamp, x.SensorName, forcedValue) + : x); + + #endregion + + #region Private fields + + private readonly string _resultOutputName = "PredictedLabel.output_0"; + + private readonly List> _bufferedBatches = []; + private string? _lastState = null; + private string? _pendingState = null; + private int _stableCount = 0; + + #endregion +} diff --git a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessorConfig.cs b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessorConfig.cs new file mode 100644 index 0000000..3473550 --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessorConfig.cs @@ -0,0 +1,14 @@ +using DataAggregator.Processor.Services.Processing.Abstraction; + +namespace DataAggregator.Processor.Services.Processing.PostProcessing.StateDeductionPostProcess; + +/// +/// Configuration for . +/// +public class StateDeductionPostProcessorConfig : IProcessorConfiguration +{ + /// + /// Gets or sets the threshold for minimum number of consecutive cycles required to deduce a state change. + /// + public int Threshold { get; set; } +} diff --git a/src/DataAggregator.Processor/Services/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs similarity index 83% rename from src/DataAggregator.Processor/Services/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs rename to src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs index b47f9a0..25fcc24 100644 --- a/src/DataAggregator.Processor/Services/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs @@ -1,40 +1,38 @@ using DataAggregator.Collector.Shared.Models; -using DataAggregator.Processor.Configuration; +using DataAggregator.Processor.Services.Processing.Abstraction; using Serilog; -namespace DataAggregator.Processor.Services.PreProcessing.ActuatorMergingCurrentPreprocessing; +namespace DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; /// /// Feature extractor for actuator current data based on ML.NET approach. /// Extracts 14 agnostic features with Z-score normalization. /// -public class ActuatorCurrentFeatureExtractor : IPreprocessingStrategy +/// +/// Initializes a new instance of the class. +/// +/// The configuration of the processor. +public class ActuatorCurrentFeatureExtractor(PreprocessingConfig config) : IDataProcessor { #region Public methods /// /// Preprocesses actuator current measurements into a feature vector. /// - /// List of raw measurements from the data window. - /// Configuration for the machine prediction. + /// List of raw measurements from the data window. /// Feature vector as dictionary mapping feature names to values for a single sample. - public IEnumerable PreprocessAsync(IEnumerable measurements, MachinePredictionConfig config) + public Task> ProcessAsync(IEnumerable input) { - Log.Debug( - "Preprocessing {Count} measurements for machine {MachineName}", - measurements.Count(), - config.MachineName); - // Extract the 14 features from measurements - float[] features = ExtractFeatures(measurements, config.InputSensors); + float[] features = ExtractFeatures(input); // Apply Z-score normalization if enabled - float[] normalizedFeatures = NormalizeFeaturesAsync(features, config.Preprocessing); + float[] normalizedFeatures = NormalizeFeaturesAsync(features, config); DateTime meanTime; - if (measurements.Count() != 0) - meanTime = measurements.ElementAt(measurements.Count() / 2).TimeStamp; + if (input.Any()) + meanTime = input.Min(x => x.TimeStamp); else meanTime = DateTime.UtcNow; @@ -57,46 +55,37 @@ public IEnumerable PreprocessAsync(IEnumerable(meanTime, "Label", string.Empty), }; - Log.Debug("Preprocessing completed for machine {MachineName}", config.MachineName); - return result; + return Task.FromResult(result.AsEnumerable()); } #endregion #region Private methods - private float[] ExtractFeatures(IEnumerable measurements, List sensors) + private float[] ExtractFeatures(IEnumerable measurements) { - if (measurements == null || sensors == null || sensors.Count == 0) + if (measurements == null) { - Log.Warning("No measurements or sensors provided for feature extraction."); + Log.Warning("No measurements provided for feature extraction."); return new float[14]; } - Log.Debug("Extracting features for {SensorCount} sensors", sensors.Count); - - // Concatenate all currents from all actuators (like in the notebook) + // Concatenate all currents from all actuators var allCurrents = new List(); foreach (IMeasurementData measurement in measurements) { - foreach (string sensor in sensors) + object value = measurement.GetRawValue(); + if (value is float floatValue) { - if (measurement.SensorName == sensor) - { - object value = measurement.GetRawValue(); - if (value is float floatValue) - { - allCurrents.Add(floatValue); - } - else if (value is double doubleValue) - { - allCurrents.Add((float)doubleValue); - } - else if (value is int intValue) - { - allCurrents.Add(intValue); - } - } + allCurrents.Add(floatValue); + } + else if (value is double doubleValue) + { + allCurrents.Add((float)doubleValue); + } + else if (value is int intValue) + { + allCurrents.Add(intValue); } } @@ -127,7 +116,7 @@ private float[] ExtractFeatures(IEnumerable measurements, List float changeDensity = significantChanges / (float)allCurrents.Count; // Extract actuator currents for correlation analysis - List> actuatorsCurrents = ExtractActuatorCurrents(measurements, sensors); + List> actuatorsCurrents = ExtractActuatorCurrents(measurements); // Features 3-5: Inter-actuator correlations List correlations = CalculateInterActuatorCorrelations(actuatorsCurrents); @@ -179,7 +168,7 @@ private float[] ExtractFeatures(IEnumerable measurements, List ]; } - private List> ExtractActuatorCurrents(IEnumerable measurements, List sensors) + private List> ExtractActuatorCurrents(IEnumerable measurements) { var actuatorsCurrents = new List>(); @@ -188,6 +177,8 @@ private List> ExtractActuatorCurrents(IEnumerable .GroupBy(m => m.SensorName) .ToDictionary(g => g.Key, g => g.ToList()); + var sensors = measurementsBySensor.Keys.ToList(); + // Extract currents for each sensor/Actuator foreach (string sensor in sensors) { diff --git a/src/DataAggregator.Processor/Services/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs similarity index 97% rename from src/DataAggregator.Processor/Services/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs rename to src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs index 8663cbc..cbe9ce8 100644 --- a/src/DataAggregator.Processor/Services/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs @@ -1,4 +1,4 @@ -namespace DataAggregator.Processor.Services.PreProcessing.ActuatorMergingCurrentPreprocessing; +namespace DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; /// /// Utility class for mathematical operations used in feature extraction. diff --git a/src/DataAggregator.Processor/Configuration/PreprocessingConfig.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs similarity index 69% rename from src/DataAggregator.Processor/Configuration/PreprocessingConfig.cs rename to src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs index 3b1a9ce..bbec98f 100644 --- a/src/DataAggregator.Processor/Configuration/PreprocessingConfig.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs @@ -1,9 +1,11 @@ -namespace DataAggregator.Processor.Configuration; +using DataAggregator.Processor.Services.Processing.Abstraction; + +namespace DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; /// /// Configuration for preprocessing operations including Z-score normalization. /// -public class PreprocessingConfig +public class PreprocessingConfig : IProcessorConfiguration { /// /// Gets or sets a value indicating whether Z-score normalization is enabled. diff --git a/src/DataAggregator.Processor/appsettings.json b/src/DataAggregator.Processor/appsettings.json index 3f0cc8d..d08c2b4 100644 --- a/src/DataAggregator.Processor/appsettings.json +++ b/src/DataAggregator.Processor/appsettings.json @@ -24,8 +24,6 @@ { "MachineName": "Micro5", "Enabled": true, - "ModelPath": "resources/opencn_model.onnx", - "PreprocessingStrategy": "actuatorcurrent", "InputSensors": [ "current-amp-x", "current-amp-y", @@ -34,27 +32,45 @@ "current-amp-c", "current-amp-s" ], - "WindowSizeSeconds": 2, - "CycleIntervalSeconds": 2, - "Preprocessing": { - "EnableZScoreNormalization": true, - "NormalizationParameters": { - "GlobalActivityRatio": [ 0.004157, 0.017644 ], - "GlobalChangeDensity": [ 0.432407, 0.143321 ], - "InterAxisMeanCorrelation": [ -0.001363, 0.072809 ], - "InterAxisMaxCorrelation": [ 0.439072, 0.267887 ], - "InterAxisCorrelationVariance": [ 0.225831, 0.129697 ], - "AxisSynchronization": [ -38.172882, 221.177505 ], - "AxisLoadBalance": [ -0.045936, 0.327285 ], - "TemporalStability": [ 0.974642, 0.046761 ], - "GlobalSkewness": [ -0.129752, 0.362594 ], - "GlobalKurtosis": [ -0.570948, 0.396184 ], - "GlobalTrendSlope": [ -0.000001, 0.000179 ], - "CoefficientOfVariation": [ -17.267527, 234.962006 ], - "NormalizedIqrMedian": [ 0.633865, 107.640205 ], - "NormalizedIqrMean": [ -14.154306, 267.521179 ] + "WindowSize": 30, + "WindowSizeInSeconds": false, + "CycleIntervalSeconds": 1, + "ProcessingPipeline": [ + { + "Name": "actuatorcurrent", + "Configuration": { + "EnableZScoreNormalization": true, + "NormalizationParameters": { + "GlobalActivityRatio": [ 0.003958, 0.018749 ], + "GlobalChangeDensity": [ 0.432119, 0.146720 ], + "InterAxisMeanCorrelation": [ -0.006328, 0.069353 ], + "InterAxisMaxCorrelation": [ 0.455234, 0.230869 ], + "InterAxisCorrelationVariance": [ 0.238157, 0.104391 ], + "AxisSynchronization": [ -38.056988, 293.112457 ], + "AxisLoadBalance": [ -0.045349, 0.330992 ], + "TemporalStability": [ 0.952670, 0.037949 ], + "GlobalSkewness": [ -0.122091, 0.380314 ], + "GlobalKurtosis": [ -0.562939, 0.403591 ], + "GlobalTrendSlope": [ 0.000049, 0.000477 ], + "CoefficientOfVariation": [ -14.905608, 295.991791 ], + "NormalizedIqrMedian": [ -26.388798, 1187.886230 ], + "NormalizedIqrMean": [ -10.343025, 301.199188 ] + } + } + }, + { + "Name": "onnxprediction", + "Configuration": { + "ModelPath": "resources/opencn_model.onnx" + } + }, + { + "Name": "statedeductionpostprocessor", + "Configuration": { + "Threshold": 3 + } } - } + ] } ] } diff --git a/tests/DataAggregator.Integration.Tests/DataAggregator.Integration.Tests.csproj b/tests/DataAggregator.Integration.Tests/DataAggregator.Integration.Tests.csproj deleted file mode 100644 index 1c5112b..0000000 --- a/tests/DataAggregator.Integration.Tests/DataAggregator.Integration.Tests.csproj +++ /dev/null @@ -1,30 +0,0 @@ - - - - net9.0 - enable - enable - false - true - - - - - - - all - runtime; build; native; contentfiles; analyzers; buildtransitive - - - - - - - - - - - - - - diff --git a/tests/DataAggregator.Integration.Tests/UnitTest1.cs b/tests/DataAggregator.Integration.Tests/UnitTest1.cs deleted file mode 100644 index 562d33d..0000000 --- a/tests/DataAggregator.Integration.Tests/UnitTest1.cs +++ /dev/null @@ -1,15 +0,0 @@ -namespace DataAggregator.Integration.Tests; - -/// -/// -Tests for the DataAggregator.Integration project. -/// -public class UnitTest1 -{ - /// - /// Tests the integration functionality of the DataAggregator project. - /// - [Fact] - public void Test1() - { - } -} diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs index 5671bbc..ceb8a75 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs @@ -1,143 +1,83 @@ using DataAggregator.Collector.Shared.Models; -using DataAggregator.Processor.Configuration; -using DataAggregator.Processor.Services.PreProcessing.ActuatorMergingCurrentPreprocessing; +using DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; namespace DataAggregator.Processor.Tests.Services.PreProcessing; -/// -/// Tests for the class. -/// public class ActuatorCurrentFeatureExtractorTests { private readonly ActuatorCurrentFeatureExtractor _featureExtractor; - /// - /// Initializes a new instance of the class. - /// public ActuatorCurrentFeatureExtractorTests() - => _featureExtractor = new ActuatorCurrentFeatureExtractor(); - - #region PreprocessAsync tests + => _featureExtractor = new ActuatorCurrentFeatureExtractor(CreateValidPreprocessingConfig()); [Fact] - public void PreprocessAsync_ShouldReturnFourteenFeatures_WhenValidDataProvided() + public async Task ProcessAsync_ShouldReturnFifteenFeatures_WhenValidDataProvided() { - // Arrange - List measurements = CreateTestMeasurements(); - MachinePredictionConfig config = CreateValidConfig(); - - // Act - var result = _featureExtractor.PreprocessAsync(measurements, config); - - // Assert + var measurements = CreateTestMeasurements(); + var result = await _featureExtractor.ProcessAsync(measurements); Assert.NotNull(result); Assert.Equal(15, result.Count()); } [Fact] - public void PreprocessAsync_ShouldReturnFourteenFeatures_WhenEmptyMeasurementsProvided() + public async Task ProcessAsync_ShouldReturnFourteenFeatures_WhenEmptyMeasurementsProvided() { - // Arrange var measurements = new List(); - MachinePredictionConfig config = CreateValidConfig(); - - // Act - var result = _featureExtractor.PreprocessAsync(measurements, config); + var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); - - // Assert - Assert.NotNull(result); - Assert.Equal(14, result.Count()); - Assert.All(result, feature => Assert.Equal(0.0f, (float)feature.GetRawValue())); - } - - [Fact] - public void PreprocessAsync_ShouldReturnFourteenFeatures_WhenEmptySensorsListProvided() - { - // Arrange - List measurements = CreateTestMeasurements(); - MachinePredictionConfig config = CreateValidConfig(); - config.InputSensors.Clear(); - - // Act - var result = _featureExtractor.PreprocessAsync(measurements, config); - result = result.Where(f => f.SensorName != "Label"); - - // Assert Assert.NotNull(result); Assert.Equal(14, result.Count()); Assert.All(result, feature => Assert.Equal(0.0f, (float)feature.GetRawValue())); } + [Fact] - public void PreprocessAsync_ShouldReturnValidFeatures_WhenValidDataProvided() + public async Task ProcessAsync_ShouldReturnValidFeatures_WhenValidDataProvided() { - // Arrange - List measurements = CreateTestMeasurements(); - MachinePredictionConfig config = CreateValidConfig(); - - // Act - var result = _featureExtractor.PreprocessAsync(measurements, config); + var measurements = CreateTestMeasurements(); + var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); - - // Assert Assert.NotNull(result); Assert.Equal(14, result.Count()); - - // Check that features are within reasonable bounds Assert.All(result, feature => Assert.False(float.IsNaN((float)feature.GetRawValue()))); Assert.All(result, feature => Assert.False(float.IsInfinity((float)feature.GetRawValue()))); } [Fact] - public void PreprocessAsync_ShouldReturnZeroFeatures_WhenNoValidValuesFound() + public async Task ProcessAsync_ShouldReturnZeroFeatures_WhenNoValidValuesFound() { - // Arrange var measurements = new List { new MeasurementData(DateTime.UtcNow, "sensor1", float.NaN), new MeasurementData(DateTime.UtcNow, "sensor2", float.PositiveInfinity), new MeasurementData(DateTime.UtcNow, "sensor1", float.NegativeInfinity), }; - MachinePredictionConfig config = CreateValidConfig(); - - // Act - var result = _featureExtractor.PreprocessAsync(measurements, config); + var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); - - // Assert Assert.NotNull(result); Assert.Equal(14, result.Count()); Assert.All(result, feature => Assert.Equal(0.0f, (float)feature.GetRawValue())); } [Fact] - public void PreprocessAsync_ShouldHandleSingleValue_WhenOnlyOneValidMeasurementProvided() + public async Task ProcessAsync_ShouldHandleSingleValue_WhenOnlyOneValidMeasurementProvided() { - // Arrange var measurements = new List { new MeasurementData(DateTime.UtcNow, "sensor1", 10.5f), }; - MachinePredictionConfig config = CreateValidConfig(); - - // Act - var result = _featureExtractor.PreprocessAsync(measurements, config); + var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); - - // Assert Assert.NotNull(result); Assert.Equal(14, result.Count()); Assert.All(result, feature => Assert.False(float.IsNaN((float)feature.GetRawValue()))); } [Fact] - public void PreprocessAsync_ShouldHandleLargeDataset_WhenManyMeasurementsProvided() + public async Task ProcessAsync_ShouldHandleLargeDataset_WhenManyMeasurementsProvided() { - // Arrange var measurements = new List(); var random = new Random(42); - for (int i = 0; i < 1000; i++) { measurements.Add(new MeasurementData( @@ -149,24 +89,14 @@ public void PreprocessAsync_ShouldHandleLargeDataset_WhenManyMeasurementsProvide "sensor2", (float)random.NextDouble() * 100)); } - - MachinePredictionConfig config = CreateValidConfig(); - - // Act - var result = _featureExtractor.PreprocessAsync(measurements, config); + var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); - - // Assert Assert.NotNull(result); Assert.Equal(14, result.Count()); Assert.All(result, feature => Assert.False(float.IsNaN((float)feature.GetRawValue()))); Assert.All(result, feature => Assert.False(float.IsInfinity((float)feature.GetRawValue()))); } - #endregion - - #region Helper methods - private static List CreateTestMeasurements() => [ new MeasurementData(DateTime.UtcNow, "sensor1", 10.5f), new MeasurementData(DateTime.UtcNow, "sensor2", 20.3f), @@ -176,20 +106,9 @@ private static List CreateTestMeasurements() => [ new MeasurementData(DateTime.UtcNow, "sensor2", 22.5f) ]; - private static MachinePredictionConfig CreateValidConfig() => new() + private static PreprocessingConfig CreateValidPreprocessingConfig() => new() { - MachineName = "test_machine", - ModelPath = "test_model.onnx", - InputSensors = ["sensor1", "sensor2"], - PreprocessingStrategy = "ActuatorMergingCurrent", - WindowSizeSeconds = 1, - CycleIntervalSeconds = 1, - Enabled = true, - Preprocessing = new PreprocessingConfig - { - EnableZScoreNormalization = true, - }, + EnableZScoreNormalization = true, + NormalizationParameters = new Dictionary() }; - - #endregion } diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs new file mode 100644 index 0000000..e4cbe7b --- /dev/null +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs @@ -0,0 +1,98 @@ +using DataAggregator.Processor.Configuration; +using DataAggregator.Processor.Services.Processing.Factory; +using DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; +using DataAggregator.Processor.Services.Prediction; +using DataAggregator.Processor.Services.Processing.PostProcessing.StateDeductionPostProcess; +using System.Collections.Generic; +using DataAggregator.Processor.Services.Processing.Onnx; + +namespace DataAggregator.Processor.Tests.Services.PreProcessing; + +public class DataProcessorFactoryTests +{ + private readonly DataProcessorFactory _factory; + + public DataProcessorFactoryTests() => _factory = new DataProcessorFactory(); + + [Fact] + public void CreateProcessors_ShouldReturnCorrectProcessors_ForValidPipeline() + { + // Arrange + var pipeline = new List + { + new ProcessorDescription + { + Name = "actuatorcurrent", + Configuration = new PreprocessingConfig + { + EnableZScoreNormalization = true, + NormalizationParameters = new Dictionary() + } + }, + new ProcessorDescription + { + Name = "onnxprediction", + Configuration = new OnnxPredictionConfig + { + ModelPath = "model.onnx" + } + }, + new ProcessorDescription + { + Name = "statedeductionpostprocessor", + Configuration = new StateDeductionPostProcessorConfig + { + Threshold = 2 + } + } + }; + + // Act + var processors = _factory.CreateProcessors(pipeline); + + // Assert + Assert.Equal(3, processors.Count); + Assert.IsType(processors[0]); + Assert.IsType(processors[1]); + Assert.IsType(processors[2]); + } + + [Fact] + public void CreateProcessors_ShouldThrowArgumentException_ForUnknownName() + { + // Arrange + var pipeline = new List + { + new ProcessorDescription { Name = "unknownstrategy", Configuration = null } + }; + + // Act & Assert + Assert.Throws(() => _factory.CreateProcessors(pipeline)); + } + + [Fact] + public void CreateProcessors_ShouldReturnEmptyList_ForEmptyPipeline() + { + // Arrange + var pipeline = new List(); + + // Act + var processors = _factory.CreateProcessors(pipeline); + + // Assert + Assert.Empty(processors); + } + + [Fact] + public void CreateProcessors_ShouldThrowArgumentException_WhenNameMissing() + { + // Arrange + var pipeline = new List + { + new ProcessorDescription { Name = "", Configuration = new PreprocessingConfig() } + }; + + // Act & Assert + Assert.Throws(() => _factory.CreateProcessors(pipeline)); + } +} diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/MathUtilsTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/MathUtilsTests.cs index 53b30cd..6d86095 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/MathUtilsTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/MathUtilsTests.cs @@ -1,4 +1,4 @@ -using DataAggregator.Processor.Services.PreProcessing.ActuatorMergingCurrentPreprocessing; +using DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; namespace DataAggregator.Processor.Tests.Services.PreProcessing; diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/PreprocessingStrategyFactoryTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/PreprocessingStrategyFactoryTests.cs deleted file mode 100644 index 9f4e41b..0000000 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/PreprocessingStrategyFactoryTests.cs +++ /dev/null @@ -1,48 +0,0 @@ -using DataAggregator.Processor.Services.PreProcessing; -using DataAggregator.Processor.Services.PreProcessing.ActuatorMergingCurrentPreprocessing; - -namespace DataAggregator.Processor.Tests.Services.PreProcessing; - -/// -/// Tests for the class. -/// -public class PreprocessingStrategyFactoryTests -{ - private readonly PreprocessingStrategyFactory _factory; - - /// - /// Initializes a new instance of the class. - /// - public PreprocessingStrategyFactoryTests() => _factory = new PreprocessingStrategyFactory(); - - #region CreateStrategy tests - - [Fact] - public void CreateStrategy_ShouldReturnGoodStrategy() - { - string strategyName = "actuatorcurrent"; - - IPreprocessingStrategy strategy = _factory.CreateStrategy(strategyName); - - Assert.NotNull(strategy); - Assert.IsType(strategy); - } - - [Fact] - public void CreateStrategy_ShouldThrowArgumentException_WhenStrategyNameIsEmpty() - { - string strategyName = string.Empty; - - ArgumentException exception = Assert.Throws(() => _factory.CreateStrategy(strategyName)); - } - - [Fact] - public void CreateStrategy_ShouldThrowArgumentException_WhenStrategyNameIsUnknown() - { - string strategyName = "UnknownStrategy"; - - ArgumentException exception = Assert.Throws(() => _factory.CreateStrategy(strategyName)); - } - - #endregion -} diff --git a/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs b/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs index b8aa884..6721c30 100644 --- a/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs @@ -2,7 +2,8 @@ using DataAggregator.Processor.Configuration; using DataAggregator.Processor.Services.DataStorage; using DataAggregator.Processor.Services.Prediction; -using DataAggregator.Processor.Services.PreProcessing; +using DataAggregator.Processor.Services.Processing.Abstraction; +using DataAggregator.Processor.Services.Processing.Factory; using DataAggregator.Processor.Services.Registration; using DataAggregator.Shared.Configuration.TimeSeries; using DataAggregator.Shared.Domain.DataType; @@ -11,205 +12,117 @@ namespace DataAggregator.Processor.Tests.Services.Prediction; -/// -/// Tests for the class. -/// public class MachinePredictionProcessorTests { private readonly Mock _mockInfluxRepository; private readonly Mock _mockRegistrationClient; - private readonly Mock _mockPredictionEngine; - private readonly Mock _mockStrategyFactory; - private readonly Mock _mockPreprocessingStrategy; + private readonly Mock _mockProcessorFactory; private readonly MachinePredictionProcessor _processor; + private readonly Mock _mockProcessor; - /// - /// Initializes a new instance of the class. - /// public MachinePredictionProcessorTests() { _mockInfluxRepository = new Mock(); _mockRegistrationClient = new Mock(); - _mockPredictionEngine = new Mock(); - _mockStrategyFactory = new Mock(); - _mockPreprocessingStrategy = new Mock(); + _mockProcessorFactory = new Mock(); + _mockProcessor = new Mock(); _processor = new MachinePredictionProcessor( _mockInfluxRepository.Object, _mockRegistrationClient.Object, - _mockPredictionEngine.Object, - _mockStrategyFactory.Object); + _mockProcessorFactory.Object); } - #region ProcessAsync tests - [Fact] public async Task ProcessAsync_ShouldReturnEarly_WhenCollectorInfoIsNull() { - // Arrange - MachinePredictionConfig config = CreateValidMachineConfig(); + var config = CreateValidMachineConfig(); _mockRegistrationClient.Setup(x => x.GetCollectorInfoAsync(config.MachineName)) .ReturnsAsync((CollectorInfoDto?)null); - // Act await _processor.ProcessAsync(config); - // Assert _mockInfluxRepository.Verify(x => x.QueryMeasurementsAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>()), Times.Never); - _mockPredictionEngine.Verify(x => x.PredictAsync(It.IsAny(), It.IsAny>()), Times.Never); } [Fact] public async Task ProcessAsync_ShouldReturnEarly_WhenNoValidSensorsFound() { - // Arrange - MachinePredictionConfig config = CreateValidMachineConfig(); - CollectorInfoDto collectorInfo = CreateCollectorInfoWithSensors(new[] { "different_sensor" }); + var config = CreateValidMachineConfig(); + var collectorInfo = CreateCollectorInfoWithSensors(new[] { "different_sensor" }); _mockRegistrationClient.Setup(x => x.GetCollectorInfoAsync(config.MachineName)) .ReturnsAsync(collectorInfo); - // Act await _processor.ProcessAsync(config); - // Assert _mockInfluxRepository.Verify(x => x.QueryMeasurementsAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>()), Times.Never); - _mockPredictionEngine.Verify(x => x.PredictAsync(It.IsAny(), It.IsAny>()), Times.Never); } [Fact] public async Task ProcessAsync_ShouldReturnEarly_WhenNoMeasurementsFound() { - // Arrange - MachinePredictionConfig config = CreateValidMachineConfig(); - CollectorInfoDto collectorInfo = CreateCollectorInfoWithSensors(config.InputSensors); + var config = CreateValidMachineConfig(); + var collectorInfo = CreateCollectorInfoWithSensors(config.InputSensors); _mockRegistrationClient.Setup(x => x.GetCollectorInfoAsync(config.MachineName)) .ReturnsAsync(collectorInfo); _mockInfluxRepository.Setup(x => x.QueryMeasurementsAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>())) .ReturnsAsync([]); - // Act - await _processor.ProcessAsync(config); - - // Assert - _mockPredictionEngine.Verify(x => x.PredictAsync(It.IsAny(), It.IsAny>()), Times.Never); - _mockInfluxRepository.Verify(x => x.WriteMeasurementAsync(It.IsAny(), It.IsAny>()), Times.Never); - } - - [Fact] - public async Task ProcessAsync_ShouldReturnEarly_WhenPreprocessingFails() - { - // Arrange - MachinePredictionConfig config = CreateValidMachineConfig(); - CollectorInfoDto collectorInfo = CreateCollectorInfoWithSensors(config.InputSensors); - List measurements = CreateTestMeasurements(); - - _mockRegistrationClient.Setup(x => x.GetCollectorInfoAsync(config.MachineName)) - .ReturnsAsync(collectorInfo); - _mockInfluxRepository.Setup(x => x.QueryMeasurementsAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>())) - .ReturnsAsync(measurements); - _mockStrategyFactory.Setup(x => x.CreateStrategy(config.PreprocessingStrategy)) - .Returns(_mockPreprocessingStrategy.Object); - _mockPreprocessingStrategy.Setup(x => x.PreprocessAsync(It.IsAny>(), It.IsAny())) - .Returns(new List()); - - // Act await _processor.ProcessAsync(config); - // Assert - _mockPredictionEngine.Verify(x => x.PredictAsync(It.IsAny(), It.IsAny>()), Times.Never); _mockInfluxRepository.Verify(x => x.WriteMeasurementAsync(It.IsAny(), It.IsAny>()), Times.Never); } [Fact] public async Task ProcessAsync_ShouldCompleteSuccessfully_WhenAllConditionsAreMet() { - // Arrange - MachinePredictionConfig config = CreateValidMachineConfig(); - CollectorInfoDto collectorInfo = CreateCollectorInfoWithSensors(config.InputSensors); - List measurements = CreateTestMeasurements(); - var preprocessedData = ProcessorTestHelper.GetValidTestData(); - - IEnumerable results = - new List - { - new MeasurementData(DateTime.Now, "Prediction", 0.85f) - }; - - float[] predictions = [0.85f]; + var config = CreateValidMachineConfig(); + var collectorInfo = CreateCollectorInfoWithSensors(config.InputSensors); + var measurements = CreateTestMeasurements(); + var processedData = new List { new MeasurementData(DateTime.Now, "Prediction", 0.85f) }; _mockRegistrationClient.Setup(x => x.GetCollectorInfoAsync(config.MachineName)) .ReturnsAsync(collectorInfo); _mockInfluxRepository.Setup(x => x.QueryMeasurementsAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>())) .ReturnsAsync(measurements); - _mockStrategyFactory.Setup(x => x.CreateStrategy(config.PreprocessingStrategy)) - .Returns(_mockPreprocessingStrategy.Object); - _mockPreprocessingStrategy.Setup(x => x.PreprocessAsync(It.IsAny>(), It.IsAny())) - .Returns(preprocessedData); - _mockPredictionEngine.Setup(x => x.PredictAsync(It.IsAny(), It.IsAny>())) - .ReturnsAsync(results); - - // Act + _mockProcessor.Setup(x => x.ProcessAsync(It.IsAny>())) + .ReturnsAsync(processedData); + _mockProcessorFactory.Setup(x => x.CreateProcessors(It.IsAny>())) + .Returns(new List { _mockProcessor.Object }); + await _processor.ProcessAsync(config); - // Assert - _mockInfluxRepository.Verify( - x => x.InitializeAsync( - collectorInfo.AssignedInfluxEndpoint.Endpoint, - collectorInfo.AssignedInfluxEndpoint.Token), - Times.Once); - _mockInfluxRepository.Verify( - x => x.QueryMeasurementsAsync( - config.MachineName, - It.IsAny(), - It.IsAny(), - It.IsAny>()), - Times.Once); - _mockPredictionEngine.Verify(x => x.PredictAsync(It.IsAny(), It.IsAny>()), Times.Once); - _mockInfluxRepository.Verify(x => x.WriteMeasurementAsync(config.MachineName, It.IsAny>()), Times.Once); + _mockInfluxRepository.Verify(x => x.WriteMeasurementAsync(config.MachineName, processedData), Times.Once); } [Fact] public async Task ProcessAsync_ShouldReinitializeInfluxConnection_WhenEndpointChanges() { - // Arrange - MachinePredictionConfig config = CreateValidMachineConfig(); - CollectorInfoDto collectorInfo1 = CreateCollectorInfoWithSensors(config.InputSensors, "endpoint1"); - CollectorInfoDto collectorInfo2 = CreateCollectorInfoWithSensors(config.InputSensors, "endpoint2"); - List measurements = CreateTestMeasurements(); - - var now = DateTime.UtcNow; - var preprocessedData = ProcessorTestHelper.GetValidTestData(); - - var predictions = new List - { - new MeasurementData(now, "Prediction", 0.85f) - }; + var config = CreateValidMachineConfig(); + var collectorInfo1 = CreateCollectorInfoWithSensors(config.InputSensors, "endpoint1"); + var collectorInfo2 = CreateCollectorInfoWithSensors(config.InputSensors, "endpoint2"); + var measurements = CreateTestMeasurements(); + var processedData = new List { new MeasurementData(DateTime.Now, "Prediction", 0.85f) }; _mockRegistrationClient.SetupSequence(x => x.GetCollectorInfoAsync(config.MachineName)) .ReturnsAsync(collectorInfo1) .ReturnsAsync(collectorInfo2); _mockInfluxRepository.Setup(x => x.QueryMeasurementsAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>())) .ReturnsAsync(measurements); - _mockStrategyFactory.Setup(x => x.CreateStrategy(config.PreprocessingStrategy)) - .Returns(_mockPreprocessingStrategy.Object); - _mockPreprocessingStrategy.Setup(x => x.PreprocessAsync(It.IsAny>(), It.IsAny())) - .Returns(preprocessedData); - _mockPredictionEngine.Setup(x => x.PredictAsync(config.ModelPath, preprocessedData)) - .ReturnsAsync(predictions); - - // Act + _mockProcessor.Setup(x => x.ProcessAsync(It.IsAny>())) + .ReturnsAsync(processedData); + _mockProcessorFactory.Setup(x => x.CreateProcessors(It.IsAny>())) + .Returns(new List { _mockProcessor.Object }); + await _processor.ProcessAsync(config); // First call with endpoint1 await _processor.ProcessAsync(config); // Second call with endpoint2 - // Assert - _mockInfluxRepository.Verify( - x => x.InitializeAsync( + _mockInfluxRepository.Verify(x => x.Initialize( collectorInfo1.AssignedInfluxEndpoint.Endpoint, collectorInfo1.AssignedInfluxEndpoint.Token), Times.Once); - _mockInfluxRepository.Verify( - x => x.InitializeAsync( + _mockInfluxRepository.Verify(x => x.Initialize( collectorInfo2.AssignedInfluxEndpoint.Endpoint, collectorInfo2.AssignedInfluxEndpoint.Token), Times.Once); } @@ -217,34 +130,24 @@ public async Task ProcessAsync_ShouldReinitializeInfluxConnection_WhenEndpointCh [Fact] public async Task ProcessAsync_ShouldNotReinitializeInfluxConnection_WhenEndpointIsSame() { - // Arrange - MachinePredictionConfig config = CreateValidMachineConfig(); - CollectorInfoDto collectorInfo = CreateCollectorInfoWithSensors(config.InputSensors); - List measurements = CreateTestMeasurements(); - - var now = DateTime.UtcNow; - var preprocessedData = ProcessorTestHelper.GetValidTestData(); - - var predictions = new List { new MeasurementData(now, "Prediction", 0.85f) }; + var config = CreateValidMachineConfig(); + var collectorInfo = CreateCollectorInfoWithSensors(config.InputSensors); + var measurements = CreateTestMeasurements(); + var processedData = new List { new MeasurementData(DateTime.Now, "Prediction", 0.85f) }; _mockRegistrationClient.Setup(x => x.GetCollectorInfoAsync(config.MachineName)) .ReturnsAsync(collectorInfo); _mockInfluxRepository.Setup(x => x.QueryMeasurementsAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>())) .ReturnsAsync(measurements); - _mockStrategyFactory.Setup(x => x.CreateStrategy(config.PreprocessingStrategy)) - .Returns(_mockPreprocessingStrategy.Object); - _mockPreprocessingStrategy.Setup(x => x.PreprocessAsync(It.IsAny>(), It.IsAny())) - .Returns(preprocessedData); - _mockPredictionEngine.Setup(x => x.PredictAsync(config.ModelPath, preprocessedData)) - .ReturnsAsync(predictions); - - // Act + _mockProcessor.Setup(x => x.ProcessAsync(It.IsAny>())) + .ReturnsAsync(processedData); + _mockProcessorFactory.Setup(x => x.CreateProcessors(It.IsAny>())) + .Returns(new List { _mockProcessor.Object }); + await _processor.ProcessAsync(config); await _processor.ProcessAsync(config); - // Assert - _mockInfluxRepository.Verify( - x => x.InitializeAsync( + _mockInfluxRepository.Verify(x => x.Initialize( It.IsAny(), It.IsAny()), Times.Once); } @@ -252,28 +155,21 @@ public async Task ProcessAsync_ShouldNotReinitializeInfluxConnection_WhenEndpoin [Fact] public async Task ProcessAsync_ShouldThrowException_WhenRegistrationClientThrows() { - // Arrange - MachinePredictionConfig config = CreateValidMachineConfig(); + var config = CreateValidMachineConfig(); _mockRegistrationClient.Setup(x => x.GetCollectorInfoAsync(config.MachineName)) .ThrowsAsync(new InvalidOperationException("Registration service error")); - // Act & Assert await Assert.ThrowsAsync(() => _processor.ProcessAsync(config)); } - #endregion - - #region Helper methods - private static MachinePredictionConfig CreateValidMachineConfig() => new() { MachineName = "test_machine", - ModelPath = "test_model.onnx", InputSensors = ["sensor1", "sensor2"], - PreprocessingStrategy = "test_strategy", - WindowSizeSeconds = 300, + WindowSize = 300, CycleIntervalSeconds = 60, Enabled = true, + ProcessingPipeline = new List { new() } }; private static CollectorInfoDto CreateCollectorInfoWithSensors(IEnumerable sensorNames, string endpoint = "http://localhost:8086") => new( @@ -290,6 +186,4 @@ private static List CreateTestMeasurements() new MeasurementData(DateTime.UtcNow, "sensor1", 10.5f), new MeasurementData(DateTime.UtcNow, "sensor2", 20.3f) ]; - - #endregion } diff --git a/tests/DataAggregator.Processor.Tests/Services/Prediction/OnnxPredictionEngineTests.cs b/tests/DataAggregator.Processor.Tests/Services/Prediction/OnnxPredictionEngineTests.cs index 5c1edd5..02733b4 100644 --- a/tests/DataAggregator.Processor.Tests/Services/Prediction/OnnxPredictionEngineTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/Prediction/OnnxPredictionEngineTests.cs @@ -1,7 +1,10 @@ using System.Reflection.Metadata; using DataAggregator.Collector.Shared.Models; +using DataAggregator.Processor.Configuration; using DataAggregator.Processor.Services.Prediction; using Microsoft.ML.OnnxRuntime; +using System.Threading.Tasks; +using DataAggregator.Processor.Services.Processing.Onnx; namespace DataAggregator.Processor.Tests.Services.Prediction; @@ -10,145 +13,83 @@ namespace DataAggregator.Processor.Tests.Services.Prediction; /// public class OnnxPredictionEngineTests : IDisposable { - private readonly OnnxPredictionEngine _predictionEngine; + private OnnxPredictionEngine _predictionEngine; - /// - /// Initializes a new instance of the class. - /// public OnnxPredictionEngineTests() - => _predictionEngine = new OnnxPredictionEngine(); - - #region PredictAsync tests + { + // L'instance sera créée dans chaque test avec la bonne config + } [Fact] - public async Task PredictAsync_ShouldThrowFileNotFoundException_WhenModelPathDoesNotExist() + public async Task ProcessAsync_ShouldThrowFileNotFoundException_WhenModelPathDoesNotExist() { - // Arrange - string nonExistentModelPath = "non_existent_model.onnx"; - + var config = new OnnxPredictionConfig { ModelPath = "non_existent_model.onnx" }; + _predictionEngine = new OnnxPredictionEngine(config); var inputData = new List { new MeasurementData(DateTime.UtcNow, "GlobalActivityRatio", 1.0f), new MeasurementData(DateTime.UtcNow, "GlobalChangeDensity", 2.0f), new MeasurementData(DateTime.UtcNow, "InterAxisMeanCorrelation", 3.0f), }; - - // Act & Assert - FileNotFoundException exception = await Assert.ThrowsAsync( - () => _predictionEngine.PredictAsync(nonExistentModelPath, inputData)); + await Assert.ThrowsAsync(() => _predictionEngine.ProcessAsync(inputData)); } [Fact] - public async Task PredictAsync_ShouldCacheModel_WhenSameModelPathIsUsedMultipleTimes() + public async Task ProcessAsync_ShouldReturnCorrectOutputShape_WhenValidInputProvided() { - // Arrange - string modelPath = _testModelPath; - string copyPath = Path.Combine("resources", "opencn_model_copy.onnx"); - File.Copy(modelPath, copyPath, true); - + var config = new OnnxPredictionConfig { ModelPath = _testModelPath }; + _predictionEngine = new OnnxPredictionEngine(config); var inputData = ProcessorTestHelper.GetValidTestData(); - - // Act - IEnumerable result1 = await _predictionEngine.PredictAsync(copyPath, inputData); - File.Delete(copyPath); // Simulate model file deletion - IEnumerable result2 = await _predictionEngine.PredictAsync(copyPath, inputData); - - // Assert - Assert.NotNull(result1); - Assert.NotNull(result2); - Assert.Equal(result1.Count(), result2.Count()); - } - - [Fact] - public async Task PredictAsync_ShouldReturnCorrectOutputShape_WhenValidInputProvided() - { - // Arrange - string modelPath = _testModelPath; - var now = DateTime.UtcNow; - var inputData = ProcessorTestHelper.GetValidTestData(); - - // Act - IEnumerable result = await _predictionEngine.PredictAsync(modelPath, inputData); - - // Assert + var result = await _predictionEngine.ProcessAsync(inputData); Assert.NotNull(result); Assert.True(result.Count() > 0); } [Fact] - public async Task PredictAsync_ShouldReturnGoodResult_DifferentStateDataProvided() + public async Task ProcessAsync_ShouldReturnGoodResult_DifferentStateDataProvided() { - // Arrange - string modelPath = _testModelPath; + var config = new OnnxPredictionConfig { ModelPath = _testModelPath }; + _predictionEngine = new OnnxPredictionEngine(config); var inputDataShutdown = ProcessorTestHelper.GetValidShutdownStateData(); var inputDataProduction = ProcessorTestHelper.GetValidProductionStateData(); var inputDataIdle = ProcessorTestHelper.GetValidIdleStateData(); - - // Act - IEnumerable resultShutdown = await _predictionEngine.PredictAsync(modelPath, inputDataShutdown); - IEnumerable resultProduction = await _predictionEngine.PredictAsync(modelPath, inputDataProduction); - IEnumerable resultIdle = await _predictionEngine.PredictAsync(modelPath, inputDataIdle); - - // Assert + var resultShutdown = await _predictionEngine.ProcessAsync(inputDataShutdown); + var resultProduction = await _predictionEngine.ProcessAsync(inputDataProduction); + var resultIdle = await _predictionEngine.ProcessAsync(inputDataIdle); Assert.NotNull(resultShutdown); Assert.NotNull(resultProduction); Assert.NotNull(resultIdle); Assert.Equal("disable", resultShutdown.First(x => x.SensorName == "PredictedLabel.output_0").GetRawValue().ToString()); Assert.Equal("production", resultProduction.First(x => x.SensorName == "PredictedLabel.output_0").GetRawValue().ToString()); - Assert.Equal("enable", resultIdle.First(static x => x.SensorName == "PredictedLabel.output_0").GetRawValue().ToString()); - + Assert.Equal("enable", resultIdle.First(x => x.SensorName == "PredictedLabel.output_0").GetRawValue().ToString()); } [Fact] - public async Task PredictAsync_ShouldHandleEmptyInputArray_WhenProvided() + public async Task ProcessAsync_ShouldHandleEmptyInputArray_WhenProvided() { - // Arrange - string modelPath = _testModelPath; + var config = new OnnxPredictionConfig { ModelPath = _testModelPath }; + _predictionEngine = new OnnxPredictionEngine(config); var inputData = new List(); - - // Act - IEnumerable result = await _predictionEngine.PredictAsync(modelPath, inputData); - - // Assert + var result = await _predictionEngine.ProcessAsync(inputData); Assert.NotNull(result); } - #endregion - - #region Dispose tests - [Fact] public void Dispose_ShouldClearModelCache_WhenCalled() { - // Arrange - string modelPath = _testModelPath; + var config = new OnnxPredictionConfig { ModelPath = _testModelPath }; + _predictionEngine = new OnnxPredictionEngine(config); IEnumerable inputData = ProcessorTestHelper.GetValidTestData(); - try { - // Act - Load model into cache - _ = _predictionEngine.PredictAsync(modelPath, inputData).Result; - - // Act - Dispose + _ = _predictionEngine.ProcessAsync(inputData).Result; _predictionEngine.Dispose(); - - // Assert - Should be able to dispose without exception - Assert.True(true); // If we reach here, no exception was thrown - } - finally - { - // do nothing + Assert.True(true); } + finally { } } - #endregion - - #region Helper methods - private static string _testModelPath = Path.Combine("resources", "opencn_model.onnx"); - #endregion - - /// public void Dispose() => _predictionEngine?.Dispose(); } diff --git a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs index 01cc656..01d35d7 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs @@ -1,8 +1,9 @@ using DataAggregator.Processor.Configuration; using DataAggregator.Processor.Services; using DataAggregator.Processor.Services.Prediction; +using DataAggregator.Processor.Services.Processing.Factory; +using DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Options; using Moq; namespace DataAggregator.Processor.Tests.Services; @@ -12,7 +13,6 @@ namespace DataAggregator.Processor.Tests.Services; /// public class PredictionBackgroundServiceTests : IDisposable { - private readonly Mock> _mockConfiguration; private readonly Mock _mockPredictionProcessor; private readonly Mock _serviceProvider; private readonly Mock _mockScope; @@ -22,7 +22,6 @@ public class PredictionBackgroundServiceTests : IDisposable public PredictionBackgroundServiceTests() { - _mockConfiguration = new Mock>(); _mockPredictionProcessor = new Mock(); _cancellationTokenSource = new CancellationTokenSource(); @@ -47,7 +46,7 @@ public PredictionBackgroundServiceTests() .Returns(_mockScope.Object); _backgroundService = new PredictionBackgroundService( - _mockConfiguration.Object, + CreateValidConfiguration(), _serviceProvider.Object); } @@ -56,33 +55,24 @@ public PredictionBackgroundServiceTests() [Fact] public async Task ExecuteAsync_ShouldStartSuccessfully_WhenValidConfigurationProvided() { - // Arrange - PredictionServiceConfiguration config = CreateValidConfiguration(); - _mockConfiguration.Setup(x => x.Value).Returns(config); + var config = CreateValidConfiguration(); - // Act Task task = _backgroundService.StartAsync(_cancellationTokenSource.Token); - await Task.Delay(100); // Give it time to start + await Task.Delay(100); await _backgroundService.StopAsync(_cancellationTokenSource.Token); - // Assert - Assert.True(true); // If we reach here, no exception was thrown + Assert.True(true); } [Fact] public async Task ExecuteAsync_ShouldScheduleEnabledMachines_WhenConfigurationContainsEnabledMachines() { - // Arrange - PredictionServiceConfiguration config = CreateValidConfiguration(); - _mockConfiguration.Setup(x => x.Value).Returns(config); + var config = CreateValidConfiguration(); - // Act Task task = _backgroundService.StartAsync(_cancellationTokenSource.Token); - await Task.Delay(100); // Give it time to start + await Task.Delay(100); await _backgroundService.StopAsync(_cancellationTokenSource.Token); - // Assert - // The service should have started without throwing exceptions _mockPredictionProcessor.Verify( x => x.ProcessAsync(It.IsAny()), Times.Exactly(config.Machines.Count(m => m.Enabled))); @@ -91,86 +81,28 @@ public async Task ExecuteAsync_ShouldScheduleEnabledMachines_WhenConfigurationCo [Fact] public async Task ExecuteAsync_ShouldHandleEmptyMachineList_WhenConfigurationContainsNoMachines() { - // Arrange var config = new PredictionServiceConfiguration { Machines = [], }; - _mockConfiguration.Setup(x => x.Value).Returns(config); - // Act Task task = _backgroundService.StartAsync(_cancellationTokenSource.Token); - await Task.Delay(100); // Give it time to start + await Task.Delay(100); await _backgroundService.StopAsync(_cancellationTokenSource.Token); - // Assert - // The service should have started without throwing exceptions Assert.True(true); } - [Fact] - public async Task ExecuteAsync_ShouldThrowFileNotFoundException_WhenModelFileDoesNotExist() - { - // Arrange - PredictionServiceConfiguration config = CreateValidConfiguration(); - config.Machines[0].ModelPath = "non_existent_model.onnx"; - _mockConfiguration.Setup(x => x.Value).Returns(config); - - // Act & Assert - await Assert.ThrowsAsync(() => _backgroundService.StartAsync(_cancellationTokenSource.Token)); - } - - [Fact] - public async Task ExecuteAsync_ShouldThrowInvalidOperationException_WhenMachineNameIsEmpty() - { - // Arrange - PredictionServiceConfiguration config = CreateValidConfiguration(); - config.Machines[0].MachineName = string.Empty; - _mockConfiguration.Setup(x => x.Value).Returns(config); - - // Act & Assert - await Assert.ThrowsAsync(() => _backgroundService.StartAsync(_cancellationTokenSource.Token)); - } - - [Fact] - public async Task ExecuteAsync_ShouldThrowInvalidOperationException_WhenNoInputSensorsConfigured() - { - // Arrange - PredictionServiceConfiguration config = CreateValidConfiguration(); - config.Machines[0].InputSensors.Clear(); - _mockConfiguration.Setup(x => x.Value).Returns(config); - - // Act & Assert - await Assert.ThrowsAsync(() => _backgroundService.StartAsync(_cancellationTokenSource.Token)); - } - - [Fact] - public async Task ExecuteAsync_ShouldThrowInvalidOperationException_WhenPreprocessingStrategyIsEmpty() - { - // Arrange - PredictionServiceConfiguration config = CreateValidConfiguration(); - config.Machines[0].PreprocessingStrategy = string.Empty; - _mockConfiguration.Setup(x => x.Value).Returns(config); - - // Act & Assert - await Assert.ThrowsAsync(() => _backgroundService.StartAsync(_cancellationTokenSource.Token)); - } - [Fact] public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() { - // Arrange - PredictionServiceConfiguration config = CreateValidConfiguration(); - _mockConfiguration.Setup(x => x.Value).Returns(config); + var config = CreateValidConfiguration(); - // Act Task task = _backgroundService.StartAsync(_cancellationTokenSource.Token); - await Task.Delay(100); // Give it time to start + await Task.Delay(100); _cancellationTokenSource.Cancel(); await _backgroundService.StopAsync(_cancellationTokenSource.Token); - // Assert - // The service should have stopped without throwing exceptions Assert.True(true); } @@ -178,47 +110,52 @@ public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() #region Helper methods - private static PredictionServiceConfiguration CreateValidConfiguration() + private static PredictionServiceConfiguration CreateValidConfiguration() => new PredictionServiceConfiguration { - // Create a temporary model file for testing - string tempModelPath = Path.GetTempFileName() + ".onnx"; - File.WriteAllText(tempModelPath, "dummy model content"); - - return new PredictionServiceConfiguration - { - Machines = + Machines = [ new MachinePredictionConfig { MachineName = "test_machine_1", - ModelPath = tempModelPath, InputSensors = ["sensor1", "sensor2"], - PreprocessingStrategy = "ActuatorMergingCurrent", - WindowSizeSeconds = 300, + WindowSize = 300, CycleIntervalSeconds = 60, Enabled = true, - Preprocessing = new PreprocessingConfig + ProcessingPipeline = new List { - EnableZScoreNormalization = true + new ProcessorDescription + { + Name = "actuatorcurrent", + Configuration = new PreprocessingConfig + { + EnableZScoreNormalization = true, + NormalizationParameters = new Dictionary() + } + } }, }, new MachinePredictionConfig { MachineName = "test_machine_2", - ModelPath = tempModelPath, InputSensors = ["sensor3", "sensor4"], - PreprocessingStrategy = "ActuatorMergingCurrent", - WindowSizeSeconds = 600, + WindowSize = 600, CycleIntervalSeconds = 120, Enabled = true, - Preprocessing = new PreprocessingConfig + ProcessingPipeline = new List { - EnableZScoreNormalization = true + new ProcessorDescription + { + Name = "actuatorcurrent", + Configuration = new PreprocessingConfig + { + EnableZScoreNormalization = true, + NormalizationParameters = new Dictionary() + } + } }, } ], - }; - } + }; #endregion