From 7fc5a694529855ad9139e4342d03a03742228cea Mon Sep 17 00:00:00 2001 From: CoJaques Date: Sun, 27 Jul 2025 18:09:58 +0200 Subject: [PATCH 01/13] feat: implement list of data processor --- .../Configuration/MachinePredictionConfig.cs | 14 +- src/DataAggregator.Processor/Program.cs | 5 +- .../Services/DataStorage/IDataRepository.cs | 2 +- .../DataStorage/InfluxV3Repository.cs | 2 +- .../PreProcessing/IPreprocessingStrategy.cs | 18 -- .../IPreprocessingStrategyFactory.cs | 14 -- .../PreprocessingStrategyFactory.cs | 26 --- .../Prediction/IOnnxPredictionEngine.cs | 17 -- .../Prediction/MachinePredictionProcessor.cs | 213 ++++++++---------- .../Services/PredictionBackgroundService.cs | 21 -- .../Processing/Abstraction/IDataProcessor.cs | 16 ++ .../Factory/DataProcessorFactory.cs | 53 +++++ .../Factory/IDataProcessorFactory.cs | 17 ++ .../Processing/Onnx/OnnxPredictionConfig.cs | 12 + .../Onnx}/OnnxPredictionEngine.cs | 28 +-- .../StateDeductionPostProcessor.cs | 74 ++++++ .../StateDeductionPostProcessorConfig.cs | 12 + .../ActuatorCurrentFeatureExtractor.cs | 75 +++--- .../MathUtils.cs | 2 +- .../PreprocessingConfig.cs | 2 +- src/DataAggregator.Processor/appsettings.json | 53 +++-- 21 files changed, 368 insertions(+), 308 deletions(-) delete mode 100644 src/DataAggregator.Processor/Services/PreProcessing/IPreprocessingStrategy.cs delete mode 100644 src/DataAggregator.Processor/Services/PreProcessing/IPreprocessingStrategyFactory.cs delete mode 100644 src/DataAggregator.Processor/Services/PreProcessing/PreprocessingStrategyFactory.cs delete mode 100644 src/DataAggregator.Processor/Services/Prediction/IOnnxPredictionEngine.cs create mode 100644 src/DataAggregator.Processor/Services/Processing/Abstraction/IDataProcessor.cs create mode 100644 src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs create mode 100644 src/DataAggregator.Processor/Services/Processing/Factory/IDataProcessorFactory.cs create mode 100644 src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionConfig.cs rename src/DataAggregator.Processor/Services/{Prediction => Processing/Onnx}/OnnxPredictionEngine.cs (89%) create mode 100644 src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs create mode 100644 src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessorConfig.cs rename src/DataAggregator.Processor/Services/{ => Processing}/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs (83%) rename src/DataAggregator.Processor/Services/{ => Processing}/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs (97%) rename src/DataAggregator.Processor/{Configuration => Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing}/PreprocessingConfig.cs (83%) diff --git a/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs b/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs index 5f94703..1878a4a 100644 --- a/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs +++ b/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs @@ -15,16 +15,6 @@ 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. - /// - public string PreprocessingStrategy { get; set; } = string.Empty; - /// /// Gets or sets the list of input sensor names. /// @@ -41,7 +31,7 @@ public class MachinePredictionConfig public int 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; } } diff --git a/src/DataAggregator.Processor/Program.cs b/src/DataAggregator.Processor/Program.cs index d71ea41..acf3413 100644 --- a/src/DataAggregator.Processor/Program.cs +++ b/src/DataAggregator.Processor/Program.cs @@ -2,7 +2,7 @@ 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; @@ -38,8 +38,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..d17cd27 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. diff --git a/src/DataAggregator.Processor/Services/DataStorage/InfluxV3Repository.cs b/src/DataAggregator.Processor/Services/DataStorage/InfluxV3Repository.cs index f8cae55..52a6281 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(); 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..c2f57b0 100644 --- a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs +++ b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs @@ -1,123 +1,54 @@ +using System.Text.Json; 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); - - if (collectorInfo == null) - { - Log.Warning("Collector info not found for machine {MachineName}", config.MachineName); - 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(); - - 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); - } + CollectorInfoDto? collectorInfo = await FetchCollectorInfo(config.MachineName); + if (collectorInfo == null) return; - // Get sensor info for requested sensors - var requestedSensorInfos = requestedSensors - .Select(sensorName => availableSensors[sensorName]) - .ToList(); + List? requestedSensors = ValidateAndGetRequestedSensors(config, collectorInfo); + if (requestedSensors == null) return; - // Fetch data window with sensor type information - List measurements = await FetchDataWindowAsync(config, requestedSensorInfos); + InitializeRepositoryIfNeeded(collectorInfo); + List measurements = await FetchDataWindowAsync(config, requestedSensors); if (measurements.Count == 0) { 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) 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 +57,101 @@ public async Task ProcessAsync(MachinePredictionConfig config) } } - #endregion - #region Private methods - - private async Task> FetchDataWindowAsync(MachinePredictionConfig config, List sensors) + private async Task FetchCollectorInfo(string machineName) { - DateTime endTime = DateTime.UtcNow; - DateTime startTime = endTime.AddSeconds(-config.WindowSizeSeconds); - - return await influxRepository.QueryMeasurementsAsync( - config.MachineName, - startTime, - endTime, - sensors); + CollectorInfoDto? collectorInfo = await registrationClient.GetCollectorInfoAsync(machineName); + if (collectorInfo == null) + Log.Warning("Collector info not found for machine {MachineName}", machineName); + return collectorInfo; } - private IEnumerable PreprocessDataAsync(IEnumerable measurements, MachinePredictionConfig config) + private List? ValidateAndGetRequestedSensors(MachinePredictionConfig config, CollectorInfoDto collectorInfo) { - try + var availableSensors = collectorInfo.Sensors.ToDictionary(s => s.SensorName); + var requestedSensors = config.InputSensors.Where(availableSensors.ContainsKey).ToList(); + + if (requestedSensors.Count != config.InputSensors.Count) { - if (string.IsNullOrEmpty(config.PreprocessingStrategy)) + 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 preprocessing strategy configured for machine {MachineName}", config.MachineName); - return Array.Empty(); + Log.Error("No valid sensors found for machine {MachineName}", config.MachineName); + return null; } + } - IPreprocessingStrategy strategy = strategyFactory.CreateStrategy(config.PreprocessingStrategy); - IEnumerable preprocessedData = strategy.PreprocessAsync(measurements, config); + return requestedSensors.Select(s => availableSensors[s]).ToList(); + } - Log.Debug( - "Preprocessed data for machine {MachineName} using strategy {Strategy}: {InputCount} inputs", - config.MachineName, - config.PreprocessingStrategy, - preprocessedData.Count()); + private void InitializeRepositoryIfNeeded(CollectorInfoDto collectorInfo) + { + if (_lastEndpoint != collectorInfo.AssignedInfluxEndpoint.Endpoint) + { + influxRepository.Initialize( + collectorInfo.AssignedInfluxEndpoint.Endpoint, + collectorInfo.AssignedInfluxEndpoint.Token); + _lastEndpoint = collectorInfo.AssignedInfluxEndpoint.Endpoint; + Log.Debug("Reinitialized InfluxDB connection with new endpoint: {Endpoint}", _lastEndpoint); + } + } - return preprocessedData; + private async Task?> RunPipelineAsync(IEnumerable data, string machineName) + { + foreach (IDataProcessor processor in _pipelineProcessors!) + { + try + { + data = await processor.ProcessAsync(data); + if (data is null || !data.Any()) + { + Log.Warning("Processor {Processor} returned no data for machine {MachineName}", processor.GetType().Name, machineName); + return null; + } + } + catch (Exception ex) + { + Log.Error(ex, "Error in processor {Processor} for machine {MachineName}", processor.GetType().Name, machineName); + return null; + } } - catch (Exception ex) + + return data; + } + + private bool BuildPipelineIfNeeded(MachinePredictionConfig config) + { + if (_pipelineProcessors == null) { - Log.Error(ex, "Error preprocessing data for machine {MachineName}", config.MachineName); - return Array.Empty(); + if (config.ProcessingPipeline == null) + { + Log.Error("No processing pipeline configured for machine {MachineName}", config.MachineName); + return false; + } + + string pipelineJson = JsonSerializer.Serialize(config.ProcessingPipeline); + var pipelineElements = JsonDocument.Parse(pipelineJson).RootElement.EnumerateArray().ToList(); + _pipelineProcessors = processorFactory.CreateProcessors(pipelineElements); } + + return true; } + private async Task> FetchDataWindowAsync(MachinePredictionConfig config, List sensors) + { + DateTime endTime = DateTime.UtcNow; + DateTime startTime = endTime.AddSeconds(-config.WindowSizeSeconds); + return await influxRepository.QueryMeasurementsAsync( + config.MachineName, + startTime, + endTime, + sensors); + } + #endregion + + #region Private fields + private List? _pipelineProcessors; + private string? _lastEndpoint; #endregion } diff --git a/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs b/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs index 796456a..3e7c7c5 100644 --- a/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs +++ b/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs @@ -96,18 +96,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 +115,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/Factory/DataProcessorFactory.cs b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs new file mode 100644 index 0000000..0e6420c --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs @@ -0,0 +1,53 @@ +using System.Text.Json; +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; + +/// +/// Factory implementation for creating processor strategies. +/// +public class DataProcessorFactory : IDataProcessorFactory +{ + /// + public List CreateProcessors(IEnumerable pipelineConfig) + { + var processors = new List(); + foreach (JsonElement element in pipelineConfig) + { + if (!element.TryGetProperty("Strategy", out JsonElement strategyProp)) + throw new ArgumentException("Each processor config must have a 'Strategy' property."); + + string? strategy = strategyProp.GetString()?.ToLowerInvariant(); + + switch (strategy) + { + case "actuatorcurrent": + PreprocessingConfig? preConfig = element.Deserialize(); + if (preConfig == null) + throw new ArgumentException("Preprocessing configuration is required for ActuatorCurrentFeatureExtractor."); + processors.Add(new ActuatorCurrentFeatureExtractor(preConfig)); + break; + case "onnxprediction": + OnnxPredictionConfig? onnxConfig = element.Deserialize(); + if (onnxConfig == null) + throw new ArgumentException("ONNX prediction configuration is required."); + processors.Add(new OnnxPredictionEngine(onnxConfig)); + break; + case "statedeductionpostprocessor": + StateDeductionPostProcessorConfig? postConfig = element.Deserialize(); + if (postConfig == null) + throw new ArgumentException("Post-processing configuration is required for MyCustomPostProcessor."); + processors.Add(new StateDeductionPostProcessor(postConfig)); + break; + default: + throw new ArgumentException($"Unknown processor strategy: {strategy}"); + } + } + + 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..2aee1bb --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/Factory/IDataProcessorFactory.cs @@ -0,0 +1,17 @@ +using System.Text.Json; +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 pipelineConfig); +} 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..d76903a --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionConfig.cs @@ -0,0 +1,12 @@ +namespace DataAggregator.Processor.Services.Processing.Onnx; + +/// +/// Configuration for ONNX prediction. +/// +public class OnnxPredictionConfig +{ + /// + /// 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 89% rename from src/DataAggregator.Processor/Services/Prediction/OnnxPredictionEngine.cs rename to src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs index d19e5e7..ef93280 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,36 @@ 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 - InferenceSession session = LoadOrGetModel(modelPath); + InferenceSession session = LoadOrGetModel(config.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 +60,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 +70,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..cc0090c --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs @@ -0,0 +1,74 @@ +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 +{ + private readonly string _resultOutputName = "PredictedLabel.output_0"; + private string? _lastState = null; + private int _stableCount = 0; + + /// + 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) + { + // first run, initialize the last state + _lastState = currentPredictedState; + _stableCount = 1; + return Task.FromResult>(inputList); + } + + if (currentPredictedState == _lastState) + { + // stable state, increment the stable count + _stableCount++; + return Task.FromResult>(inputList); + } + else + { + // state change detected + _stableCount = 1; + + // Accept the state change only if the stable count reaches the threshold + if (_stableCount >= config.Threshold) + { + _lastState = currentPredictedState; + return Task.FromResult>(inputList); + } + else + { + if (prediction != null) + { + var forced = new MeasurementDataWrapper(prediction, _lastState!); + var output = inputList.Select(x => x.SensorName == _resultOutputName ? forced : x).ToList(); + return Task.FromResult>(output); + } + else + { + return Task.FromResult>(inputList); + } + } + } + } + + private class MeasurementDataWrapper(IMeasurementData original, string forcedValue) : IMeasurementData + { + public DateTime TimeStamp => original.TimeStamp; + + public string SensorName => original.SensorName; + + public Type ValueType => typeof(string); + + public object GetRawValue() => forcedValue; + } +} 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..efd57b6 --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessorConfig.cs @@ -0,0 +1,12 @@ +namespace DataAggregator.Processor.Services.Processing.PostProcessing.StateDeductionPostProcess; + +/// +/// Configuration for . +/// +public class StateDeductionPostProcessorConfig +{ + /// + /// 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..10ceb2b 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.Count() != 0) + meanTime = input.ElementAt(input.Count() / 2).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 83% rename from src/DataAggregator.Processor/Configuration/PreprocessingConfig.cs rename to src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs index 3b1a9ce..5825ccb 100644 --- a/src/DataAggregator.Processor/Configuration/PreprocessingConfig.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs @@ -1,4 +1,4 @@ -namespace DataAggregator.Processor.Configuration; +namespace DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; /// /// Configuration for preprocessing operations including Z-score normalization. diff --git a/src/DataAggregator.Processor/appsettings.json b/src/DataAggregator.Processor/appsettings.json index 3f0cc8d..e5f7af0 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,38 @@ "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 ] + "WindowSizeSeconds": 3, + "CycleIntervalSeconds": 1, + "ProcessingPipeline": [ + { + "Strategy": "actuatorcurrent", + "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] + } + }, + { + "Strategy": "onnxprediction", + "ModelPath": "resources/opencn_model.onnx" + }, + { + "Strategy": "myCustomPostProcess", + "Threshold": 0.5 } - } + ] } ] } From 90f4d8cbd9fd88a499adbdf80841062b9f01b5c0 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Sun, 27 Jul 2025 18:10:09 +0200 Subject: [PATCH 02/13] feat: adapt tests --- .../ActuatorCurrentFeatureExtractorTests.cs | 121 ++--------- .../DataProcessorFactoryTests.cs | 76 +++++++ .../Services/PreProcessing/MathUtilsTests.cs | 2 +- .../PreprocessingStrategyFactoryTests.cs | 48 ----- .../MachinePredictionProcessorTests.cs | 198 ++++-------------- .../Prediction/OnnxPredictionEngineTests.cs | 121 +++-------- .../PredictionBackgroundServiceTests.cs | 88 ++------ 7 files changed, 191 insertions(+), 463 deletions(-) create mode 100644 tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs delete mode 100644 tests/DataAggregator.Processor.Tests/Services/PreProcessing/PreprocessingStrategyFactoryTests.cs 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..b84065b --- /dev/null +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs @@ -0,0 +1,76 @@ +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.Text.Json; + +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 + string pipelineJson = @"[ + { ""Strategy"": ""actuatorcurrent"", ""EnableZScoreNormalization"": true, ""NormalizationParameters"": {} }, + { ""Strategy"": ""onnxprediction"", ""ModelPath"": ""model.onnx"" }, + { ""Strategy"": ""statedeductionpostprocessor"", ""Threshold"": 2 } + ]"; + var pipelineElements = JsonDocument.Parse(pipelineJson).RootElement.EnumerateArray().ToList(); + + // Act + var processors = _factory.CreateProcessors(pipelineElements); + + // Assert + Assert.Equal(3, processors.Count); + Assert.IsType(processors[0]); + Assert.IsType(processors[1]); + Assert.IsType(processors[2]); + } + + [Fact] + public void CreateProcessors_ShouldThrowArgumentException_ForUnknownStrategy() + { + // Arrange + string pipelineJson = @"[ + { ""Strategy"": ""unknownstrategy"" } + ]"; + var pipelineElements = JsonDocument.Parse(pipelineJson).RootElement.EnumerateArray().ToList(); + + // Act & Assert + Assert.Throws(() => _factory.CreateProcessors(pipelineElements)); + } + + [Fact] + public void CreateProcessors_ShouldReturnEmptyList_ForEmptyPipeline() + { + // Arrange + string pipelineJson = "[]"; + var pipelineElements = JsonDocument.Parse(pipelineJson).RootElement.EnumerateArray().ToList(); + + // Act + var processors = _factory.CreateProcessors(pipelineElements); + + // Assert + Assert.Empty(processors); + } + + [Fact] + public void CreateProcessors_ShouldThrowArgumentException_WhenStrategyMissing() + { + // Arrange + string pipelineJson = @"[ + { ""EnableZScoreNormalization"": true } + ]"; + var pipelineElements = JsonDocument.Parse(pipelineJson).RootElement.EnumerateArray().ToList(); + + // Act & Assert + Assert.Throws(() => _factory.CreateProcessors(pipelineElements)); + } +} 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..24cf397 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, CycleIntervalSeconds = 60, Enabled = true, + ProcessingPipeline = new List { new { Strategy = "mock" } } }; 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..d274df0 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs @@ -56,33 +56,26 @@ public PredictionBackgroundServiceTests() [Fact] public async Task ExecuteAsync_ShouldStartSuccessfully_WhenValidConfigurationProvided() { - // Arrange - PredictionServiceConfiguration config = CreateValidConfiguration(); + var config = CreateValidConfiguration(); _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 - 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(); + var config = CreateValidConfiguration(); _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 _mockPredictionProcessor.Verify( x => x.ProcessAsync(It.IsAny()), Times.Exactly(config.Machines.Count(m => m.Enabled))); @@ -91,86 +84,50 @@ 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(); + var 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(); + var 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(); + var config = CreateValidConfiguration(); _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); _cancellationTokenSource.Cancel(); await _backgroundService.StopAsync(_cancellationTokenSource.Token); - // Assert - // The service should have stopped without throwing exceptions Assert.True(true); } @@ -178,47 +135,36 @@ 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, CycleIntervalSeconds = 60, Enabled = true, - Preprocessing = new PreprocessingConfig + ProcessingPipeline = new List { - EnableZScoreNormalization = true + new { Strategy = "actuatorcurrent", EnableZScoreNormalization = true, NormalizationParameters = new Dictionary() }, }, }, new MachinePredictionConfig { MachineName = "test_machine_2", - ModelPath = tempModelPath, InputSensors = ["sensor3", "sensor4"], - PreprocessingStrategy = "ActuatorMergingCurrent", WindowSizeSeconds = 600, CycleIntervalSeconds = 120, Enabled = true, - Preprocessing = new PreprocessingConfig + ProcessingPipeline = new List { - EnableZScoreNormalization = true + new { Strategy = "actuatorcurrent", EnableZScoreNormalization = true, NormalizationParameters = new Dictionary() }, }, } ], - }; - } + }; #endregion From b5f462a7a543d0407bf043343b70dfe7e5a0165c Mon Sep 17 00:00:00 2001 From: CoJaques Date: Mon, 28 Jul 2025 10:36:09 +0200 Subject: [PATCH 03/13] fix: Configuration management --- .../Configuration/MachinePredictionConfig.cs | 6 +- src/DataAggregator.Processor/Program.cs | 15 +++- .../Prediction/MachinePredictionProcessor.cs | 5 +- .../Services/PredictionBackgroundService.cs | 9 ++- .../Abstraction/IProcessorConfiguration.cs | 8 +++ .../Factory/DataProcessorFactory.cs | 40 +++++------ .../Factory/IDataProcessorFactory.cs | 5 +- .../Factory/ProcessorDescription.cs | 21 ++++++ .../ProcessorDescriptionJsonConverter.cs | 44 ++++++++++++ .../Processing/Onnx/OnnxPredictionConfig.cs | 4 +- .../Processing/Onnx/OnnxPredictionEngine.cs | 4 +- .../StateDeductionPostProcessorConfig.cs | 4 +- .../PreprocessingConfig.cs | 6 +- .../DataProcessorFactoryTests.cs | 68 ++++++++++++------- .../MachinePredictionProcessorTests.cs | 8 +-- .../PredictionBackgroundServiceTests.cs | 26 +++++-- 16 files changed, 198 insertions(+), 75 deletions(-) create mode 100644 src/DataAggregator.Processor/Services/Processing/Abstraction/IProcessorConfiguration.cs create mode 100644 src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescription.cs create mode 100644 src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs diff --git a/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs b/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs index 1878a4a..d3ba85e 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. @@ -33,5 +35,5 @@ public class MachinePredictionConfig /// /// Gets or sets the processing pipeline for this machine. /// - public List? ProcessingPipeline { get; set; } + public List ProcessingPipeline { get; set; } = new(); } diff --git a/src/DataAggregator.Processor/Program.cs b/src/DataAggregator.Processor/Program.cs index acf3413..6827537 100644 --- a/src/DataAggregator.Processor/Program.cs +++ b/src/DataAggregator.Processor/Program.cs @@ -1,3 +1,4 @@ +using System.Text.Json; using DataAggregator.Processor.Configuration; using DataAggregator.Processor.Services; using DataAggregator.Processor.Services.DataStorage; @@ -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 => diff --git a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs index c2f57b0..0aed8a9 100644 --- a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs +++ b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs @@ -1,4 +1,3 @@ -using System.Text.Json; using DataAggregator.Collector.Shared.Models; using DataAggregator.Processor.Configuration; using DataAggregator.Processor.Services.DataStorage; @@ -130,9 +129,7 @@ private bool BuildPipelineIfNeeded(MachinePredictionConfig config) return false; } - string pipelineJson = JsonSerializer.Serialize(config.ProcessingPipeline); - var pipelineElements = JsonDocument.Parse(pipelineJson).RootElement.EnumerateArray().ToList(); - _pipelineProcessors = processorFactory.CreateProcessors(pipelineElements); + _pipelineProcessors = processorFactory.CreateProcessors(config.ProcessingPipeline); } return true; diff --git a/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs b/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs index 3e7c7c5..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) { 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 index 0e6420c..4e0a97a 100644 --- a/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs +++ b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs @@ -1,4 +1,3 @@ -using System.Text.Json; using DataAggregator.Processor.Services.Prediction; using DataAggregator.Processor.Services.Processing.Abstraction; using DataAggregator.Processor.Services.Processing.Onnx; @@ -8,43 +7,38 @@ namespace DataAggregator.Processor.Services.Processing.Factory; /// -/// Factory implementation for creating processor strategies. +/// Defines a factory for creating data processors based on a pipeline description. /// public class DataProcessorFactory : IDataProcessorFactory { /// - public List CreateProcessors(IEnumerable pipelineConfig) + public List CreateProcessors(IEnumerable pipeline) { var processors = new List(); - foreach (JsonElement element in pipelineConfig) + foreach (ProcessorDescription desc in pipeline) { - if (!element.TryGetProperty("Strategy", out JsonElement strategyProp)) - throw new ArgumentException("Each processor config must have a 'Strategy' property."); - - string? strategy = strategyProp.GetString()?.ToLowerInvariant(); - - switch (strategy) + switch (desc.Name.ToLowerInvariant()) { case "actuatorcurrent": - PreprocessingConfig? preConfig = element.Deserialize(); - if (preConfig == null) - throw new ArgumentException("Preprocessing configuration is required for ActuatorCurrentFeatureExtractor."); - processors.Add(new ActuatorCurrentFeatureExtractor(preConfig)); + if (desc.Configuration is PreprocessingConfig preConfig) + processors.Add(new ActuatorCurrentFeatureExtractor(preConfig)); + else + throw new ArgumentException("Invalid config type for actuatorcurrent"); break; case "onnxprediction": - OnnxPredictionConfig? onnxConfig = element.Deserialize(); - if (onnxConfig == null) - throw new ArgumentException("ONNX prediction configuration is required."); - processors.Add(new OnnxPredictionEngine(onnxConfig)); + if (desc.Configuration is OnnxPredictionConfig onnxConfig) + processors.Add(new OnnxPredictionEngine(onnxConfig)); + else + throw new ArgumentException("Invalid config type for onnxprediction"); break; case "statedeductionpostprocessor": - StateDeductionPostProcessorConfig? postConfig = element.Deserialize(); - if (postConfig == null) - throw new ArgumentException("Post-processing configuration is required for MyCustomPostProcessor."); - processors.Add(new StateDeductionPostProcessor(postConfig)); + 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 strategy: {strategy}"); + throw new ArgumentException($"Unknown processor name: {desc.Name}"); } } diff --git a/src/DataAggregator.Processor/Services/Processing/Factory/IDataProcessorFactory.cs b/src/DataAggregator.Processor/Services/Processing/Factory/IDataProcessorFactory.cs index 2aee1bb..2d2789d 100644 --- a/src/DataAggregator.Processor/Services/Processing/Factory/IDataProcessorFactory.cs +++ b/src/DataAggregator.Processor/Services/Processing/Factory/IDataProcessorFactory.cs @@ -1,4 +1,3 @@ -using System.Text.Json; using DataAggregator.Processor.Services.Processing.Abstraction; namespace DataAggregator.Processor.Services.Processing.Factory; @@ -11,7 +10,7 @@ public interface IDataProcessorFactory /// /// Create multiple data processor which correspond to the pipeline configuration. /// - /// The pipeline configuration. + /// The pipeline configuration. /// Configured pipeline strategy. - public List CreateProcessors(IEnumerable pipelineConfig); + 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 index d76903a..d8c1c09 100644 --- a/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionConfig.cs +++ b/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionConfig.cs @@ -1,9 +1,11 @@ +using DataAggregator.Processor.Services.Processing.Abstraction; + namespace DataAggregator.Processor.Services.Processing.Onnx; /// /// Configuration for ONNX prediction. /// -public class OnnxPredictionConfig +public class OnnxPredictionConfig : IProcessorConfiguration { /// /// Gets or sets the path of the model. diff --git a/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs b/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs index ef93280..ddb36b1 100644 --- a/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs +++ b/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs @@ -27,7 +27,9 @@ public async Task> ProcessAsync( try { // Load the model or get the existing session - InferenceSession session = LoadOrGetModel(config.ModelPath); + string executablePath = AppContext.BaseDirectory; + string modelPath = Path.Combine(executablePath, config.ModelPath); + InferenceSession session = LoadOrGetModel(modelPath); if (!input.Any()) { diff --git a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessorConfig.cs b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessorConfig.cs index efd57b6..3473550 100644 --- a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessorConfig.cs +++ b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessorConfig.cs @@ -1,9 +1,11 @@ +using DataAggregator.Processor.Services.Processing.Abstraction; + namespace DataAggregator.Processor.Services.Processing.PostProcessing.StateDeductionPostProcess; /// /// Configuration for . /// -public class StateDeductionPostProcessorConfig +public class StateDeductionPostProcessorConfig : IProcessorConfiguration { /// /// Gets or sets the threshold for minimum number of consecutive cycles required to deduce a state change. diff --git a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs index 5825ccb..bbec98f 100644 --- a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs @@ -1,9 +1,11 @@ -namespace DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; +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/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs index b84065b..e4cbe7b 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs @@ -3,7 +3,8 @@ using DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; using DataAggregator.Processor.Services.Prediction; using DataAggregator.Processor.Services.Processing.PostProcessing.StateDeductionPostProcess; -using System.Text.Json; +using System.Collections.Generic; +using DataAggregator.Processor.Services.Processing.Onnx; namespace DataAggregator.Processor.Tests.Services.PreProcessing; @@ -17,15 +18,37 @@ public class DataProcessorFactoryTests public void CreateProcessors_ShouldReturnCorrectProcessors_ForValidPipeline() { // Arrange - string pipelineJson = @"[ - { ""Strategy"": ""actuatorcurrent"", ""EnableZScoreNormalization"": true, ""NormalizationParameters"": {} }, - { ""Strategy"": ""onnxprediction"", ""ModelPath"": ""model.onnx"" }, - { ""Strategy"": ""statedeductionpostprocessor"", ""Threshold"": 2 } - ]"; - var pipelineElements = JsonDocument.Parse(pipelineJson).RootElement.EnumerateArray().ToList(); + 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(pipelineElements); + var processors = _factory.CreateProcessors(pipeline); // Assert Assert.Equal(3, processors.Count); @@ -35,42 +58,41 @@ public void CreateProcessors_ShouldReturnCorrectProcessors_ForValidPipeline() } [Fact] - public void CreateProcessors_ShouldThrowArgumentException_ForUnknownStrategy() + public void CreateProcessors_ShouldThrowArgumentException_ForUnknownName() { // Arrange - string pipelineJson = @"[ - { ""Strategy"": ""unknownstrategy"" } - ]"; - var pipelineElements = JsonDocument.Parse(pipelineJson).RootElement.EnumerateArray().ToList(); + var pipeline = new List + { + new ProcessorDescription { Name = "unknownstrategy", Configuration = null } + }; // Act & Assert - Assert.Throws(() => _factory.CreateProcessors(pipelineElements)); + Assert.Throws(() => _factory.CreateProcessors(pipeline)); } [Fact] public void CreateProcessors_ShouldReturnEmptyList_ForEmptyPipeline() { // Arrange - string pipelineJson = "[]"; - var pipelineElements = JsonDocument.Parse(pipelineJson).RootElement.EnumerateArray().ToList(); + var pipeline = new List(); // Act - var processors = _factory.CreateProcessors(pipelineElements); + var processors = _factory.CreateProcessors(pipeline); // Assert Assert.Empty(processors); } [Fact] - public void CreateProcessors_ShouldThrowArgumentException_WhenStrategyMissing() + public void CreateProcessors_ShouldThrowArgumentException_WhenNameMissing() { // Arrange - string pipelineJson = @"[ - { ""EnableZScoreNormalization"": true } - ]"; - var pipelineElements = JsonDocument.Parse(pipelineJson).RootElement.EnumerateArray().ToList(); + var pipeline = new List + { + new ProcessorDescription { Name = "", Configuration = new PreprocessingConfig() } + }; // Act & Assert - Assert.Throws(() => _factory.CreateProcessors(pipelineElements)); + Assert.Throws(() => _factory.CreateProcessors(pipeline)); } } diff --git a/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs b/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs index 24cf397..5118463 100644 --- a/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs @@ -89,7 +89,7 @@ public async Task ProcessAsync_ShouldCompleteSuccessfully_WhenAllConditionsAreMe .ReturnsAsync(measurements); _mockProcessor.Setup(x => x.ProcessAsync(It.IsAny>())) .ReturnsAsync(processedData); - _mockProcessorFactory.Setup(x => x.CreateProcessors(It.IsAny>())) + _mockProcessorFactory.Setup(x => x.CreateProcessors(It.IsAny>())) .Returns(new List { _mockProcessor.Object }); await _processor.ProcessAsync(config); @@ -113,7 +113,7 @@ public async Task ProcessAsync_ShouldReinitializeInfluxConnection_WhenEndpointCh .ReturnsAsync(measurements); _mockProcessor.Setup(x => x.ProcessAsync(It.IsAny>())) .ReturnsAsync(processedData); - _mockProcessorFactory.Setup(x => x.CreateProcessors(It.IsAny>())) + _mockProcessorFactory.Setup(x => x.CreateProcessors(It.IsAny>())) .Returns(new List { _mockProcessor.Object }); await _processor.ProcessAsync(config); // First call with endpoint1 @@ -141,7 +141,7 @@ public async Task ProcessAsync_ShouldNotReinitializeInfluxConnection_WhenEndpoin .ReturnsAsync(measurements); _mockProcessor.Setup(x => x.ProcessAsync(It.IsAny>())) .ReturnsAsync(processedData); - _mockProcessorFactory.Setup(x => x.CreateProcessors(It.IsAny>())) + _mockProcessorFactory.Setup(x => x.CreateProcessors(It.IsAny>())) .Returns(new List { _mockProcessor.Object }); await _processor.ProcessAsync(config); @@ -169,7 +169,7 @@ public async Task ProcessAsync_ShouldThrowException_WhenRegistrationClientThrows WindowSizeSeconds = 300, CycleIntervalSeconds = 60, Enabled = true, - ProcessingPipeline = new List { new { Strategy = "mock" } } + ProcessingPipeline = new List { new() } }; private static CollectorInfoDto CreateCollectorInfoWithSensors(IEnumerable sensorNames, string endpoint = "http://localhost:8086") => new( diff --git a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs index d274df0..c4a3c27 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs @@ -1,6 +1,8 @@ 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; @@ -146,9 +148,17 @@ public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() WindowSizeSeconds = 300, CycleIntervalSeconds = 60, Enabled = true, - ProcessingPipeline = new List + ProcessingPipeline = new List { - new { Strategy = "actuatorcurrent", EnableZScoreNormalization = true, NormalizationParameters = new Dictionary() }, + new ProcessorDescription + { + Name = "actuatorcurrent", + Configuration = new PreprocessingConfig + { + EnableZScoreNormalization = true, + NormalizationParameters = new Dictionary() + } + } }, }, new MachinePredictionConfig @@ -158,9 +168,17 @@ public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() WindowSizeSeconds = 600, CycleIntervalSeconds = 120, Enabled = true, - ProcessingPipeline = new List + ProcessingPipeline = new List { - new { Strategy = "actuatorcurrent", EnableZScoreNormalization = true, NormalizationParameters = new Dictionary() }, + new ProcessorDescription + { + Name = "actuatorcurrent", + Configuration = new PreprocessingConfig + { + EnableZScoreNormalization = true, + NormalizationParameters = new Dictionary() + } + } }, } ], From beb1809e94e13439474e76e7d9f413322beceab9 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Wed, 30 Jul 2025 20:58:34 +0200 Subject: [PATCH 04/13] fix: projects folder --- DataAggregator.sln | 2 -- 1 file changed, 2 deletions(-) diff --git a/DataAggregator.sln b/DataAggregator.sln index cddc1c4..371bf54 100644 --- a/DataAggregator.sln +++ b/DataAggregator.sln @@ -38,8 +38,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}" From c9ecf5877c48b70a0f23715e58d27deaad1671a2 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Wed, 30 Jul 2025 21:00:38 +0200 Subject: [PATCH 05/13] feat: add flush interval to collector configuration --- .../Abstraction/CollectorService.cs | 2 +- .../Abstraction/Configuration/CollectorConfiguration.cs | 5 +++++ src/DataAggregator.Collector/appsettings.json | 1 + 3 files changed, 7 insertions(+), 1 deletion(-) 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", From 62131cbdf6126e68fec9a7bfbac3ed01e3793e39 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Wed, 30 Jul 2025 21:01:30 +0200 Subject: [PATCH 06/13] feat: enhance post-processor --- .../Configuration/MachinePredictionConfig.cs | 11 +- .../Services/DataStorage/IDataRepository.cs | 9 + .../DataStorage/InfluxV3Repository.cs | 211 +++++++++++------- .../Prediction/MachinePredictionProcessor.cs | 24 +- .../StateDeductionPostProcessor.cs | 106 ++++++--- src/DataAggregator.Processor/appsettings.json | 53 +++-- .../MachinePredictionProcessorTests.cs | 2 +- .../PredictionBackgroundServiceTests.cs | 33 +-- 8 files changed, 275 insertions(+), 174 deletions(-) diff --git a/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs b/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs index d3ba85e..f681ef4 100644 --- a/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs +++ b/src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs @@ -23,14 +23,19 @@ public class MachinePredictionConfig public List InputSensors { get; set; } = []; /// - /// Gets or sets the window size in seconds for data collection. + /// Gets or sets a value indicating whether Window size in seconds if true, otherwise in elements number. /// - public int WindowSizeSeconds { get; set; } = 60; + public bool WindowSizeInSeconds { get; set; } = true; + + /// + /// Gets or sets the window size unit depends on WindowSizeInSeconds property. + /// + 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 processing pipeline for this machine. diff --git a/src/DataAggregator.Processor/Services/DataStorage/IDataRepository.cs b/src/DataAggregator.Processor/Services/DataStorage/IDataRepository.cs index d17cd27..833a89c 100644 --- a/src/DataAggregator.Processor/Services/DataStorage/IDataRepository.cs +++ b/src/DataAggregator.Processor/Services/DataStorage/IDataRepository.cs @@ -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 52a6281..7269174 100644 --- a/src/DataAggregator.Processor/Services/DataStorage/InfluxV3Repository.cs +++ b/src/DataAggregator.Processor/Services/DataStorage/InfluxV3Repository.cs @@ -43,97 +43,52 @@ public void Initialize(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/Prediction/MachinePredictionProcessor.cs b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs index 0aed8a9..dfc6f60 100644 --- a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs +++ b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs @@ -137,13 +137,23 @@ private bool BuildPipelineIfNeeded(MachinePredictionConfig config) private async Task> FetchDataWindowAsync(MachinePredictionConfig config, List sensors) { - DateTime endTime = DateTime.UtcNow; - DateTime startTime = endTime.AddSeconds(-config.WindowSizeSeconds); - return await influxRepository.QueryMeasurementsAsync( - config.MachineName, - startTime, - endTime, - sensors); + if (config.WindowSizeInSeconds) + { + DateTime endTime = DateTime.UtcNow; + DateTime startTime = endTime.AddSeconds(-config.WindowSize); + return await influxRepository.QueryMeasurementsAsync( + config.MachineName, + startTime, + endTime, + sensors); + } + else + { + return await influxRepository.QueryLastMeasurements( + config.MachineName, + config.WindowSize, + sensors); + } } #endregion diff --git a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs index cc0090c..0b2d757 100644 --- a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs @@ -9,9 +9,7 @@ namespace DataAggregator.Processor.Services.Processing.PostProcessing.StateDeduc /// public class StateDeductionPostProcessor(StateDeductionPostProcessorConfig config) : IDataProcessor { - private readonly string _resultOutputName = "PredictedLabel.output_0"; - private string? _lastState = null; - private int _stableCount = 0; + #region Public methods /// public Task> ProcessAsync(IEnumerable input) @@ -21,46 +19,82 @@ public Task> ProcessAsync(IEnumerable= config.Threshold) + return ConfirmPendingState(); + + return Task.FromResult(Enumerable.Empty()); + } + #endregion + + #region Private Methods + private Task> InitializeFirstState(List inputList, string state) + { + _lastState = state; + _pendingState = state; + return Task.FromResult>(inputList); + } + + private Task> HandleSameAsLastState(List inputList, string state) + { + if (_bufferedBatches.Count > 0) { - // first run, initialize the last state - _lastState = currentPredictedState; - _stableCount = 1; - return Task.FromResult>(inputList); + var corrected = _bufferedBatches + .SelectMany(batch => OverrideStateInBatch(batch, _lastState!)) + .ToList(); + + _bufferedBatches.Clear(); + _pendingState = state; + _stableCount = 0; + + return Task.FromResult>(corrected); } - if (currentPredictedState == _lastState) + _pendingState = state; + _stableCount = 0; + return Task.FromResult>(inputList); + } + + private void HandlePotentialStateChange(string currentState, List inputList) + { + if (currentState == _pendingState) { - // stable state, increment the stable count _stableCount++; - return Task.FromResult>(inputList); + _bufferedBatches.Add(inputList); } else { - // state change detected + _pendingState = currentState; _stableCount = 1; - - // Accept the state change only if the stable count reaches the threshold - if (_stableCount >= config.Threshold) - { - _lastState = currentPredictedState; - return Task.FromResult>(inputList); - } - else - { - if (prediction != null) - { - var forced = new MeasurementDataWrapper(prediction, _lastState!); - var output = inputList.Select(x => x.SensorName == _resultOutputName ? forced : x).ToList(); - return Task.FromResult>(output); - } - else - { - return Task.FromResult>(inputList); - } - } + _bufferedBatches.Clear(); + _bufferedBatches.Add(inputList); } } + private Task> ConfirmPendingState() + { + _lastState = _pendingState; + + var confirmed = _bufferedBatches + .SelectMany(batch => OverrideStateInBatch(batch, _lastState!)) + .ToList(); + + _bufferedBatches.Clear(); + return Task.FromResult>(confirmed); + } + + private IEnumerable OverrideStateInBatch(List batch, string forcedValue) + => batch.Select( + x => x.SensorName == _resultOutputName + ? new MeasurementDataWrapper(x, forcedValue) + : x); + private class MeasurementDataWrapper(IMeasurementData original, string forcedValue) : IMeasurementData { public DateTime TimeStamp => original.TimeStamp; @@ -71,4 +105,14 @@ private class MeasurementDataWrapper(IMeasurementData original, string forcedVal public object GetRawValue() => forcedValue; } + #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/appsettings.json b/src/DataAggregator.Processor/appsettings.json index e5f7af0..19ac701 100644 --- a/src/DataAggregator.Processor/appsettings.json +++ b/src/DataAggregator.Processor/appsettings.json @@ -32,36 +32,43 @@ "current-amp-c", "current-amp-s" ], - "WindowSizeSeconds": 3, - "CycleIntervalSeconds": 1, + "WindowSize": 50, + "WindowSizeInSeconds": false, + "CycleIntervalSeconds": 0.5, "ProcessingPipeline": [ { - "Strategy": "actuatorcurrent", - "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] + "Name": "actuatorcurrent", + "Configuration": { + "EnableZScoreNormalization": true, + "NormalizationParameters": { + "GlobalActivityRatio": [ 0.004067, 0.018027 ], + "GlobalChangeDensity": [ 0.433931, 0.146315 ], + "InterAxisMeanCorrelation": [ -0.006368, 0.068250 ], + "InterAxisMaxCorrelation": [ 0.432673, 0.243142 ], + "InterAxisCorrelationVariance": [ 0.224938, 0.112596 ], + "AxisSynchronization": [ -60.031723, 818.528442 ], + "AxisLoadBalance": [ -0.044534, 0.330180 ], + "TemporalStability": [ 0.964727, 0.038193 ], + "GlobalSkewness": [ -0.119681, 0.371627 ], + "GlobalKurtosis": [ -0.568834, 0.398853 ], + "GlobalTrendSlope": [ 0.000011, 0.000304 ], + "CoefficientOfVariation": [ -11.505261, 823.236694 ], + "NormalizedIqrMedian": [ -22.622700, 1259.548584 ], + "NormalizedIqrMean": [ -6.469841, 807.403015 ] + } } }, { - "Strategy": "onnxprediction", - "ModelPath": "resources/opencn_model.onnx" + "Name": "onnxprediction", + "Configuration": { + "ModelPath": "resources/opencn_model.onnx" + } }, { - "Strategy": "myCustomPostProcess", - "Threshold": 0.5 + "Name": "statedeductionpostprocessor", + "Configuration": { + "Threshold": 2 + } } ] } diff --git a/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs b/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs index 5118463..6721c30 100644 --- a/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/Prediction/MachinePredictionProcessorTests.cs @@ -166,7 +166,7 @@ public async Task ProcessAsync_ShouldThrowException_WhenRegistrationClientThrows { MachineName = "test_machine", InputSensors = ["sensor1", "sensor2"], - WindowSizeSeconds = 300, + WindowSize = 300, CycleIntervalSeconds = 60, Enabled = true, ProcessingPipeline = new List { new() } diff --git a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs index c4a3c27..01d35d7 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs @@ -4,7 +4,6 @@ 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; @@ -14,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; @@ -24,7 +22,6 @@ public class PredictionBackgroundServiceTests : IDisposable public PredictionBackgroundServiceTests() { - _mockConfiguration = new Mock>(); _mockPredictionProcessor = new Mock(); _cancellationTokenSource = new CancellationTokenSource(); @@ -49,7 +46,7 @@ public PredictionBackgroundServiceTests() .Returns(_mockScope.Object); _backgroundService = new PredictionBackgroundService( - _mockConfiguration.Object, + CreateValidConfiguration(), _serviceProvider.Object); } @@ -59,7 +56,6 @@ public PredictionBackgroundServiceTests() public async Task ExecuteAsync_ShouldStartSuccessfully_WhenValidConfigurationProvided() { var config = CreateValidConfiguration(); - _mockConfiguration.Setup(x => x.Value).Returns(config); Task task = _backgroundService.StartAsync(_cancellationTokenSource.Token); await Task.Delay(100); @@ -72,7 +68,6 @@ public async Task ExecuteAsync_ShouldStartSuccessfully_WhenValidConfigurationPro public async Task ExecuteAsync_ShouldScheduleEnabledMachines_WhenConfigurationContainsEnabledMachines() { var config = CreateValidConfiguration(); - _mockConfiguration.Setup(x => x.Value).Returns(config); Task task = _backgroundService.StartAsync(_cancellationTokenSource.Token); await Task.Delay(100); @@ -90,7 +85,6 @@ public async Task ExecuteAsync_ShouldHandleEmptyMachineList_WhenConfigurationCon { Machines = [], }; - _mockConfiguration.Setup(x => x.Value).Returns(config); Task task = _backgroundService.StartAsync(_cancellationTokenSource.Token); await Task.Delay(100); @@ -99,31 +93,10 @@ public async Task ExecuteAsync_ShouldHandleEmptyMachineList_WhenConfigurationCon Assert.True(true); } - [Fact] - public async Task ExecuteAsync_ShouldThrowInvalidOperationException_WhenMachineNameIsEmpty() - { - var config = CreateValidConfiguration(); - config.Machines[0].MachineName = string.Empty; - _mockConfiguration.Setup(x => x.Value).Returns(config); - - await Assert.ThrowsAsync(() => _backgroundService.StartAsync(_cancellationTokenSource.Token)); - } - - [Fact] - public async Task ExecuteAsync_ShouldThrowInvalidOperationException_WhenNoInputSensorsConfigured() - { - var config = CreateValidConfiguration(); - config.Machines[0].InputSensors.Clear(); - _mockConfiguration.Setup(x => x.Value).Returns(config); - - await Assert.ThrowsAsync(() => _backgroundService.StartAsync(_cancellationTokenSource.Token)); - } - [Fact] public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() { var config = CreateValidConfiguration(); - _mockConfiguration.Setup(x => x.Value).Returns(config); Task task = _backgroundService.StartAsync(_cancellationTokenSource.Token); await Task.Delay(100); @@ -145,7 +118,7 @@ public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() { MachineName = "test_machine_1", InputSensors = ["sensor1", "sensor2"], - WindowSizeSeconds = 300, + WindowSize = 300, CycleIntervalSeconds = 60, Enabled = true, ProcessingPipeline = new List @@ -165,7 +138,7 @@ public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() { MachineName = "test_machine_2", InputSensors = ["sensor3", "sensor4"], - WindowSizeSeconds = 600, + WindowSize = 600, CycleIntervalSeconds = 120, Enabled = true, ProcessingPipeline = new List From f0373eefe211c49bc615bec8881baae97cfdb1a5 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Wed, 30 Jul 2025 21:52:20 +0200 Subject: [PATCH 07/13] feat: implement batch processing --- .../Prediction/MachinePredictionProcessor.cs | 101 +++++++++++++++--- src/DataAggregator.Processor/appsettings.json | 2 +- 2 files changed, 85 insertions(+), 18 deletions(-) diff --git a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs index dfc6f60..c0d31ed 100644 --- a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs +++ b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs @@ -30,8 +30,8 @@ public async Task ProcessAsync(MachinePredictionConfig config) InitializeRepositoryIfNeeded(collectorInfo); - List measurements = await FetchDataWindowAsync(config, requestedSensors); - 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; @@ -96,27 +96,49 @@ private void InitializeRepositoryIfNeeded(CollectorInfoDto collectorInfo) } } - private async Task?> RunPipelineAsync(IEnumerable data, string machineName) + private async Task> RunPipelineAsync( + IEnumerable> dataBlocks, + string machineName) { - foreach (IDataProcessor processor in _pipelineProcessors!) + if (_pipelineProcessors is null || _pipelineProcessors.Count == 0) { - try + 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) { - data = await processor.ProcessAsync(data); - if (data is null || !data.Any()) + 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.Warning("Processor {Processor} returned no data for machine {MachineName}", processor.GetType().Name, machineName); - return null; + Log.Error(ex, "Error in processor {Processor} for machine {MachineName} (block skipped)", processor.GetType().Name, machineName); + current = null; + break; } } - catch (Exception ex) + + if (current is not null) { - Log.Error(ex, "Error in processor {Processor} for machine {MachineName}", processor.GetType().Name, machineName); - return null; + results.AddRange(current); } } - return data; + return results; } private bool BuildPipelineIfNeeded(MachinePredictionConfig config) @@ -135,30 +157,75 @@ private bool BuildPipelineIfNeeded(MachinePredictionConfig config) return true; } - private async Task> FetchDataWindowAsync(MachinePredictionConfig config, List sensors) + private async Task>> FetchDataWindowAsync(MachinePredictionConfig config, List sensors) { + List> measurements = []; + + // If we want to fetch the last N measurements, we must take inaccount the number of sensors + int windowSize = config.WindowSize * config.InputSensors.Count; + + // 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); - return await influxRepository.QueryMeasurementsAsync( + + List datas = await influxRepository.QueryMeasurementsAsync( config.MachineName, startTime, endTime, sensors); + + measurements.Add(datas); } else { - return await influxRepository.QueryLastMeasurements( + // 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, - config.WindowSize, + _lastQueryTime, + DateTime.UtcNow, sensors); + + if (datas.Count < config.WindowSize) + return measurements; + + datas = [.. datas.OrderBy(d => d.TimeStamp)]; + + int fullBlockCount = datas.Count / config.WindowSize; + + // Slice the data into windows of size config.WindowSize + for (int i = 0; i < fullBlockCount; i++) + { + List block = datas.GetRange(i * config.WindowSize * config.InputSensors.Count, config.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 * config.WindowSize) - 1; + _lastQueryTime = datas[lastProcessedIndex].TimeStamp; } + + Log.Debug("Fetched {Count} data blocks for machine {MachineName}", measurements.Count, config.MachineName); + + // TODO CJS : REMOVE + Log.Debug("First element timestamp: {Timestamp}", measurements.First().First().TimeStamp); + Log.Debug("Last element timestamp: {Timestamp}", measurements.Last().Last().TimeStamp); + + return measurements; } #endregion #region Private fields private List? _pipelineProcessors; private string? _lastEndpoint; + private DateTime _lastQueryTime = DateTime.MinValue; #endregion } diff --git a/src/DataAggregator.Processor/appsettings.json b/src/DataAggregator.Processor/appsettings.json index 19ac701..b418184 100644 --- a/src/DataAggregator.Processor/appsettings.json +++ b/src/DataAggregator.Processor/appsettings.json @@ -34,7 +34,7 @@ ], "WindowSize": 50, "WindowSizeInSeconds": false, - "CycleIntervalSeconds": 0.5, + "CycleIntervalSeconds": 1, "ProcessingPipeline": [ { "Name": "actuatorcurrent", From 11b9197098aab0a6f48b637054bd7ec29a850df5 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Thu, 31 Jul 2025 10:01:30 +0200 Subject: [PATCH 08/13] feat: set result timestamp to first sample timestamp --- .../ActuatorCurrentFeatureExtractor.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs index 10ceb2b..25fcc24 100644 --- a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs @@ -31,8 +31,8 @@ public Task> ProcessAsync(IEnumerable x.TimeStamp); else meanTime = DateTime.UtcNow; From d8ae2f9250c73c470ae50fae927e78796410fe57 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Thu, 31 Jul 2025 10:04:31 +0200 Subject: [PATCH 09/13] clean: clean window slicing --- .../Prediction/MachinePredictionProcessor.cs | 16 ++++++++-------- .../StateDeductionPostProcessor.cs | 13 +------------ 2 files changed, 9 insertions(+), 20 deletions(-) diff --git a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs index c0d31ed..8a3981f 100644 --- a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs +++ b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs @@ -161,9 +161,6 @@ private async Task>> FetchDataWindow { List> measurements = []; - // If we want to fetch the last N measurements, we must take inaccount the number of sensors - int windowSize = config.WindowSize * config.InputSensors.Count; - // If the window size is in secondes, simply query the measurements for the last N seconds. if (config.WindowSizeInSeconds) { @@ -180,6 +177,9 @@ private async Task>> FetchDataWindow } else { + // 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) { @@ -193,23 +193,23 @@ private async Task>> FetchDataWindow DateTime.UtcNow, sensors); - if (datas.Count < config.WindowSize) + if (datas.Count < windowSize) return measurements; datas = [.. datas.OrderBy(d => d.TimeStamp)]; - int fullBlockCount = datas.Count / config.WindowSize; + int fullBlockCount = datas.Count / windowSize; - // Slice the data into windows of size config.WindowSize + // Slice the data into windows of size windowSize. for (int i = 0; i < fullBlockCount; i++) { - List block = datas.GetRange(i * config.WindowSize * config.InputSensors.Count, config.WindowSize); + 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 * config.WindowSize) - 1; + int lastProcessedIndex = (fullBlockCount * windowSize) - 1; _lastQueryTime = datas[lastProcessedIndex].TimeStamp; } diff --git a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs index 0b2d757..2302c37 100644 --- a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs @@ -92,19 +92,8 @@ private Task> ConfirmPendingState() private IEnumerable OverrideStateInBatch(List batch, string forcedValue) => batch.Select( x => x.SensorName == _resultOutputName - ? new MeasurementDataWrapper(x, forcedValue) + ? new MeasurementData(x.TimeStamp, x.SensorName, forcedValue) : x); - - private class MeasurementDataWrapper(IMeasurementData original, string forcedValue) : IMeasurementData - { - public DateTime TimeStamp => original.TimeStamp; - - public string SensorName => original.SensorName; - - public Type ValueType => typeof(string); - - public object GetRawValue() => forcedValue; - } #endregion #region private fields From 4c24f5894d4af1ea1b9ea9a443c05634ac48e7b4 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Thu, 31 Jul 2025 10:04:56 +0200 Subject: [PATCH 10/13] feat: adapt settings for model using 30 sample window width --- src/DataAggregator.Processor/appsettings.json | 30 +++++++++---------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/src/DataAggregator.Processor/appsettings.json b/src/DataAggregator.Processor/appsettings.json index b418184..f5429fc 100644 --- a/src/DataAggregator.Processor/appsettings.json +++ b/src/DataAggregator.Processor/appsettings.json @@ -32,7 +32,7 @@ "current-amp-c", "current-amp-s" ], - "WindowSize": 50, + "WindowSize": 30, "WindowSizeInSeconds": false, "CycleIntervalSeconds": 1, "ProcessingPipeline": [ @@ -41,20 +41,20 @@ "Configuration": { "EnableZScoreNormalization": true, "NormalizationParameters": { - "GlobalActivityRatio": [ 0.004067, 0.018027 ], - "GlobalChangeDensity": [ 0.433931, 0.146315 ], - "InterAxisMeanCorrelation": [ -0.006368, 0.068250 ], - "InterAxisMaxCorrelation": [ 0.432673, 0.243142 ], - "InterAxisCorrelationVariance": [ 0.224938, 0.112596 ], - "AxisSynchronization": [ -60.031723, 818.528442 ], - "AxisLoadBalance": [ -0.044534, 0.330180 ], - "TemporalStability": [ 0.964727, 0.038193 ], - "GlobalSkewness": [ -0.119681, 0.371627 ], - "GlobalKurtosis": [ -0.568834, 0.398853 ], - "GlobalTrendSlope": [ 0.000011, 0.000304 ], - "CoefficientOfVariation": [ -11.505261, 823.236694 ], - "NormalizedIqrMedian": [ -22.622700, 1259.548584 ], - "NormalizedIqrMean": [ -6.469841, 807.403015 ] + "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 ] } } }, From 59a45573c5aa168a938057f3fbccbb03f5e92de1 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Thu, 31 Jul 2025 14:38:11 +0200 Subject: [PATCH 11/13] fix: state deduction --- .../StateDeductionPostProcessor.cs | 83 +++++++++---------- 1 file changed, 40 insertions(+), 43 deletions(-) diff --git a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs index 2302c37..2a2d67e 100644 --- a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs @@ -22,61 +22,42 @@ public Task> ProcessAsync(IEnumerable>(inputList); - HandlePotentialStateChange(currentPredictedState, inputList); - - if (_stableCount >= config.Threshold) - return ConfirmPendingState(); - - return Task.FromResult(Enumerable.Empty()); - } - #endregion - - #region Private Methods - private Task> InitializeFirstState(List inputList, string state) - { - _lastState = state; - _pendingState = state; - return Task.FromResult>(inputList); - } - - private Task> HandleSameAsLastState(List inputList, string state) - { - if (_bufferedBatches.Count > 0) - { - var corrected = _bufferedBatches - .SelectMany(batch => OverrideStateInBatch(batch, _lastState!)) - .ToList(); - - _bufferedBatches.Clear(); - _pendingState = state; - _stableCount = 0; - - return Task.FromResult>(corrected); - } - - _pendingState = state; - _stableCount = 0; - return Task.FromResult>(inputList); - } - - private void HandlePotentialStateChange(string currentState, List inputList) - { - if (currentState == _pendingState) + if (currentPredictedState == _pendingState) { _stableCount++; _bufferedBatches.Add(inputList); + + if (_stableCount == config.Threshold) + return ConfirmPendingState(); + + return Task.FromResult(Enumerable.Empty()); } else { - _pendingState = currentState; + 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; @@ -89,19 +70,35 @@ private Task> ConfirmPendingState() 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 + #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 } From 60553cc812253b0bf245bb2b65f5113e73fd159b Mon Sep 17 00:00:00 2001 From: CoJaques Date: Tue, 5 Aug 2025 09:56:59 +0200 Subject: [PATCH 12/13] feat: remove unused logs and update configuration --- .../Services/Prediction/MachinePredictionProcessor.cs | 4 ---- src/DataAggregator.Processor/appsettings.json | 2 +- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs index 8a3981f..79e2bf4 100644 --- a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs +++ b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs @@ -215,10 +215,6 @@ private async Task>> FetchDataWindow Log.Debug("Fetched {Count} data blocks for machine {MachineName}", measurements.Count, config.MachineName); - // TODO CJS : REMOVE - Log.Debug("First element timestamp: {Timestamp}", measurements.First().First().TimeStamp); - Log.Debug("Last element timestamp: {Timestamp}", measurements.Last().Last().TimeStamp); - return measurements; } #endregion diff --git a/src/DataAggregator.Processor/appsettings.json b/src/DataAggregator.Processor/appsettings.json index f5429fc..d08c2b4 100644 --- a/src/DataAggregator.Processor/appsettings.json +++ b/src/DataAggregator.Processor/appsettings.json @@ -67,7 +67,7 @@ { "Name": "statedeductionpostprocessor", "Configuration": { - "Threshold": 2 + "Threshold": 3 } } ] From dc46cbfefeff1d02d79d9fd8c534a702844a0445 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Tue, 5 Aug 2025 10:00:47 +0200 Subject: [PATCH 13/13] feat: remove integration tests and update MachinePrediction behaviour when no results --- DataAggregator.sln | 15 ---------- .../Prediction/MachinePredictionProcessor.cs | 2 +- .../DataAggregator.Integration.Tests.csproj | 30 ------------------- .../UnitTest1.cs | 15 ---------- 4 files changed, 1 insertion(+), 61 deletions(-) delete mode 100644 tests/DataAggregator.Integration.Tests/DataAggregator.Integration.Tests.csproj delete mode 100644 tests/DataAggregator.Integration.Tests/UnitTest1.cs diff --git a/DataAggregator.sln b/DataAggregator.sln index 371bf54..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 @@ -112,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 @@ -182,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.Processor/Services/Prediction/MachinePredictionProcessor.cs b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs index 79e2bf4..b8ff3d7 100644 --- a/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs +++ b/src/DataAggregator.Processor/Services/Prediction/MachinePredictionProcessor.cs @@ -44,7 +44,7 @@ public async Task ProcessAsync(MachinePredictionConfig config) } IEnumerable? processedData = await RunPipelineAsync(measurements, config.MachineName); - if (processedData == null) return; + if (processedData == null || !processedData.Any()) return; await influxRepository.WriteMeasurementAsync(config.MachineName, processedData); Log.Information("Prediction pipeline completed for machine {MachineName}", config.MachineName); 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() - { - } -}