From 46ee09603f72293da7323639e6e28eb3c6d8f15b Mon Sep 17 00:00:00 2001 From: CoJaques Date: Mon, 11 Aug 2025 21:31:50 +0200 Subject: [PATCH 1/9] feat: delete useless feature --- .../Services/PredictionBackgroundService.cs | 41 +++++++++++++++-- .../Factory/DataProcessorFactory.cs | 1 - .../Processing/Onnx/OnnxPredictionEngine.cs | 3 +- .../StateDeductionPostProcessor.cs | 8 ++-- .../ActuatorCurrentFeatureExtractor.cs | 45 +++---------------- .../MathUtils.cs | 26 ----------- src/DataAggregator.Processor/appsettings.json | 2 - .../ActuatorCurrentFeatureExtractorTests.cs | 16 +++---- .../DataProcessorFactoryTests.cs | 1 - .../Services/PreProcessing/MathUtilsTests.cs | 44 ------------------ .../Prediction/OnnxPredictionEngineTests.cs | 5 --- 11 files changed, 56 insertions(+), 136 deletions(-) diff --git a/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs b/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs index 0406eb6..eadf91a 100644 --- a/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs +++ b/src/DataAggregator.Processor/Services/PredictionBackgroundService.cs @@ -20,6 +20,8 @@ public class PredictionBackgroundService( private readonly Dictionary _machineTimers = []; private readonly Dictionary _machineLocks = []; + private readonly Dictionary _machineScopes = []; + private readonly Dictionary _machineProcessors = []; #endregion @@ -74,6 +76,22 @@ public override async Task StopAsync(CancellationToken cancellationToken) _machineTimers.Clear(); + // Dispose per-machine scopes (which own the scoped services, including processors and repositories) + foreach ((string _, IServiceScope scope) in _machineScopes) + { + try + { + scope.Dispose(); + } + catch (Exception ex) + { + Log.Warning(ex, "Error disposing scope for a machine"); + } + } + + _machineScopes.Clear(); + _machineProcessors.Clear(); + await base.StopAsync(cancellationToken); } @@ -135,11 +153,26 @@ private void ScheduleMachine(MachinePredictionConfig machineConfig) _machineLocks[machineConfig.MachineName] = new SemaphoreSlim(1, 1); } - using IServiceScope scope = serviceProvider.CreateScope(); - IMachinePredictionProcessor predictionProcessor - = scope.ServiceProvider.GetRequiredService(); + // Create and retain a dedicated DI scope per machine to keep stateful services alive across ticks + IServiceScope scope = serviceProvider.CreateScope(); + IMachinePredictionProcessor predictionProcessor = + scope.ServiceProvider.GetRequiredService(); - var timer = new Timer(async _ => await ProcessMachineAsync(machineConfig, predictionProcessor), null, TimeSpan.Zero, interval); + _machineScopes[machineConfig.MachineName] = scope; + _machineProcessors[machineConfig.MachineName] = predictionProcessor; + + var timer = new Timer( + async _ => + { + // Retrieve the dedicated processor for this machine + if (_machineProcessors.TryGetValue(machineConfig.MachineName, out IMachinePredictionProcessor? proc)) + { + await ProcessMachineAsync(machineConfig, proc); + } + }, + null, + TimeSpan.Zero, + interval); _machineTimers[machineConfig.MachineName] = timer; diff --git a/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs index 4e0a97a..681ae13 100644 --- a/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs +++ b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs @@ -1,4 +1,3 @@ -using DataAggregator.Processor.Services.Prediction; using DataAggregator.Processor.Services.Processing.Abstraction; using DataAggregator.Processor.Services.Processing.Onnx; using DataAggregator.Processor.Services.Processing.PostProcessing.StateDeductionPostProcess; diff --git a/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs b/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs index ddb36b1..7d58276 100644 --- a/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs +++ b/src/DataAggregator.Processor/Services/Processing/Onnx/OnnxPredictionEngine.cs @@ -1,11 +1,10 @@ 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; -namespace DataAggregator.Processor.Services.Prediction; +namespace DataAggregator.Processor.Services.Processing.Onnx; /// /// Implementation of ONNX prediction engine. diff --git a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs index 2a2d67e..9e8fe6d 100644 --- a/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PostProcessing/StateDeductionPostProcess/StateDeductionPostProcessor.cs @@ -19,7 +19,10 @@ public Task> ProcessAsync(IEnumerable>(inputList); + } if (currentPredictedState == _lastState) return Task.FromResult>(inputList); @@ -51,11 +54,10 @@ public Task> ProcessAsync(IEnumerable> InitializeFirstState(List inputList, string state) + private void InitializeFirstState(string state) { _lastState = state; _pendingState = state; - return Task.FromResult>(inputList); } private Task> ConfirmPendingState() diff --git a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs index 25fcc24..06594bc 100644 --- a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs @@ -47,11 +47,9 @@ public Task> ProcessAsync(IEnumerable(meanTime, "AxisLoadBalance", normalizedFeatures[6]), new MeasurementData(meanTime, "TemporalStability", normalizedFeatures[7]), new MeasurementData(meanTime, "GlobalSkewness", normalizedFeatures[8]), - new MeasurementData(meanTime, "GlobalKurtosis", normalizedFeatures[9]), - new MeasurementData(meanTime, "GlobalTrendSlope", normalizedFeatures[10]), - new MeasurementData(meanTime, "CoefficientOfVariation", normalizedFeatures[11]), - new MeasurementData(meanTime, "NormalizedIqrMedian", normalizedFeatures[12]), - new MeasurementData(meanTime, "NormalizedIqrMean", normalizedFeatures[13]), + new MeasurementData(meanTime, "CoefficientOfVariation", normalizedFeatures[9]), + new MeasurementData(meanTime, "NormalizedIqrMedian", normalizedFeatures[10]), + new MeasurementData(meanTime, "NormalizedIqrMean", normalizedFeatures[11]), new MeasurementData(meanTime, "Label", string.Empty), }; @@ -106,45 +104,31 @@ private float[] ExtractFeatures(IEnumerable measurements) float globalQ75 = MathUtils.Percentile(allCurrents, 75); float globalIqr = globalQ75 - globalQ25; - // Feature 1: Global Activity Ratio float activityThreshold = globalStd * 2; float activeRatio = allCurrents.Count(x => Math.Abs(x) > activityThreshold) / (float)allCurrents.Count; - // Feature 2: Global Change Density var diffSignals = allCurrents.Zip(allCurrents.Skip(1), (a, b) => Math.Abs(b - a)).ToList(); int significantChanges = diffSignals.Count(x => x > globalStd * 1.5); float changeDensity = significantChanges / (float)allCurrents.Count; - // Extract actuator currents for correlation analysis List> actuatorsCurrents = ExtractActuatorCurrents(measurements); - // Features 3-5: Inter-actuator correlations List correlations = CalculateInterActuatorCorrelations(actuatorsCurrents); float meanCorrelation = correlations.Count > 0 ? correlations.Average() : 0f; float maxCorrelation = correlations.Count > 0 ? correlations.Max() : 0f; float correlationVariance = correlations.Count > 0 ? MathUtils.StandardDeviation(correlations) : 0f; - // Feature 6: Actuator Synchronization var actuatorsMeans = actuatorsCurrents.Select(MathUtils.Mean).ToList(); float meanOfMeans = actuatorsMeans.Average(); float synchronization = meanOfMeans != 0 ? 1 - (MathUtils.StandardDeviation(actuatorsMeans) / Math.Abs(meanOfMeans)) : 1f; - // Feature 7: Actuator Load Balance var actuatorsEnergies = actuatorsCurrents.Select(actuator => actuator.Sum(x => x * x)).ToList(); float meanEnergy = actuatorsEnergies.Average(); float loadBalance = meanEnergy != 0 ? 1 - (MathUtils.StandardDeviation(actuatorsEnergies) / meanEnergy) : 1f; - - // Feature 8: Temporal Stability float temporalStability = CalculateTemporalStability(allCurrents); - // Features 9-10: Global Skewness and Kurtosis float globalSkewness = MathUtils.Skewness(allCurrents); - float globalKurtosis = MathUtils.Kurtosis(allCurrents); - - // Feature 11: Global Trend Slope - float trendSlope = CalculateTrendSlope(allCurrents); - // Features 12-14: Normalized coefficients float coeffVar = Math.Abs(globalMean) > 1e-8f ? globalStd / globalMean : 0f; float normIqrMedian = Math.Abs(globalMedian) > 1e-8f ? globalIqr / globalMedian : 0f; float normIqrMean = Math.Abs(globalMean) > 1e-8f ? globalIqr / globalMean : 0f; @@ -160,8 +144,6 @@ private float[] ExtractFeatures(IEnumerable measurements) loadBalance, temporalStability, globalSkewness, - globalKurtosis, - trendSlope, coeffVar, normIqrMedian, normIqrMean, @@ -254,23 +236,6 @@ private float CalculateTemporalStability(List allCurrents) : 1f; } - private float CalculateTrendSlope(List allCurrents) - { - if (allCurrents.Count < 2) - { - return 0f; - } - - var timeIndices = Enumerable.Range(0, allCurrents.Count).Select(x => (float)x).ToList(); - float meanTime = timeIndices.Average(); - float meanCurrent = allCurrents.Average(); - - float numerator = timeIndices.Zip(allCurrents, (t, c) => (t - meanTime) * (c - meanCurrent)).Sum(); - float denominator = timeIndices.Sum(t => (t - meanTime) * (t - meanTime)); - - return denominator != 0 ? numerator / denominator : 0f; - } - private float[] NormalizeFeaturesAsync(float[] features, PreprocessingConfig preprocessing) { if (!preprocessing.EnableZScoreNormalization) @@ -282,8 +247,8 @@ private float[] NormalizeFeaturesAsync(float[] features, PreprocessingConfig pre [ "GlobalActivityRatio", "GlobalChangeDensity", "InterAxisMeanCorrelation", "InterAxisMaxCorrelation", "InterAxisCorrelationVariance", "AxisSynchronization", - "AxisLoadBalance", "TemporalStability", "GlobalSkewness", "GlobalKurtosis", - "GlobalTrendSlope", "CoefficientOfVariation", "NormalizedIqrMedian", "NormalizedIqrMean", + "AxisLoadBalance", "TemporalStability", "GlobalSkewness", + "CoefficientOfVariation", "NormalizedIqrMedian", "NormalizedIqrMean", ]; float[] normalized = new float[features.Length]; diff --git a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs index cbe9ce8..a810516 100644 --- a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/MathUtils.cs @@ -84,32 +84,6 @@ public static float Skewness(IEnumerable values) return (float)skew; } - /// - /// Calculates the kurtosis of a collection of values. - /// - /// Collection of float values. - /// Kurtosis value. - public static float Kurtosis(IEnumerable values) - { - if (values == null || values.Count() < 4) - { - return 0.0f; - } - - var valuesList = values.ToList(); - float mean = valuesList.Average(); - float std = StandardDeviation(valuesList); - - if (std == 0) - { - return 0.0f; - } - - double kurt = valuesList.Select(x => Math.Pow((x - mean) / std, 4)).Average() - 3; - - return (float)kurt; - } - /// /// Calculates the correlation coefficient between two collections of values. /// diff --git a/src/DataAggregator.Processor/appsettings.json b/src/DataAggregator.Processor/appsettings.json index 797729f..a0d76f8 100644 --- a/src/DataAggregator.Processor/appsettings.json +++ b/src/DataAggregator.Processor/appsettings.json @@ -55,8 +55,6 @@ "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 ] diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs index ceb8a75..f939ed6 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs @@ -11,22 +11,22 @@ public ActuatorCurrentFeatureExtractorTests() => _featureExtractor = new ActuatorCurrentFeatureExtractor(CreateValidPreprocessingConfig()); [Fact] - public async Task ProcessAsync_ShouldReturnFifteenFeatures_WhenValidDataProvided() + public async Task ProcessAsync_ShouldReturnThirteenFeatures_WhenValidDataProvided() { var measurements = CreateTestMeasurements(); var result = await _featureExtractor.ProcessAsync(measurements); Assert.NotNull(result); - Assert.Equal(15, result.Count()); + Assert.Equal(13, result.Count()); } [Fact] - public async Task ProcessAsync_ShouldReturnFourteenFeatures_WhenEmptyMeasurementsProvided() + public async Task ProcessAsync_ShouldReturnTwelvesFeatures_WhenEmptyMeasurementsProvided() { var measurements = new List(); var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); Assert.NotNull(result); - Assert.Equal(14, result.Count()); + Assert.Equal(12, result.Count()); Assert.All(result, feature => Assert.Equal(0.0f, (float)feature.GetRawValue())); } @@ -38,7 +38,7 @@ public async Task ProcessAsync_ShouldReturnValidFeatures_WhenValidDataProvided() var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); Assert.NotNull(result); - Assert.Equal(14, result.Count()); + Assert.Equal(12, result.Count()); Assert.All(result, feature => Assert.False(float.IsNaN((float)feature.GetRawValue()))); Assert.All(result, feature => Assert.False(float.IsInfinity((float)feature.GetRawValue()))); } @@ -55,7 +55,7 @@ public async Task ProcessAsync_ShouldReturnZeroFeatures_WhenNoValidValuesFound() var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); Assert.NotNull(result); - Assert.Equal(14, result.Count()); + Assert.Equal(12, result.Count()); Assert.All(result, feature => Assert.Equal(0.0f, (float)feature.GetRawValue())); } @@ -69,7 +69,7 @@ public async Task ProcessAsync_ShouldHandleSingleValue_WhenOnlyOneValidMeasureme var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); Assert.NotNull(result); - Assert.Equal(14, result.Count()); + Assert.Equal(12, result.Count()); Assert.All(result, feature => Assert.False(float.IsNaN((float)feature.GetRawValue()))); } @@ -92,7 +92,7 @@ public async Task ProcessAsync_ShouldHandleLargeDataset_WhenManyMeasurementsProv var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); Assert.NotNull(result); - Assert.Equal(14, result.Count()); + Assert.Equal(12, result.Count()); Assert.All(result, feature => Assert.False(float.IsNaN((float)feature.GetRawValue()))); Assert.All(result, feature => Assert.False(float.IsInfinity((float)feature.GetRawValue()))); } diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs index e4cbe7b..cd95cb5 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs @@ -1,7 +1,6 @@ using DataAggregator.Processor.Configuration; using DataAggregator.Processor.Services.Processing.Factory; using DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; -using DataAggregator.Processor.Services.Prediction; using DataAggregator.Processor.Services.Processing.PostProcessing.StateDeductionPostProcess; using System.Collections.Generic; using DataAggregator.Processor.Services.Processing.Onnx; diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/MathUtilsTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/MathUtilsTests.cs index 6d86095..314e27c 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/MathUtilsTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/MathUtilsTests.cs @@ -247,50 +247,6 @@ public void Skewness_ShouldReturnPositiveValue_WhenValuesAreRightSkewed() #endregion - #region Kurtosis tests - - [Fact] - public void Kurtosis_ShouldReturnZero_WhenValuesContainsLessThan4Elements() - { - // Arrange - var values = new List { 1.0f, 2.0f, 3.0f }; - - // Act - float result = MathUtils.Kurtosis(values); - - // Assert - Assert.Equal(0.0f, result); - } - - [Fact] - public void Kurtosis_ShouldReturnZero_WhenStandardDeviationIsZero() - { - // Arrange - var values = new List { 3.0f, 3.0f, 3.0f, 3.0f }; - - // Act - float result = MathUtils.Kurtosis(values); - - // Assert - Assert.Equal(0.0f, result); - } - - [Fact] - public void Kurtosis_ShouldReturnCorrectValue_WhenValuesAreNormallyDistributed() - { - // Arrange - var values = new List { 1.0f, 2.0f, 3.0f, 4.0f, 5.0f }; - - // Act - float result = MathUtils.Kurtosis(values); - - // Assert - // For normal distribution, kurtosis should be close to 0 (excess kurtosis) - Assert.Equal(-1.3f, result, 1); - } - - #endregion - #region Correlation tests [Fact] diff --git a/tests/DataAggregator.Processor.Tests/Services/Prediction/OnnxPredictionEngineTests.cs b/tests/DataAggregator.Processor.Tests/Services/Prediction/OnnxPredictionEngineTests.cs index 02733b4..ecb216c 100644 --- a/tests/DataAggregator.Processor.Tests/Services/Prediction/OnnxPredictionEngineTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/Prediction/OnnxPredictionEngineTests.cs @@ -1,9 +1,4 @@ -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; From e120bf737337b01d4ee84943d910154c4fa51547 Mon Sep 17 00:00:00 2001 From: Colin Jaques Date: Tue, 12 Aug 2025 10:48:42 +0200 Subject: [PATCH 2/9] feat: implement file collector used for simulation --- .../FileConnectorConfiguration.cs | 19 ++ .../Connector/FileConnector.cs | 188 ++++++++++++++++++ ...aAggregator.Collector.FileCollector.csproj | 21 ++ .../configuration-example.json | 108 ++++++++++ DataAggregator.sln | 15 ++ ....Collector.OpenCNCapnProtoConnector.csproj | 8 + .../configuration-example.json | 113 +++++++++++ .../DataAggregator.Collector.App.csproj | 1 + src/DataAggregator.Collector/Program.cs | 12 ++ src/DataAggregator.Collector/appsettings.json | 29 ++- 10 files changed, 497 insertions(+), 17 deletions(-) create mode 100644 DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs create mode 100644 DataAggregator.Collector.FileCollector/Connector/FileConnector.cs create mode 100644 DataAggregator.Collector.FileCollector/DataAggregator.Collector.FileCollector.csproj create mode 100644 DataAggregator.Collector.FileCollector/configuration-example.json create mode 100644 src/DataAggregator.Collector.OpenCNCapnProtoConnector/configuration-example.json diff --git a/DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs b/DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs new file mode 100644 index 0000000..3385c73 --- /dev/null +++ b/DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs @@ -0,0 +1,19 @@ +using DataAggregator.Collector.Shared.Abstraction.Configuration; + +namespace DataAggregator.Collector.FileCollector.Configuration; + +/// +/// Specifies the configuration for the File Connector. +/// +public class FileConnectorConfiguration : CollectorConfiguration +{ + /// + /// Gets or sets the list of file paths to read data from. + /// + public List Files { get; set; } = []; + + /// + /// Gets or sets the sampling rate (in Hz) for data provision. + /// + public double SamplingRate { get; set; } = 100; +} diff --git a/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs b/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs new file mode 100644 index 0000000..52cb631 --- /dev/null +++ b/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs @@ -0,0 +1,188 @@ +using System.Globalization; +using DataAggregator.Collector.FileCollector.Configuration; +using DataAggregator.Collector.Shared.Abstraction; +using DataAggregator.Collector.Shared.Abstraction.Configuration; +using DataAggregator.Collector.Shared.Models; +using DataAggregator.Shared.Domain.DataType; + +namespace DataAggregator.Collector.FileCollector.Connector; + +/// +/// Define a connector for file-based data sources, used to read data from files. +/// Useful for testing or when data is stored in files. +/// +public class FileConnector(FileConnectorConfiguration config) : IDataSourceConnector +{ + #region Private Fields + private List _readers = []; + private List _sensorNames = []; + private int _currentFileIndex = 0; + private DateTime _lastFetchTime = DateTime.MinValue; + private bool _initialized = false; + #endregion + + /// + /// Initializes the connector by opening the files and reading the sensor headers. + /// Files are read line by line on demand, without loading the entire content into memory. + /// + /// Task.CompletedTask. + public Task ConnectAsync() + { + DisposeReaders(); + _readers = []; + _sensorNames = []; + _currentFileIndex = 0; + _lastFetchTime = DateTime.MinValue; + _initialized = false; + + foreach (string file in config.Files) + { + if (!File.Exists(file)) + continue; + + var reader = new StreamReader(file); + string? header = reader.ReadLine(); + if (header == null) + { + reader.Dispose(); + continue; + } + + if (_sensorNames.Count == 0) + _sensorNames.AddRange(header.Split(',')); + + _readers.Add(reader); + } + + _initialized = _readers.Count > 0 && _sensorNames.Count > 0; + return Task.CompletedTask; + } + + /// + /// Closes and disposes all open files, and resets the connector state. + /// + /// Task.CompletedTask. + public Task DisconnectAsync() + { + DisposeReaders(); + _readers = []; + _sensorNames = []; + _currentFileIndex = 0; + _lastFetchTime = DateTime.MinValue; + _initialized = false; + + return Task.CompletedTask; + } + + /// + /// Reads data from the files line by line, respecting the configured sampling rate (SamplingRate). + /// Returns a batch of N lines, where N depends on the elapsed time since the last call and the frequency. + /// Timestamps are retroactive: the last data point is timestamped at "now", + /// previous ones are spaced according to the frequency. + /// Files are read on demand and automatically loop when reaching the end. + /// + /// A batch of sensor data, with correct timestamps. + public Task> FetchDataAsync() + { + if (!_initialized) + return Task.FromResult(Enumerable.Empty()); + + double samplingRate = config.SamplingRate > 0 ? config.SamplingRate : 1.0; + double intervalMs = 1000.0 / samplingRate; + + DateTime now = DateTime.UtcNow; + if (_lastFetchTime == DateTime.MinValue) + _lastFetchTime = now; + + double elapsedMs = (now - _lastFetchTime).TotalMilliseconds; + int linesToProvide = (int)(elapsedMs / intervalMs); + + if (linesToProvide <= 0) + return Task.FromResult>([]); + + List result = []; + + // Calculate timestamps for each line: oldest = now - (linesToProvide-1)*interval, newest = now + for (int i = 0; i < linesToProvide; i++) + { + string? line = null; + int attempts = 0; + while (attempts < _readers.Count) + { + if (_readers[_currentFileIndex].EndOfStream) + { + _readers[_currentFileIndex].DiscardBufferedData(); + _readers[_currentFileIndex].BaseStream.Seek(0, SeekOrigin.Begin); + _readers[_currentFileIndex].ReadLine(); // skip header + } + + line = _readers[_currentFileIndex].ReadLine(); + _currentFileIndex = (_currentFileIndex + 1) % _readers.Count; + + if (!string.IsNullOrWhiteSpace(line)) + break; + + attempts++; + } + + if (string.IsNullOrWhiteSpace(line)) + continue; + + string[] row = line.Split(','); + + // Retro-timestamping: oldest = now - (linesToProvide-1)*interval, newest = now + DateTime timestamp = now - TimeSpan.FromMilliseconds(intervalMs * (linesToProvide - 1 - i)); + foreach (SensorConfig sensor in config.Sensors) + { + int colIdx = _sensorNames.FindIndex(n => n == sensor.Name); + + if (colIdx == -1 || colIdx >= row.Length) + continue; + + string valueStr = row[colIdx]; + + IMeasurementData? measurement = sensor.DataType switch + { + SensorDataType.Boolean => new MeasurementData(timestamp, sensor.Name, float.Parse(valueStr, CultureInfo.InvariantCulture) != 0), + SensorDataType.Integer => new MeasurementData(timestamp, sensor.Name, (int)float.Parse(valueStr, CultureInfo.InvariantCulture)), + SensorDataType.Double or SensorDataType.Float => new MeasurementData(timestamp, sensor.Name, double.Parse(valueStr, CultureInfo.InvariantCulture)), + SensorDataType.String => new MeasurementData(timestamp, sensor.Name, valueStr), + _ => null, + }; + + if (measurement != null) + result.Add(measurement); + } + } + + _lastFetchTime = _lastFetchTime.AddMilliseconds(linesToProvide * intervalMs); + return Task.FromResult>(result); + } + + /// + /// Checks that all files specified in the configuration exist and are accessible. + /// + /// True if all files exist, false otherwise. + public Task IsConnectedAsync() + { + bool isConnected = true; + foreach (string filePath in config.Files) + { + if (!File.Exists(filePath)) + { + isConnected = false; + break; + } + } + + return Task.FromResult(isConnected); + } + + private void DisposeReaders() + { + foreach (StreamReader reader in _readers) + { + reader.Dispose(); + } + } +} diff --git a/DataAggregator.Collector.FileCollector/DataAggregator.Collector.FileCollector.csproj b/DataAggregator.Collector.FileCollector/DataAggregator.Collector.FileCollector.csproj new file mode 100644 index 0000000..7830542 --- /dev/null +++ b/DataAggregator.Collector.FileCollector/DataAggregator.Collector.FileCollector.csproj @@ -0,0 +1,21 @@ + + + + net9.0 + enable + enable + + + + + PreserveNewest + true + PreserveNewest + + + + + + + + diff --git a/DataAggregator.Collector.FileCollector/configuration-example.json b/DataAggregator.Collector.FileCollector/configuration-example.json new file mode 100644 index 0000000..39f92dd --- /dev/null +++ b/DataAggregator.Collector.FileCollector/configuration-example.json @@ -0,0 +1,108 @@ +{ + "AllowedHosts": "*", + + "Serilog": { + "MinimumLevel": { + "Default": "Information", + "Override": { + "Microsoft.AspNetCore": "Warning", + "System": "Warning" + } + }, + "WriteTo": [ + { + "Name": "Console", + "Args": { + "outputTemplate": "[{Timestamp:HH:mm:ss} {Level:u3}] {ServiceName}: {Message:lj}{NewLine}{Exception}" + } + } + ], + "Enrich": [ "FromLogContext" ], + "Properties": { + "ServiceName": "Collector" + } + }, + + "CollectorType": "File", + "Collector": { + "DeviceName": "Micro5", + "Location": "Test", + "HealthCheckEndpoint": "http://localhost:5000/health", + "FlushIntervalMilliseconds": 1000, + "SamplingRate": 100, + "RegistrationService": { + "BaseUrl": "http://localhost:5137", + "Endpoint": "api/DeviceRegistration/register" + }, + "BufferSettings": { + "MaxBufferSize": 10000 + }, + "Sensors": [ + { + "Name": "current-amp-x", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float" + }, + { + "Name": "current-amp-y", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float" + }, + { + "Name": "current-amp-z", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float" + }, + { + "Name": "current-amp-b", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float" + }, + { + "Name": "current-amp-c", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float" + }, + { + "Name": "current-amp-s", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float" + } + ], + "Files": [ + "file1.txt", + "file2.txt", + "file3.txt" + ] + } +} diff --git a/DataAggregator.sln b/DataAggregator.sln index 7c3b036..a20751f 100644 --- a/DataAggregator.sln +++ b/DataAggregator.sln @@ -40,6 +40,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "processor", "processor", "{ EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Processor.Tests", "tests\DataAggregator.Processor.Tests\DataAggregator.Processor.Tests.csproj", "{EFE47FE6-F41E-CDD6-0991-472080AD88B0}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Collector.FileCollector", "DataAggregator.Collector.FileCollector\DataAggregator.Collector.FileCollector.csproj", "{D72B893B-E69F-44A4-8747-6967CF3BA3BE}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -158,6 +160,18 @@ Global {EFE47FE6-F41E-CDD6-0991-472080AD88B0}.Release|x64.Build.0 = Release|x64 {EFE47FE6-F41E-CDD6-0991-472080AD88B0}.Release|x86.ActiveCfg = Release|x86 {EFE47FE6-F41E-CDD6-0991-472080AD88B0}.Release|x86.Build.0 = Release|x86 + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|Any CPU.Build.0 = Debug|Any CPU + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|x64.ActiveCfg = Debug|x64 + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|x64.Build.0 = Debug|x64 + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|x86.ActiveCfg = Debug|x86 + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|x86.Build.0 = Debug|x86 + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|Any CPU.ActiveCfg = Release|Any CPU + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|Any CPU.Build.0 = Release|Any CPU + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|x64.ActiveCfg = Release|x64 + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|x64.Build.0 = Release|x64 + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|x86.ActiveCfg = Release|x86 + {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|x86.Build.0 = Release|x86 EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -176,6 +190,7 @@ Global {039EC00D-5EDF-4C48-B449-29CFB6750232} = {EB9A5576-3E00-4007-8C72-2235E0A1546D} {EB9A5576-3E00-4007-8C72-2235E0A1546D} = {E1AD9667-4C40-4CAF-8096-5FA749EBB2B1} {EFE47FE6-F41E-CDD6-0991-472080AD88B0} = {247EF7A2-1DFD-4B51-AC7D-0FD13827CAC2} + {D72B893B-E69F-44A4-8747-6967CF3BA3BE} = {F5250AC4-9CCC-432B-8725-DAC6AE01CCF0} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {9C8E6DBA-9F77-4FD0-9A1D-25150F7806FF} diff --git a/src/DataAggregator.Collector.OpenCNCapnProtoConnector/DataAggregator.Collector.OpenCNCapnProtoConnector.csproj b/src/DataAggregator.Collector.OpenCNCapnProtoConnector/DataAggregator.Collector.OpenCNCapnProtoConnector.csproj index 9d101c0..ab783b2 100644 --- a/src/DataAggregator.Collector.OpenCNCapnProtoConnector/DataAggregator.Collector.OpenCNCapnProtoConnector.csproj +++ b/src/DataAggregator.Collector.OpenCNCapnProtoConnector/DataAggregator.Collector.OpenCNCapnProtoConnector.csproj @@ -9,6 +9,14 @@ true + + + PreserveNewest + true + PreserveNewest + + + diff --git a/src/DataAggregator.Collector.OpenCNCapnProtoConnector/configuration-example.json b/src/DataAggregator.Collector.OpenCNCapnProtoConnector/configuration-example.json new file mode 100644 index 0000000..3dbac9b --- /dev/null +++ b/src/DataAggregator.Collector.OpenCNCapnProtoConnector/configuration-example.json @@ -0,0 +1,113 @@ +{ + "AllowedHosts": "*", + + "Serilog": { + "MinimumLevel": { + "Default": "Information", + "Override": { + "Microsoft.AspNetCore": "Warning", + "System": "Warning" + } + }, + "WriteTo": [ + { + "Name": "Console", + "Args": { + "outputTemplate": "[{Timestamp:HH:mm:ss} {Level:u3}] {ServiceName}: {Message:lj}{NewLine}{Exception}" + } + } + ], + "Enrich": [ "FromLogContext" ], + "Properties": { + "ServiceName": "Collector" + } + }, + + "CollectorType": "OpenCN", + "Collector": { + "DeviceName": "Micro5", + "Location": "Test", + "HealthCheckEndpoint": "http://localhost:5000/health", + "FlushIntervalMilliseconds": 1000, + "SamplingRate": 100, + "CapnProto": { + "ServerAddress": "192.168.53.15", + "Port": 7002 + }, + "RegistrationService": { + "BaseUrl": "http://localhost:5137", + "Endpoint": "api/DeviceRegistration/register" + }, + "BufferSettings": { + "MaxBufferSize": 10000 + }, + "Sensors": [ + { + "Name": "current-amp-x", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float", + "PinName": "streamer.0.pin.0" + }, + { + "Name": "current-amp-y", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float", + "PinName": "streamer.0.pin.1" + }, + { + "Name": "current-amp-z", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float", + "PinName": "streamer.0.pin.2" + }, + { + "Name": "current-amp-b", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float", + "PinName": "streamer.0.pin.3" + }, + { + "Name": "current-amp-c", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float", + "PinName": "streamer.0.pin.4" + }, + { + "Name": "current-amp-s", + "Type": "Current", + "Unit": "A", + "Metadata": { + "MinValue": "-20", + "MaxValue": "20" + }, + "DataType": "float", + "PinName": "streamer.0.pin.5" + } + ] + } +} diff --git a/src/DataAggregator.Collector/DataAggregator.Collector.App.csproj b/src/DataAggregator.Collector/DataAggregator.Collector.App.csproj index 59a0c22..26b7735 100644 --- a/src/DataAggregator.Collector/DataAggregator.Collector.App.csproj +++ b/src/DataAggregator.Collector/DataAggregator.Collector.App.csproj @@ -19,6 +19,7 @@ + diff --git a/src/DataAggregator.Collector/Program.cs b/src/DataAggregator.Collector/Program.cs index 5a8aaa1..f687b48 100644 --- a/src/DataAggregator.Collector/Program.cs +++ b/src/DataAggregator.Collector/Program.cs @@ -1,3 +1,5 @@ +using DataAggregator.Collector.FileCollector.Configuration; +using DataAggregator.Collector.FileCollector.Connector; using DataAggregator.Collector.OpenCNCapnProtoConnector.CapnProto; using DataAggregator.Collector.OpenCNCapnProtoConnector.OpenCN; using DataAggregator.Collector.Shared.Abstraction; @@ -186,6 +188,9 @@ static void SetupConfiguration(WebApplicationBuilder builder) case "OPENCN": builder.Services.Configure(builder.Configuration.GetSection("Collector")); break; + case "FILE": + builder.Services.Configure(builder.Configuration.GetSection("Collector")); + break; default: Log.Warning("Collector type not specified or unsupported, application will close"); throw new InvalidOperationException("Collector type not specified or unsupported."); @@ -203,6 +208,13 @@ static void RegisterCollectorSpecificServices(WebApplicationBuilder builder, str return new CapnProtoConnector(config); }); break; + case "FILE": + builder.Services.AddSingleton(sp => + { + FileConnectorConfiguration config = sp.GetRequiredService>().Value; + return new FileConnector(config); + }); + break; default: Log.Fatal($"Unsupported collector type: {collectorType}"); throw new InvalidOperationException($"Unsupported collector type: {collectorType}"); diff --git a/src/DataAggregator.Collector/appsettings.json b/src/DataAggregator.Collector/appsettings.json index 3dbac9b..39f92dd 100644 --- a/src/DataAggregator.Collector/appsettings.json +++ b/src/DataAggregator.Collector/appsettings.json @@ -23,17 +23,13 @@ } }, - "CollectorType": "OpenCN", + "CollectorType": "File", "Collector": { "DeviceName": "Micro5", "Location": "Test", "HealthCheckEndpoint": "http://localhost:5000/health", "FlushIntervalMilliseconds": 1000, "SamplingRate": 100, - "CapnProto": { - "ServerAddress": "192.168.53.15", - "Port": 7002 - }, "RegistrationService": { "BaseUrl": "http://localhost:5137", "Endpoint": "api/DeviceRegistration/register" @@ -50,8 +46,7 @@ "MinValue": "-20", "MaxValue": "20" }, - "DataType": "float", - "PinName": "streamer.0.pin.0" + "DataType": "float" }, { "Name": "current-amp-y", @@ -61,8 +56,7 @@ "MinValue": "-20", "MaxValue": "20" }, - "DataType": "float", - "PinName": "streamer.0.pin.1" + "DataType": "float" }, { "Name": "current-amp-z", @@ -72,8 +66,7 @@ "MinValue": "-20", "MaxValue": "20" }, - "DataType": "float", - "PinName": "streamer.0.pin.2" + "DataType": "float" }, { "Name": "current-amp-b", @@ -83,8 +76,7 @@ "MinValue": "-20", "MaxValue": "20" }, - "DataType": "float", - "PinName": "streamer.0.pin.3" + "DataType": "float" }, { "Name": "current-amp-c", @@ -94,8 +86,7 @@ "MinValue": "-20", "MaxValue": "20" }, - "DataType": "float", - "PinName": "streamer.0.pin.4" + "DataType": "float" }, { "Name": "current-amp-s", @@ -105,9 +96,13 @@ "MinValue": "-20", "MaxValue": "20" }, - "DataType": "float", - "PinName": "streamer.0.pin.5" + "DataType": "float" } + ], + "Files": [ + "file1.txt", + "file2.txt", + "file3.txt" ] } } From 44d9ffea464d1ed4f4bdea55994f78b24c25d751 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Tue, 12 Aug 2025 21:15:47 +0200 Subject: [PATCH 3/9] feat: debug file connector --- .../Connector/FileConnector.cs | 6 ++++-- src/DataAggregator.Collector/Program.cs | 9 +++++++-- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs b/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs index 52cb631..7656b89 100644 --- a/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs +++ b/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs @@ -37,10 +37,12 @@ public Task ConnectAsync() foreach (string file in config.Files) { - if (!File.Exists(file)) + string executablePath = AppContext.BaseDirectory; + string filePath = Path.Combine(executablePath, file); + if (!File.Exists(filePath)) continue; - var reader = new StreamReader(file); + var reader = new StreamReader(filePath); string? header = reader.ReadLine(); if (header == null) { diff --git a/src/DataAggregator.Collector/Program.cs b/src/DataAggregator.Collector/Program.cs index f687b48..e222400 100644 --- a/src/DataAggregator.Collector/Program.cs +++ b/src/DataAggregator.Collector/Program.cs @@ -3,6 +3,7 @@ using DataAggregator.Collector.OpenCNCapnProtoConnector.CapnProto; using DataAggregator.Collector.OpenCNCapnProtoConnector.OpenCN; using DataAggregator.Collector.Shared.Abstraction; +using DataAggregator.Collector.Shared.Abstraction.Configuration; using DataAggregator.Collector.Shared.DataStorage; using DataAggregator.Collector.Shared.DataStorage.Influx; using DataAggregator.Collector.Shared.LocalStorage; @@ -79,7 +80,7 @@ builder.Services.AddSingleton(sp => { RegistrationService registrationService = sp.GetRequiredService(); - OpenCnCollectorConfiguration config = sp.GetRequiredService>().Value; + CollectorConfiguration config = sp.GetRequiredService>().Value; return new CollectorEndpointManager(registrationService, config); }); @@ -95,7 +96,7 @@ builder.Services.AddSingleton(sp => { - OpenCnCollectorConfiguration config = sp.GetRequiredService>().Value; + CollectorConfiguration config = sp.GetRequiredService>().Value; IDataSourceConnector dataSourceConnector = sp.GetRequiredService(); IDataRepository dataRepository = sp.GetRequiredService(); CollectorEndpointManager initService = sp.GetRequiredService(); @@ -187,9 +188,13 @@ static void SetupConfiguration(WebApplicationBuilder builder) { case "OPENCN": builder.Services.Configure(builder.Configuration.GetSection("Collector")); + builder.Services.AddSingleton>(sp => + sp.GetRequiredService>()); break; case "FILE": builder.Services.Configure(builder.Configuration.GetSection("Collector")); + builder.Services.AddSingleton>(sp => + sp.GetRequiredService>()); break; default: Log.Warning("Collector type not specified or unsupported, application will close"); From cfbd30875518f526f94b80cfa2796fcf5b7b0122 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Wed, 13 Aug 2025 02:55:55 +0200 Subject: [PATCH 4/9] feat: delete normalization --- .../Factory/DataProcessorFactory.cs | 5 +- .../ProcessorDescriptionJsonConverter.cs | 2 - .../ActuatorCurrentFeatureExtractor.cs | 65 ++++--------------- .../PreprocessingConfig.cs | 19 ------ src/DataAggregator.Processor/appsettings.json | 15 ----- .../ActuatorCurrentFeatureExtractorTests.cs | 8 +-- .../DataProcessorFactoryTests.cs | 7 +- .../PredictionBackgroundServiceTests.cs | 10 --- 8 files changed, 16 insertions(+), 115 deletions(-) delete mode 100644 src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs diff --git a/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs index 681ae13..635b6b6 100644 --- a/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs +++ b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs @@ -19,10 +19,7 @@ public List CreateProcessors(IEnumerable p switch (desc.Name.ToLowerInvariant()) { case "actuatorcurrent": - if (desc.Configuration is PreprocessingConfig preConfig) - processors.Add(new ActuatorCurrentFeatureExtractor(preConfig)); - else - throw new ArgumentException("Invalid config type for actuatorcurrent"); + processors.Add(new ActuatorCurrentFeatureExtractor()); break; case "onnxprediction": if (desc.Configuration is OnnxPredictionConfig onnxConfig) diff --git a/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs b/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs index a54c262..29a2f84 100644 --- a/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs +++ b/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs @@ -3,7 +3,6 @@ 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; @@ -21,7 +20,6 @@ public override ProcessorDescription Read(ref Utf8JsonReader reader, Type typeTo JsonElement configElement = root.GetProperty("Configuration"); IProcessorConfiguration? config = name.ToLowerInvariant() switch { - "actuatorcurrent" => configElement.Deserialize(options), "onnxprediction" => configElement.Deserialize(options), "statedeductionpostprocessor" => configElement.Deserialize(options), _ => null, diff --git a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs index 06594bc..73e80d4 100644 --- a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs @@ -11,8 +11,7 @@ namespace DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMer /// /// Initializes a new instance of the class. /// -/// The configuration of the processor. -public class ActuatorCurrentFeatureExtractor(PreprocessingConfig config) : IDataProcessor +public class ActuatorCurrentFeatureExtractor() : IDataProcessor { #region Public methods @@ -26,9 +25,6 @@ public Task> ProcessAsync(IEnumerable> ProcessAsync(IEnumerable { - new MeasurementData(meanTime, "GlobalActivityRatio", normalizedFeatures[0]), - new MeasurementData(meanTime, "GlobalChangeDensity", normalizedFeatures[1]), - new MeasurementData(meanTime, "InterAxisMeanCorrelation", normalizedFeatures[2]), - new MeasurementData(meanTime, "InterAxisMaxCorrelation", normalizedFeatures[3]), - new MeasurementData(meanTime, "InterAxisCorrelationVariance", normalizedFeatures[4]), - new MeasurementData(meanTime, "AxisSynchronization", normalizedFeatures[5]), - new MeasurementData(meanTime, "AxisLoadBalance", normalizedFeatures[6]), - new MeasurementData(meanTime, "TemporalStability", normalizedFeatures[7]), - new MeasurementData(meanTime, "GlobalSkewness", normalizedFeatures[8]), - new MeasurementData(meanTime, "CoefficientOfVariation", normalizedFeatures[9]), - new MeasurementData(meanTime, "NormalizedIqrMedian", normalizedFeatures[10]), - new MeasurementData(meanTime, "NormalizedIqrMean", normalizedFeatures[11]), + new MeasurementData(meanTime, "GlobalActivityRatio", features[0]), + new MeasurementData(meanTime, "GlobalChangeDensity", features[1]), + new MeasurementData(meanTime, "InterAxisMeanCorrelation", features[2]), + new MeasurementData(meanTime, "InterAxisMaxCorrelation", features[3]), + new MeasurementData(meanTime, "InterAxisCorrelationVariance", features[4]), + new MeasurementData(meanTime, "AxisSynchronization", features[5]), + new MeasurementData(meanTime, "AxisLoadBalance", features[6]), + new MeasurementData(meanTime, "TemporalStability", features[7]), + new MeasurementData(meanTime, "GlobalSkewness", features[8]), + new MeasurementData(meanTime, "CoefficientOfVariation", features[9]), + new MeasurementData(meanTime, "NormalizedIqrMedian", features[10]), + new MeasurementData(meanTime, "NormalizedIqrMean", features[11]), new MeasurementData(meanTime, "Label", string.Empty), }; @@ -236,40 +232,5 @@ private float CalculateTemporalStability(List allCurrents) : 1f; } - private float[] NormalizeFeaturesAsync(float[] features, PreprocessingConfig preprocessing) - { - if (!preprocessing.EnableZScoreNormalization) - { - return features; - } - - string[] featureNames = - [ - "GlobalActivityRatio", "GlobalChangeDensity", "InterAxisMeanCorrelation", - "InterAxisMaxCorrelation", "InterAxisCorrelationVariance", "AxisSynchronization", - "AxisLoadBalance", "TemporalStability", "GlobalSkewness", - "CoefficientOfVariation", "NormalizedIqrMedian", "NormalizedIqrMean", - ]; - - float[] normalized = new float[features.Length]; - - for (int i = 0; i < features.Length && i < featureNames.Length; i++) - { - string featureName = featureNames[i]; - if (preprocessing.NormalizationParameters.TryGetValue(featureName, out float[]? parameters) && parameters.Length >= 2) - { - float mean = parameters[0]; - float std = parameters[1]; - normalized[i] = std > 1e-6f ? (features[i] - mean) / std : features[i]; - } - else - { - normalized[i] = features[i]; // No normalization if parameters not found - } - } - - return normalized; - } - #endregion } diff --git a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs deleted file mode 100644 index bbec98f..0000000 --- a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs +++ /dev/null @@ -1,19 +0,0 @@ -using DataAggregator.Processor.Services.Processing.Abstraction; - -namespace DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; - -/// -/// Configuration for preprocessing operations including Z-score normalization. -/// -public class PreprocessingConfig : IProcessorConfiguration -{ - /// - /// Gets or sets a value indicating whether Z-score normalization is enabled. - /// - public bool EnableZScoreNormalization { get; set; } = true; - - /// - /// Gets or sets the normalization parameters for each feature (name, [mean, standard deviation]). - /// - public Dictionary NormalizationParameters { get; set; } = new(); -} diff --git a/src/DataAggregator.Processor/appsettings.json b/src/DataAggregator.Processor/appsettings.json index a0d76f8..4f4df93 100644 --- a/src/DataAggregator.Processor/appsettings.json +++ b/src/DataAggregator.Processor/appsettings.json @@ -44,21 +44,6 @@ { "Name": "actuatorcurrent", "Configuration": { - "EnableZScoreNormalization": true, - "NormalizationParameters": { - "GlobalActivityRatio": [ 0.003958, 0.018749 ], - "GlobalChangeDensity": [ 0.432119, 0.146720 ], - "InterAxisMeanCorrelation": [ -0.006328, 0.069353 ], - "InterAxisMaxCorrelation": [ 0.455234, 0.230869 ], - "InterAxisCorrelationVariance": [ 0.238157, 0.104391 ], - "AxisSynchronization": [ -38.056988, 293.112457 ], - "AxisLoadBalance": [ -0.045349, 0.330992 ], - "TemporalStability": [ 0.952670, 0.037949 ], - "GlobalSkewness": [ -0.122091, 0.380314 ], - "CoefficientOfVariation": [ -14.905608, 295.991791 ], - "NormalizedIqrMedian": [ -26.388798, 1187.886230 ], - "NormalizedIqrMean": [ -10.343025, 301.199188 ] - } } }, { diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs index f939ed6..41baa31 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs @@ -8,7 +8,7 @@ public class ActuatorCurrentFeatureExtractorTests private readonly ActuatorCurrentFeatureExtractor _featureExtractor; public ActuatorCurrentFeatureExtractorTests() - => _featureExtractor = new ActuatorCurrentFeatureExtractor(CreateValidPreprocessingConfig()); + => _featureExtractor = new ActuatorCurrentFeatureExtractor(); [Fact] public async Task ProcessAsync_ShouldReturnThirteenFeatures_WhenValidDataProvided() @@ -105,10 +105,4 @@ private static List CreateTestMeasurements() => [ new MeasurementData(DateTime.UtcNow, "sensor1", 12.1f), new MeasurementData(DateTime.UtcNow, "sensor2", 22.5f) ]; - - private static PreprocessingConfig CreateValidPreprocessingConfig() => new() - { - EnableZScoreNormalization = true, - NormalizationParameters = new Dictionary() - }; } diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs index cd95cb5..af1d148 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs @@ -22,11 +22,6 @@ public void CreateProcessors_ShouldReturnCorrectProcessors_ForValidPipeline() new ProcessorDescription { Name = "actuatorcurrent", - Configuration = new PreprocessingConfig - { - EnableZScoreNormalization = true, - NormalizationParameters = new Dictionary() - } }, new ProcessorDescription { @@ -88,7 +83,7 @@ public void CreateProcessors_ShouldThrowArgumentException_WhenNameMissing() // Arrange var pipeline = new List { - new ProcessorDescription { Name = "", Configuration = new PreprocessingConfig() } + new ProcessorDescription { Name = "",} }; // Act & Assert diff --git a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs index 01d35d7..dd3d283 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs @@ -126,11 +126,6 @@ public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() new ProcessorDescription { Name = "actuatorcurrent", - Configuration = new PreprocessingConfig - { - EnableZScoreNormalization = true, - NormalizationParameters = new Dictionary() - } } }, }, @@ -146,11 +141,6 @@ public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() new ProcessorDescription { Name = "actuatorcurrent", - Configuration = new PreprocessingConfig - { - EnableZScoreNormalization = true, - NormalizationParameters = new Dictionary() - } } }, } From c25fa5b5b46534cdaa56b700f1e9b205efaa0938 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Sat, 16 Aug 2025 01:26:25 +0200 Subject: [PATCH 5/9] Revert "feat: delete normalization" This reverts commit cfbd30875518f526f94b80cfa2796fcf5b7b0122. --- .../Factory/DataProcessorFactory.cs | 5 +- .../ProcessorDescriptionJsonConverter.cs | 2 + .../ActuatorCurrentFeatureExtractor.cs | 65 +++++++++++++++---- .../PreprocessingConfig.cs | 19 ++++++ src/DataAggregator.Processor/appsettings.json | 15 +++++ .../ActuatorCurrentFeatureExtractorTests.cs | 8 ++- .../DataProcessorFactoryTests.cs | 7 +- .../PredictionBackgroundServiceTests.cs | 10 +++ 8 files changed, 115 insertions(+), 16 deletions(-) create mode 100644 src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs diff --git a/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs index 635b6b6..681ae13 100644 --- a/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs +++ b/src/DataAggregator.Processor/Services/Processing/Factory/DataProcessorFactory.cs @@ -19,7 +19,10 @@ public List CreateProcessors(IEnumerable p switch (desc.Name.ToLowerInvariant()) { case "actuatorcurrent": - processors.Add(new ActuatorCurrentFeatureExtractor()); + if (desc.Configuration is PreprocessingConfig preConfig) + processors.Add(new ActuatorCurrentFeatureExtractor(preConfig)); + else + throw new ArgumentException("Invalid config type for actuatorcurrent"); break; case "onnxprediction": if (desc.Configuration is OnnxPredictionConfig onnxConfig) diff --git a/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs b/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs index 29a2f84..a54c262 100644 --- a/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs +++ b/src/DataAggregator.Processor/Services/Processing/Factory/ProcessorDescriptionJsonConverter.cs @@ -3,6 +3,7 @@ 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; @@ -20,6 +21,7 @@ public override ProcessorDescription Read(ref Utf8JsonReader reader, Type typeTo JsonElement configElement = root.GetProperty("Configuration"); IProcessorConfiguration? config = name.ToLowerInvariant() switch { + "actuatorcurrent" => configElement.Deserialize(options), "onnxprediction" => configElement.Deserialize(options), "statedeductionpostprocessor" => configElement.Deserialize(options), _ => null, diff --git a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs index 73e80d4..06594bc 100644 --- a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs @@ -11,7 +11,8 @@ namespace DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMer /// /// Initializes a new instance of the class. /// -public class ActuatorCurrentFeatureExtractor() : IDataProcessor +/// The configuration of the processor. +public class ActuatorCurrentFeatureExtractor(PreprocessingConfig config) : IDataProcessor { #region Public methods @@ -25,6 +26,9 @@ public Task> ProcessAsync(IEnumerable> ProcessAsync(IEnumerable { - new MeasurementData(meanTime, "GlobalActivityRatio", features[0]), - new MeasurementData(meanTime, "GlobalChangeDensity", features[1]), - new MeasurementData(meanTime, "InterAxisMeanCorrelation", features[2]), - new MeasurementData(meanTime, "InterAxisMaxCorrelation", features[3]), - new MeasurementData(meanTime, "InterAxisCorrelationVariance", features[4]), - new MeasurementData(meanTime, "AxisSynchronization", features[5]), - new MeasurementData(meanTime, "AxisLoadBalance", features[6]), - new MeasurementData(meanTime, "TemporalStability", features[7]), - new MeasurementData(meanTime, "GlobalSkewness", features[8]), - new MeasurementData(meanTime, "CoefficientOfVariation", features[9]), - new MeasurementData(meanTime, "NormalizedIqrMedian", features[10]), - new MeasurementData(meanTime, "NormalizedIqrMean", features[11]), + new MeasurementData(meanTime, "GlobalActivityRatio", normalizedFeatures[0]), + new MeasurementData(meanTime, "GlobalChangeDensity", normalizedFeatures[1]), + new MeasurementData(meanTime, "InterAxisMeanCorrelation", normalizedFeatures[2]), + new MeasurementData(meanTime, "InterAxisMaxCorrelation", normalizedFeatures[3]), + new MeasurementData(meanTime, "InterAxisCorrelationVariance", normalizedFeatures[4]), + new MeasurementData(meanTime, "AxisSynchronization", normalizedFeatures[5]), + new MeasurementData(meanTime, "AxisLoadBalance", normalizedFeatures[6]), + new MeasurementData(meanTime, "TemporalStability", normalizedFeatures[7]), + new MeasurementData(meanTime, "GlobalSkewness", normalizedFeatures[8]), + new MeasurementData(meanTime, "CoefficientOfVariation", normalizedFeatures[9]), + new MeasurementData(meanTime, "NormalizedIqrMedian", normalizedFeatures[10]), + new MeasurementData(meanTime, "NormalizedIqrMean", normalizedFeatures[11]), new MeasurementData(meanTime, "Label", string.Empty), }; @@ -232,5 +236,40 @@ private float CalculateTemporalStability(List allCurrents) : 1f; } + private float[] NormalizeFeaturesAsync(float[] features, PreprocessingConfig preprocessing) + { + if (!preprocessing.EnableZScoreNormalization) + { + return features; + } + + string[] featureNames = + [ + "GlobalActivityRatio", "GlobalChangeDensity", "InterAxisMeanCorrelation", + "InterAxisMaxCorrelation", "InterAxisCorrelationVariance", "AxisSynchronization", + "AxisLoadBalance", "TemporalStability", "GlobalSkewness", + "CoefficientOfVariation", "NormalizedIqrMedian", "NormalizedIqrMean", + ]; + + float[] normalized = new float[features.Length]; + + for (int i = 0; i < features.Length && i < featureNames.Length; i++) + { + string featureName = featureNames[i]; + if (preprocessing.NormalizationParameters.TryGetValue(featureName, out float[]? parameters) && parameters.Length >= 2) + { + float mean = parameters[0]; + float std = parameters[1]; + normalized[i] = std > 1e-6f ? (features[i] - mean) / std : features[i]; + } + else + { + normalized[i] = features[i]; // No normalization if parameters not found + } + } + + return normalized; + } + #endregion } diff --git a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs new file mode 100644 index 0000000..bbec98f --- /dev/null +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/PreprocessingConfig.cs @@ -0,0 +1,19 @@ +using DataAggregator.Processor.Services.Processing.Abstraction; + +namespace DataAggregator.Processor.Services.Processing.PreProcessing.ActuatorMergingCurrentPreprocessing; + +/// +/// Configuration for preprocessing operations including Z-score normalization. +/// +public class PreprocessingConfig : IProcessorConfiguration +{ + /// + /// Gets or sets a value indicating whether Z-score normalization is enabled. + /// + public bool EnableZScoreNormalization { get; set; } = true; + + /// + /// Gets or sets the normalization parameters for each feature (name, [mean, standard deviation]). + /// + public Dictionary NormalizationParameters { get; set; } = new(); +} diff --git a/src/DataAggregator.Processor/appsettings.json b/src/DataAggregator.Processor/appsettings.json index 4f4df93..a0d76f8 100644 --- a/src/DataAggregator.Processor/appsettings.json +++ b/src/DataAggregator.Processor/appsettings.json @@ -44,6 +44,21 @@ { "Name": "actuatorcurrent", "Configuration": { + "EnableZScoreNormalization": true, + "NormalizationParameters": { + "GlobalActivityRatio": [ 0.003958, 0.018749 ], + "GlobalChangeDensity": [ 0.432119, 0.146720 ], + "InterAxisMeanCorrelation": [ -0.006328, 0.069353 ], + "InterAxisMaxCorrelation": [ 0.455234, 0.230869 ], + "InterAxisCorrelationVariance": [ 0.238157, 0.104391 ], + "AxisSynchronization": [ -38.056988, 293.112457 ], + "AxisLoadBalance": [ -0.045349, 0.330992 ], + "TemporalStability": [ 0.952670, 0.037949 ], + "GlobalSkewness": [ -0.122091, 0.380314 ], + "CoefficientOfVariation": [ -14.905608, 295.991791 ], + "NormalizedIqrMedian": [ -26.388798, 1187.886230 ], + "NormalizedIqrMean": [ -10.343025, 301.199188 ] + } } }, { diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs index 41baa31..f939ed6 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs @@ -8,7 +8,7 @@ public class ActuatorCurrentFeatureExtractorTests private readonly ActuatorCurrentFeatureExtractor _featureExtractor; public ActuatorCurrentFeatureExtractorTests() - => _featureExtractor = new ActuatorCurrentFeatureExtractor(); + => _featureExtractor = new ActuatorCurrentFeatureExtractor(CreateValidPreprocessingConfig()); [Fact] public async Task ProcessAsync_ShouldReturnThirteenFeatures_WhenValidDataProvided() @@ -105,4 +105,10 @@ private static List CreateTestMeasurements() => [ new MeasurementData(DateTime.UtcNow, "sensor1", 12.1f), new MeasurementData(DateTime.UtcNow, "sensor2", 22.5f) ]; + + private static PreprocessingConfig CreateValidPreprocessingConfig() => new() + { + EnableZScoreNormalization = true, + NormalizationParameters = new Dictionary() + }; } diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs index af1d148..cd95cb5 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/DataProcessorFactoryTests.cs @@ -22,6 +22,11 @@ public void CreateProcessors_ShouldReturnCorrectProcessors_ForValidPipeline() new ProcessorDescription { Name = "actuatorcurrent", + Configuration = new PreprocessingConfig + { + EnableZScoreNormalization = true, + NormalizationParameters = new Dictionary() + } }, new ProcessorDescription { @@ -83,7 +88,7 @@ public void CreateProcessors_ShouldThrowArgumentException_WhenNameMissing() // Arrange var pipeline = new List { - new ProcessorDescription { Name = "",} + new ProcessorDescription { Name = "", Configuration = new PreprocessingConfig() } }; // Act & Assert diff --git a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs index dd3d283..01d35d7 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PredictionBackgroundServiceTests.cs @@ -126,6 +126,11 @@ public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() new ProcessorDescription { Name = "actuatorcurrent", + Configuration = new PreprocessingConfig + { + EnableZScoreNormalization = true, + NormalizationParameters = new Dictionary() + } } }, }, @@ -141,6 +146,11 @@ public async Task ExecuteAsync_ShouldStopGracefully_WhenCancellationRequested() new ProcessorDescription { Name = "actuatorcurrent", + Configuration = new PreprocessingConfig + { + EnableZScoreNormalization = true, + NormalizationParameters = new Dictionary() + } } }, } From d913359b0836743334a6f7dac55115f85d84e836 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Sat, 16 Aug 2025 01:28:32 +0200 Subject: [PATCH 6/9] feat: rework feature selection --- .../Connector/FileConnector.cs | 3 +- src/DataAggregator.Collector/appsettings.json | 4 +- .../ActuatorCurrentFeatureExtractor.cs | 71 ++++--------------- src/DataAggregator.Processor/appsettings.json | 13 +--- 4 files changed, 20 insertions(+), 71 deletions(-) diff --git a/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs b/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs index 7656b89..f3d46e1 100644 --- a/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs +++ b/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs @@ -136,7 +136,8 @@ public Task> FetchDataAsync() DateTime timestamp = now - TimeSpan.FromMilliseconds(intervalMs * (linesToProvide - 1 - i)); foreach (SensorConfig sensor in config.Sensors) { - int colIdx = _sensorNames.FindIndex(n => n == sensor.Name); + int colIdx = _sensorNames.FindIndex(n => + string.Equals(n, sensor.Name, StringComparison.OrdinalIgnoreCase)); if (colIdx == -1 || colIdx >= row.Length) continue; diff --git a/src/DataAggregator.Collector/appsettings.json b/src/DataAggregator.Collector/appsettings.json index 39f92dd..305a3cc 100644 --- a/src/DataAggregator.Collector/appsettings.json +++ b/src/DataAggregator.Collector/appsettings.json @@ -100,9 +100,7 @@ } ], "Files": [ - "file1.txt", - "file2.txt", - "file3.txt" + "complete.csv" ] } } diff --git a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs index 06594bc..08a9ec9 100644 --- a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs @@ -36,22 +36,16 @@ public Task> ProcessAsync(IEnumerable + string[] featureNames = config.NormalizationParameters.Keys.ToArray(); + + var result = new List(); + + for (int i = 0; i < normalizedFeatures.Length && i < featureNames.Length; i++) { - new MeasurementData(meanTime, "GlobalActivityRatio", normalizedFeatures[0]), - new MeasurementData(meanTime, "GlobalChangeDensity", normalizedFeatures[1]), - new MeasurementData(meanTime, "InterAxisMeanCorrelation", normalizedFeatures[2]), - new MeasurementData(meanTime, "InterAxisMaxCorrelation", normalizedFeatures[3]), - new MeasurementData(meanTime, "InterAxisCorrelationVariance", normalizedFeatures[4]), - new MeasurementData(meanTime, "AxisSynchronization", normalizedFeatures[5]), - new MeasurementData(meanTime, "AxisLoadBalance", normalizedFeatures[6]), - new MeasurementData(meanTime, "TemporalStability", normalizedFeatures[7]), - new MeasurementData(meanTime, "GlobalSkewness", normalizedFeatures[8]), - new MeasurementData(meanTime, "CoefficientOfVariation", normalizedFeatures[9]), - new MeasurementData(meanTime, "NormalizedIqrMedian", normalizedFeatures[10]), - new MeasurementData(meanTime, "NormalizedIqrMean", normalizedFeatures[11]), - new MeasurementData(meanTime, "Label", string.Empty), - }; + result.Add(new MeasurementData(meanTime, featureNames[i], normalizedFeatures[i])); + } + + result.Add(new MeasurementData(meanTime, "Label", string.Empty)); // Placeholder for label return Task.FromResult(result.AsEnumerable()); } @@ -65,7 +59,7 @@ private float[] ExtractFeatures(IEnumerable measurements) if (measurements == null) { Log.Warning("No measurements provided for feature extraction."); - return new float[14]; + return new float[5]; } // Concatenate all currents from all actuators @@ -93,35 +87,15 @@ private float[] ExtractFeatures(IEnumerable measurements) if (allCurrents.Count == 0) { Log.Warning("No valid current values found for feature extraction."); - return new float[14]; + return new float[5]; } - // Calculate global statistics - float globalStd = MathUtils.StandardDeviation(allCurrents); - float globalMean = MathUtils.Mean(allCurrents); - float globalMedian = MathUtils.Percentile(allCurrents, 50); - float globalQ25 = MathUtils.Percentile(allCurrents, 25); - float globalQ75 = MathUtils.Percentile(allCurrents, 75); - float globalIqr = globalQ75 - globalQ25; - - float activityThreshold = globalStd * 2; - float activeRatio = allCurrents.Count(x => Math.Abs(x) > activityThreshold) / (float)allCurrents.Count; - - var diffSignals = allCurrents.Zip(allCurrents.Skip(1), (a, b) => Math.Abs(b - a)).ToList(); - int significantChanges = diffSignals.Count(x => x > globalStd * 1.5); - float changeDensity = significantChanges / (float)allCurrents.Count; - List> actuatorsCurrents = ExtractActuatorCurrents(measurements); List correlations = CalculateInterActuatorCorrelations(actuatorsCurrents); - float meanCorrelation = correlations.Count > 0 ? correlations.Average() : 0f; float maxCorrelation = correlations.Count > 0 ? correlations.Max() : 0f; float correlationVariance = correlations.Count > 0 ? MathUtils.StandardDeviation(correlations) : 0f; - var actuatorsMeans = actuatorsCurrents.Select(MathUtils.Mean).ToList(); - float meanOfMeans = actuatorsMeans.Average(); - float synchronization = meanOfMeans != 0 ? 1 - (MathUtils.StandardDeviation(actuatorsMeans) / Math.Abs(meanOfMeans)) : 1f; - var actuatorsEnergies = actuatorsCurrents.Select(actuator => actuator.Sum(x => x * x)).ToList(); float meanEnergy = actuatorsEnergies.Average(); float loadBalance = meanEnergy != 0 ? 1 - (MathUtils.StandardDeviation(actuatorsEnergies) / meanEnergy) : 1f; @@ -129,24 +103,13 @@ private float[] ExtractFeatures(IEnumerable measurements) float globalSkewness = MathUtils.Skewness(allCurrents); - float coeffVar = Math.Abs(globalMean) > 1e-8f ? globalStd / globalMean : 0f; - float normIqrMedian = Math.Abs(globalMedian) > 1e-8f ? globalIqr / globalMedian : 0f; - float normIqrMean = Math.Abs(globalMean) > 1e-8f ? globalIqr / globalMean : 0f; - return [ - activeRatio, - changeDensity, - meanCorrelation, - maxCorrelation, - correlationVariance, - synchronization, loadBalance, + correlationVariance, + maxCorrelation, temporalStability, globalSkewness, - coeffVar, - normIqrMedian, - normIqrMean, ]; } @@ -243,13 +206,7 @@ private float[] NormalizeFeaturesAsync(float[] features, PreprocessingConfig pre return features; } - string[] featureNames = - [ - "GlobalActivityRatio", "GlobalChangeDensity", "InterAxisMeanCorrelation", - "InterAxisMaxCorrelation", "InterAxisCorrelationVariance", "AxisSynchronization", - "AxisLoadBalance", "TemporalStability", "GlobalSkewness", - "CoefficientOfVariation", "NormalizedIqrMedian", "NormalizedIqrMean", - ]; + string[] featureNames = preprocessing.NormalizationParameters.Keys.ToArray(); float[] normalized = new float[features.Length]; diff --git a/src/DataAggregator.Processor/appsettings.json b/src/DataAggregator.Processor/appsettings.json index a0d76f8..5ab478c 100644 --- a/src/DataAggregator.Processor/appsettings.json +++ b/src/DataAggregator.Processor/appsettings.json @@ -46,18 +46,11 @@ "Configuration": { "EnableZScoreNormalization": true, "NormalizationParameters": { - "GlobalActivityRatio": [ 0.003958, 0.018749 ], - "GlobalChangeDensity": [ 0.432119, 0.146720 ], - "InterAxisMeanCorrelation": [ -0.006328, 0.069353 ], - "InterAxisMaxCorrelation": [ 0.455234, 0.230869 ], - "InterAxisCorrelationVariance": [ 0.238157, 0.104391 ], - "AxisSynchronization": [ -38.056988, 293.112457 ], "AxisLoadBalance": [ -0.045349, 0.330992 ], + "InterAxisCorrelationVariance": [ 0.238157, 0.104391 ], + "InterAxisMaxCorrelation": [ 0.455234, 0.230869 ], "TemporalStability": [ 0.952670, 0.037949 ], - "GlobalSkewness": [ -0.122091, 0.380314 ], - "CoefficientOfVariation": [ -14.905608, 295.991791 ], - "NormalizedIqrMedian": [ -26.388798, 1187.886230 ], - "NormalizedIqrMean": [ -10.343025, 301.199188 ] + "GlobalSkewness": [ -0.122091, 0.380314 ] } } }, From d28c3aad5074ca0191e1df15cf914739ef47ef23 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Sat, 16 Aug 2025 01:58:22 +0200 Subject: [PATCH 7/9] fix: tests --- .../ActuatorCurrentFeatureExtractor.cs | 2 +- .../ProcessorTestHelper.cs | 98 ++++++------------ .../ActuatorCurrentFeatureExtractorTests.cs | 42 +++----- .../resources/opencn_model.onnx | Bin 14010 -> 9816 bytes 4 files changed, 49 insertions(+), 93 deletions(-) diff --git a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs index 08a9ec9..5543c9f 100644 --- a/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs +++ b/src/DataAggregator.Processor/Services/Processing/PreProcessing/ActuatorMergingCurrentPreprocessing/ActuatorCurrentFeatureExtractor.cs @@ -23,7 +23,7 @@ public class ActuatorCurrentFeatureExtractor(PreprocessingConfig config) : IData /// Feature vector as dictionary mapping feature names to values for a single sample. public Task> ProcessAsync(IEnumerable input) { - // Extract the 14 features from measurements + // Extract the 5 features from measurements float[] features = ExtractFeatures(input); // Apply Z-score normalization if enabled diff --git a/tests/DataAggregator.Processor.Tests/ProcessorTestHelper.cs b/tests/DataAggregator.Processor.Tests/ProcessorTestHelper.cs index a2d5689..9abc57c 100644 --- a/tests/DataAggregator.Processor.Tests/ProcessorTestHelper.cs +++ b/tests/DataAggregator.Processor.Tests/ProcessorTestHelper.cs @@ -10,90 +10,54 @@ public static IEnumerable GetValidTestData() return new List { - new MeasurementData(now, "GlobalActivityRatio", 1.0f), - new MeasurementData(now, "GlobalChangeDensity", 2.0f), - new MeasurementData(now, "InterAxisMeanCorrelation", 3.0f), - new MeasurementData(now, "InterAxisMaxCorrelation", 4.0f), - new MeasurementData(now, "InterAxisCorrelationVariance", 5.0f), - new MeasurementData(now, "AxisSynchronization", 6.0f), - new MeasurementData(now, "AxisLoadBalance", 7.0f), - new MeasurementData(now, "TemporalStability", 8.0f), - new MeasurementData(now, "GlobalSkewness", 9.0f), - new MeasurementData(now, "GlobalKurtosis", 10.0f), - new MeasurementData(now, "GlobalTrendSlope", 11.0f), - new MeasurementData(now, "CoefficientOfVariation", 12.0f), - new MeasurementData(now, "NormalizedIqrMedian", 13.0f), - new MeasurementData(now, "NormalizedIqrMean", 14.0f), + new MeasurementData(now, "InterAxisMaxCorrelation", 0.167771f), + new MeasurementData(now, "InterAxisCorrelationVariance", 0.079077f), + new MeasurementData(now, "AxisLoadBalance", 0.145757f), + new MeasurementData(now, "TemporalStability", -0.189833f), + new MeasurementData(now, "GlobalSkewness", -0.014682f), new MeasurementData(now, "Label", string.Empty), }; } - public static IEnumerable GetValidShutdownStateData() + public static IEnumerable GetValidProductionStateData() { DateTime now = DateTime.Now; return new List - { - new MeasurementData(now, "GlobalActivityRatio", -0.235608f), - new MeasurementData(now, "GlobalChangeDensity", -3.017057f), - new MeasurementData(now, "InterAxisMeanCorrelation", 0.018719f), - new MeasurementData(now, "InterAxisMaxCorrelation", -1.639018f), - new MeasurementData(now, "InterAxisCorrelationVariance", -1.741219f), - new MeasurementData(now, "AxisSynchronization", 0.177111f), - new MeasurementData(now, "AxisLoadBalance", 3.195790f), - new MeasurementData(now, "TemporalStability", 0.542284f), - new MeasurementData(now, "GlobalSkewness", 0.357844f), - new MeasurementData(now, "GlobalKurtosis", 1.441117f), - new MeasurementData(now, "GlobalTrendSlope", 0.005194f), - new MeasurementData(now, "CoefficientOfVariation",0.073491f), - new MeasurementData(now, "NormalizedIqrMedian", -0.005889f), - new MeasurementData(now, "NormalizedIqrMean", 0.052909f), - new MeasurementData(now, "Label", string.Empty), - }; + { + new MeasurementData(now, "AxisLoadBalance", -0.115415f), + new MeasurementData(now, "InterAxisCorrelationVariance", 0.335840f), + new MeasurementData(now, "InterAxisMaxCorrelation", 0.262740f), + new MeasurementData(now, "TemporalStability", -0.271154f), + new MeasurementData(now, "GlobalSkewness", -0.203212f), + new MeasurementData(now, "Label", string.Empty), + }; } - public static IEnumerable GetValidProductionStateData() + public static IEnumerable GetValidShutdownStateData() { DateTime now = DateTime.Now; return new List - { - new MeasurementData(now, "GlobalActivityRatio", -0.235608f), - new MeasurementData(now, "GlobalChangeDensity", -0.063313f), - new MeasurementData(now, "InterAxisMeanCorrelation", 0.353871f), - new MeasurementData(now, "InterAxisMaxCorrelation", 0.167771f), - new MeasurementData(now, "InterAxisCorrelationVariance", 0.079077f), - new MeasurementData(now, "AxisSynchronization", 0.142284f), - new MeasurementData(now, "AxisLoadBalance", 0.145757f), - new MeasurementData(now, "TemporalStability", -0.189833f), - new MeasurementData(now, "GlobalSkewness", -0.014682f), - new MeasurementData(now, "GlobalKurtosis", -1.229796f), - new MeasurementData(now, "GlobalTrendSlope", -2.419304f), - new MeasurementData(now, "CoefficientOfVariation", 0.037766f), - new MeasurementData(now, "NormalizedIqrMedian", -0.051617f), - new MeasurementData(now, "NormalizedIqrMean", 0.008711f), - new MeasurementData(now, "Label", string.Empty), - }; + { + new MeasurementData(now, "AxisLoadBalance", 3.184065f), + new MeasurementData(now, "InterAxisCorrelationVariance", -2.281908f), + new MeasurementData(now, "InterAxisMaxCorrelation", -1.977913f), + new MeasurementData(now, "TemporalStability", 1.253469f), + new MeasurementData(now, "GlobalSkewness", 0.315182f), + new MeasurementData(now, "Label", string.Empty), + }; } public static IEnumerable GetValidIdleStateData() { DateTime now = DateTime.Now; return new List - { - new MeasurementData(now, "GlobalActivityRatio", -0.235608f), - new MeasurementData(now, "GlobalChangeDensity", 0.471617f), - new MeasurementData(now, "InterAxisMeanCorrelation", 0.087918f), - new MeasurementData(now, "InterAxisMaxCorrelation", -0.901309f), - new MeasurementData(now, "InterAxisCorrelationVariance", -1.025950f), - new MeasurementData(now, "AxisSynchronization", -0.045073f), - new MeasurementData(now, "AxisLoadBalance", -1.059821f), - new MeasurementData(now, "TemporalStability", 0.526964f), - new MeasurementData(now, "GlobalSkewness", 1.200339f), - new MeasurementData(now, "GlobalKurtosis", 1.352620f), - new MeasurementData(now, "GlobalTrendSlope", -0.011385f), - new MeasurementData(now, "CoefficientOfVariation", -0.135674f), - new MeasurementData(now, "NormalizedIqrMedian", -0.005889f), - new MeasurementData(now, "NormalizedIqrMean", -0.004563f), - new MeasurementData(now, "Label", string.Empty), - }; + { + new MeasurementData(now, "AxisLoadBalance", -1.060774f), + new MeasurementData(now, "InterAxisCorrelationVariance", -0.848838f), + new MeasurementData(now, "InterAxisMaxCorrelation", -0.555483f), + new MeasurementData(now, "TemporalStability", 1.017235f), + new MeasurementData(now, "GlobalSkewness", 1.114139f), + new MeasurementData(now, "Label", string.Empty), + }; } } diff --git a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs index f939ed6..c78be1f 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs @@ -5,6 +5,8 @@ namespace DataAggregator.Processor.Tests.Services.PreProcessing; public class ActuatorCurrentFeatureExtractorTests { + private const int _expectedFeatureCount = 5; + private readonly ActuatorCurrentFeatureExtractor _featureExtractor; public ActuatorCurrentFeatureExtractorTests() @@ -16,21 +18,20 @@ public async Task ProcessAsync_ShouldReturnThirteenFeatures_WhenValidDataProvide var measurements = CreateTestMeasurements(); var result = await _featureExtractor.ProcessAsync(measurements); Assert.NotNull(result); - Assert.Equal(13, result.Count()); + Assert.Equal(_expectedFeatureCount + 1, result.Count()); } [Fact] - public async Task ProcessAsync_ShouldReturnTwelvesFeatures_WhenEmptyMeasurementsProvided() + public async Task ProcessAsync_ShouldReturnFiveFeatures_WhenEmptyMeasurementsProvided() { var measurements = new List(); var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); Assert.NotNull(result); - Assert.Equal(12, result.Count()); - Assert.All(result, feature => Assert.Equal(0.0f, (float)feature.GetRawValue())); + Assert.Equal(_expectedFeatureCount, result.Count()); } - + [Fact] public async Task ProcessAsync_ShouldReturnValidFeatures_WhenValidDataProvided() { @@ -38,27 +39,11 @@ public async Task ProcessAsync_ShouldReturnValidFeatures_WhenValidDataProvided() var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); Assert.NotNull(result); - Assert.Equal(12, result.Count()); + Assert.Equal(_expectedFeatureCount, result.Count()); Assert.All(result, feature => Assert.False(float.IsNaN((float)feature.GetRawValue()))); Assert.All(result, feature => Assert.False(float.IsInfinity((float)feature.GetRawValue()))); } - [Fact] - public async Task ProcessAsync_ShouldReturnZeroFeatures_WhenNoValidValuesFound() - { - 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), - }; - var result = await _featureExtractor.ProcessAsync(measurements); - result = result.Where(f => f.SensorName != "Label"); - Assert.NotNull(result); - Assert.Equal(12, result.Count()); - Assert.All(result, feature => Assert.Equal(0.0f, (float)feature.GetRawValue())); - } - [Fact] public async Task ProcessAsync_ShouldHandleSingleValue_WhenOnlyOneValidMeasurementProvided() { @@ -69,7 +54,7 @@ public async Task ProcessAsync_ShouldHandleSingleValue_WhenOnlyOneValidMeasureme var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); Assert.NotNull(result); - Assert.Equal(12, result.Count()); + Assert.Equal(_expectedFeatureCount, result.Count()); Assert.All(result, feature => Assert.False(float.IsNaN((float)feature.GetRawValue()))); } @@ -92,7 +77,7 @@ public async Task ProcessAsync_ShouldHandleLargeDataset_WhenManyMeasurementsProv var result = await _featureExtractor.ProcessAsync(measurements); result = result.Where(f => f.SensorName != "Label"); Assert.NotNull(result); - Assert.Equal(12, result.Count()); + Assert.Equal(_expectedFeatureCount, result.Count()); Assert.All(result, feature => Assert.False(float.IsNaN((float)feature.GetRawValue()))); Assert.All(result, feature => Assert.False(float.IsInfinity((float)feature.GetRawValue()))); } @@ -109,6 +94,13 @@ private static List CreateTestMeasurements() => [ private static PreprocessingConfig CreateValidPreprocessingConfig() => new() { EnableZScoreNormalization = true, - NormalizationParameters = new Dictionary() + NormalizationParameters = new Dictionary + { + { "AxisLoadBalance", [-0.045349f, 0.330992f] }, + { "InterAxisCorrelationVariance", [0.238157f, 0.104391f] }, + { "InterAxisMaxCorrelation", [0.455234f, 0.230869f] }, + { "TemporalStability", [0.952670f, 0.037949f] }, + { "GlobalSkewness", [-0.122091f, 0.380314f] }, + }, }; } diff --git a/tests/DataAggregator.Processor.Tests/resources/opencn_model.onnx b/tests/DataAggregator.Processor.Tests/resources/opencn_model.onnx index 06c1a332941031f00f5fc7c478dce50f28b438f1..91a843f4ed39cfff44bb8df077541812ce02f766 100644 GIT binary patch delta 1151 zcma)*ZD>R<<+ z`+x5FpL6c>@B$)=PGMC8z1H^B?QKS7s^m zR}bc8k4EXvZ_ig&&$Imh?!&pp0h?I>;RP%<$j@2CHSB;V1kKKC^vXiQC}K)Cf-Y?u z*7yICr5`-V>K7ve^pD@D{_}|dt)EQlaV1I9k2-Yz&JaCL;`;oNX8P_czH+l;iMY8c zD95v}7~EA6NR^kdvf3YdzkG`Ku~&a-`aiv|zwtcNTfA_XChKM^t9eDnYyB4#KKDBK zoC^$+0kSYon$j;H?$^5)vMeam3rn5)^_@u;l<7f-N9S)0v7n28yThp0^12sWt|u^A>K`wTfoF5F8!Y?5Zj6_UsT=E+ejbUB-Q{$lY+axZsh8p|;p@ zx#d;N8~V50k6PJ|m!FR!E8FY4r8SAvZe={uU^f+WvX{IBr@syNU#^H#C z7v_j{KuW3MyrxZtbK}xx1ll_LYKAh3R#Hp*`uw)XsnjKj+tc}i(qK0woS!Het=P1@ zb^~VGN-OYXXM2QTA{c@_*pHim7)}M>!QBo&Qe9Ep@7#*hof|RRH3n-aNkoi>t{@Zc z!Jj7&-71ojzNsTCqo>;a+IB~-y4MMvLNy&*VG@?b>PDD$x6D2O{7ExE?QeK+e zRd<(;;`HF6Xi_*mwJ6XO2w)T~dT7x@i=sVQK@L3>4w6HE(p!5fQux?YQ8axsyUU%~ zogrlhJTwUbak;bad-G=I%{On}tkOxGTv?u8J@Z0dDHRTM)XmGfp|v&DGIV2Wx?wto z=FI)KjczFDzGvFzhT2}(bo8sb^UCw8qnmiYpsEcBVUR^hVE^#mlO( zrJdFc8!YmT;R;Lnv}a6dNY?y7L5G$MXzjv|Zm(#nv1nSB)+XI!;QM7y0=vpXvg`W_ zIvCWh?yzPt26GPrt13gX>MaEw?pei{bWyc*)!5YV1O1)`_RS2*KH^(sAlF_oHZNPI zpio4T|xU^&^p_urMA}`bwh8H z*@h!UfhA84$&zm>DC;uL+7<0(L$hr>$=?SyJT-J1&UG!vwDq11{(WG>i6PnWZ3X4r zcE4b0Mq{mQZfkg7xFoRSZon~n$}qQy0xM;fLri>K_O$#uz2`8FndBZgE2T##F#QH^8HC12^Va?lwUcK zeDPveJbnQ|{rMXSc@#xB4GGuyEukoj(JGAsl?E21&uDN8(JG2@e zgnxED)zEEqqpj7^wq-WD5c!!#J*gR_?52_~Vl>gvnrgT0tm}sJkG*=^Hy)*VYV>?W{EA+dUX^$clI4viV> zj%sabj!kFiMgM+`+@EW5CrEUpyLvw$q~_oAZAd`t)oeu*aayHttR1KMn~ushJch3 z(DfJ^cTpW;Umew+Uszpyc76HGbJE|5(TH3LoD%d2*5IK7r2f(x4nBV`SzV( z|El`Kul?o}?LOw~(B!+jzqtC<>a~Afi#$vIwLE{U^46*L>QAc-g#O(4*&lY_`Qk6D zmrCgf4mLl?*b!Fba{7V_>Lp&qMB;k4qytn34FW@ahl+c zz-s7-3+w2%4K(2G*I)u{ULii()p`y}kdY73@hHl=6Qk$TIU+|Knm9A@WM0f~RE~Ht z`$xncLderb+D$dFwGxpRdlkN{>06f_d+yZ8$aU!G>%0En(c71+@VD##{rROY?ZRKx zyY9ySvQ&jXa;1OaUFYM-rw}yi23!*>-S#SshPNEV;kH1&&=ZJ%gp%i*Oc2kNxWy1+_22I--(pCy8YaYJ zJ|UJ-k`iJGXE{a$27>#N%ZW26^bn<7QY_&KmK6a6j0hmYi)F+=45bvqXg*1 zjD5H<=7xvfB@v_RqZc!NlDN_5#UH)-CgDgDB+_8<)t7D%j@-R<-wxqM8Zh4ZK&`&N zu|k8z4`184)p&17&W#n18_OYXEK_bQi@CAvb7Pt3MgRdT0!YA(AjeGu^$D zs-VbAVT_U8J4F0}6dqf$K@Xwve0p^&}STv*cjO%?ngG%``bO&l!Uv zK#ZA2teOU`H34O?B9X?Nf{2mogmtgRJw z#~a?HdspzT#YvRDqG{WZ3G)&roW7%oL!(ieQ^I55?4Ca2{T0{yXHZa%vllE#@oqX= zgJy-?!Jpv~i_AukdhINyK1IA1PTWGGc#^lQ zl%HVJZ>3QFZ3>Ok{2Oh!?PtRv-IMjwWP41^aoNPPes-5Txh>(mdtAVkPMdCKGriP+ zbUvK&6nekVGrk@^OYH06N_{4 zdJ!d0GD@EIv+(|jTnSI{M{*JPQzCdTiSNVxf(ZNs178rN{PDr?L0Ck^Nk+x0^oRgI zAU?zL)gj-z;V3JZC(nvc7H~dv;zQ1)qMBH*J}o?rI26(&(!#p(G34dSv779V*2#J& ztZ(cY1HO;Fi$Ym{{8Jnq?zG{Bk!cPCTSI46?PxYWB6!+0y0+%?dtZ2nD}Ngc9af0m zPpNx_968mQw;%Y|HR`e#ChC3PS*VNN7^sI2?(5Mr^m^zp+rRf+zaBZ{txJwu>!K6S zdi30pY*Py1Q-Niq4CTF#P!?)GLOQm7@p#nwWwdPG;l93wC)VdNB;^j_?`+68ZG2-ea8?~C7#E78EQB*mMDi<@k)SRN!+<2<^ zIEpDsE?tS624wODQ?l3P;3oh^EpC-iOB1ANd~z#nMV3&&^3Jm7igk zc7D3bK^6I{PtuB>8t`J06bL;aSi0rL6{GL;NOd0*y?G;*emwGa%wSQUYvlGG6})RA z)mxUo5YopZLQ5KbT|-(0^hrn_-lvdCf6hM)ClYCaM{XQPeL`WW;sC}uvd3<6*c=)Z z;>xin(STr94xLAPiCE=$=p=!Y4{R;CgE%6U*8~$@7J^DdqQ?eN^{k{R!Vk7M7Us zRPQ4V#30@%_Q60mhUO`m-hBYjGa*1+@WSSJn)r^(dEQChEonJk)xGO*BiTz{|rq zXnD=hJJBq&TpSt{q-p!O2q#zGqrQq<4%!sX2+M@y#CzD>FBKH=I_b||MprHcvz7M< zx7{M%Z-Mvl8Ctp8dyDm2Js3yjgCd63$B6Y>_a5uDK4z>(S}UOVfvvs!SdXk-cOC1s z?mE^B5I*$}Y|A~wdaZko^+<<@xP33N9_hUhY;dec<|$=nwhM$rdzY~u>12M!vNZ_j U?Ox-xS{5V!6!!03+?0ng9R* From 7773c94b4b66f949f377041af49a06832714138d Mon Sep 17 00:00:00 2001 From: CoJaques Date: Sat, 16 Aug 2025 14:05:10 +0200 Subject: [PATCH 8/9] feat: adapt collector dockerfile to new fileConnector --- DataAggregator.sln | 28 +++++++++---------- docker/Dockerfile.Collector | 1 + .../FileConnectorConfiguration.cs | 0 .../Connector/FileConnector.cs | 0 ...aAggregator.Collector.FileCollector.csproj | 21 ++++++++++++++ .../configuration-example-FileConnector.json | 0 ....Collector.OpenCNCapnProtoConnector.csproj | 2 +- ...configuration-example-CapnpConnector.json} | 0 .../DataAggregator.Collector.App.csproj | 2 +- 9 files changed, 38 insertions(+), 16 deletions(-) rename {DataAggregator.Collector.FileCollector => src/DataAggregator.Collector.FileCollector}/Configuration/FileConnectorConfiguration.cs (100%) rename {DataAggregator.Collector.FileCollector => src/DataAggregator.Collector.FileCollector}/Connector/FileConnector.cs (100%) create mode 100644 src/DataAggregator.Collector.FileCollector/DataAggregator.Collector.FileCollector.csproj rename DataAggregator.Collector.FileCollector/configuration-example.json => src/DataAggregator.Collector.FileCollector/configuration-example-FileConnector.json (100%) rename src/DataAggregator.Collector.OpenCNCapnProtoConnector/{configuration-example.json => configuration-example-CapnpConnector.json} (100%) diff --git a/DataAggregator.sln b/DataAggregator.sln index a20751f..ded92da 100644 --- a/DataAggregator.sln +++ b/DataAggregator.sln @@ -40,7 +40,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "processor", "processor", "{ EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Processor.Tests", "tests\DataAggregator.Processor.Tests\DataAggregator.Processor.Tests.csproj", "{EFE47FE6-F41E-CDD6-0991-472080AD88B0}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Collector.FileCollector", "DataAggregator.Collector.FileCollector\DataAggregator.Collector.FileCollector.csproj", "{D72B893B-E69F-44A4-8747-6967CF3BA3BE}" +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Collector.FileCollector", "src\DataAggregator.Collector.FileCollector\DataAggregator.Collector.FileCollector.csproj", "{40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}" EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution @@ -160,18 +160,18 @@ Global {EFE47FE6-F41E-CDD6-0991-472080AD88B0}.Release|x64.Build.0 = Release|x64 {EFE47FE6-F41E-CDD6-0991-472080AD88B0}.Release|x86.ActiveCfg = Release|x86 {EFE47FE6-F41E-CDD6-0991-472080AD88B0}.Release|x86.Build.0 = Release|x86 - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|Any CPU.Build.0 = Debug|Any CPU - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|x64.ActiveCfg = Debug|x64 - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|x64.Build.0 = Debug|x64 - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|x86.ActiveCfg = Debug|x86 - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Debug|x86.Build.0 = Debug|x86 - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|Any CPU.ActiveCfg = Release|Any CPU - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|Any CPU.Build.0 = Release|Any CPU - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|x64.ActiveCfg = Release|x64 - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|x64.Build.0 = Release|x64 - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|x86.ActiveCfg = Release|x86 - {D72B893B-E69F-44A4-8747-6967CF3BA3BE}.Release|x86.Build.0 = Release|x86 + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Debug|Any CPU.Build.0 = Debug|Any CPU + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Debug|x64.ActiveCfg = Debug|x64 + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Debug|x64.Build.0 = Debug|x64 + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Debug|x86.ActiveCfg = Debug|x86 + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Debug|x86.Build.0 = Debug|x86 + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Release|Any CPU.ActiveCfg = Release|Any CPU + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Release|Any CPU.Build.0 = Release|Any CPU + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Release|x64.ActiveCfg = Release|x64 + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Release|x64.Build.0 = Release|x64 + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Release|x86.ActiveCfg = Release|x86 + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}.Release|x86.Build.0 = Release|x86 EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -190,7 +190,7 @@ Global {039EC00D-5EDF-4C48-B449-29CFB6750232} = {EB9A5576-3E00-4007-8C72-2235E0A1546D} {EB9A5576-3E00-4007-8C72-2235E0A1546D} = {E1AD9667-4C40-4CAF-8096-5FA749EBB2B1} {EFE47FE6-F41E-CDD6-0991-472080AD88B0} = {247EF7A2-1DFD-4B51-AC7D-0FD13827CAC2} - {D72B893B-E69F-44A4-8747-6967CF3BA3BE} = {F5250AC4-9CCC-432B-8725-DAC6AE01CCF0} + {40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1} = {F5250AC4-9CCC-432B-8725-DAC6AE01CCF0} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {9C8E6DBA-9F77-4FD0-9A1D-25150F7806FF} diff --git a/docker/Dockerfile.Collector b/docker/Dockerfile.Collector index 3e8f84d..bb5f57b 100644 --- a/docker/Dockerfile.Collector +++ b/docker/Dockerfile.Collector @@ -7,6 +7,7 @@ COPY lib /lib COPY src/DataAggregator.Collector/*.csproj ./DataAggregator.Collector/ COPY src/DataAggregator.Shared/*.csproj ./DataAggregator.Shared/ COPY src/DataAggregator.Collector.OpenCNCapnProtoConnector/*.csproj ./DataAggregator.Collector.OpenCNCapnProtoConnector/ +COPY src/DataAggregator.Collector.FileCollector/*.csproj ./DataAggregator.Collector.FileCollector/ RUN dotnet restore ./DataAggregator.Collector/DataAggregator.Collector.App.csproj --configfile /NuGet.config diff --git a/DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs b/src/DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs similarity index 100% rename from DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs rename to src/DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs diff --git a/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs b/src/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs similarity index 100% rename from DataAggregator.Collector.FileCollector/Connector/FileConnector.cs rename to src/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs diff --git a/src/DataAggregator.Collector.FileCollector/DataAggregator.Collector.FileCollector.csproj b/src/DataAggregator.Collector.FileCollector/DataAggregator.Collector.FileCollector.csproj new file mode 100644 index 0000000..9e64fd2 --- /dev/null +++ b/src/DataAggregator.Collector.FileCollector/DataAggregator.Collector.FileCollector.csproj @@ -0,0 +1,21 @@ + + + + net9.0 + enable + enable + + + + + PreserveNewest + true + PreserveNewest + + + + + + + + diff --git a/DataAggregator.Collector.FileCollector/configuration-example.json b/src/DataAggregator.Collector.FileCollector/configuration-example-FileConnector.json similarity index 100% rename from DataAggregator.Collector.FileCollector/configuration-example.json rename to src/DataAggregator.Collector.FileCollector/configuration-example-FileConnector.json diff --git a/src/DataAggregator.Collector.OpenCNCapnProtoConnector/DataAggregator.Collector.OpenCNCapnProtoConnector.csproj b/src/DataAggregator.Collector.OpenCNCapnProtoConnector/DataAggregator.Collector.OpenCNCapnProtoConnector.csproj index ab783b2..a555b97 100644 --- a/src/DataAggregator.Collector.OpenCNCapnProtoConnector/DataAggregator.Collector.OpenCNCapnProtoConnector.csproj +++ b/src/DataAggregator.Collector.OpenCNCapnProtoConnector/DataAggregator.Collector.OpenCNCapnProtoConnector.csproj @@ -10,7 +10,7 @@ - + PreserveNewest true PreserveNewest diff --git a/src/DataAggregator.Collector.OpenCNCapnProtoConnector/configuration-example.json b/src/DataAggregator.Collector.OpenCNCapnProtoConnector/configuration-example-CapnpConnector.json similarity index 100% rename from src/DataAggregator.Collector.OpenCNCapnProtoConnector/configuration-example.json rename to src/DataAggregator.Collector.OpenCNCapnProtoConnector/configuration-example-CapnpConnector.json diff --git a/src/DataAggregator.Collector/DataAggregator.Collector.App.csproj b/src/DataAggregator.Collector/DataAggregator.Collector.App.csproj index 26b7735..f299abc 100644 --- a/src/DataAggregator.Collector/DataAggregator.Collector.App.csproj +++ b/src/DataAggregator.Collector/DataAggregator.Collector.App.csproj @@ -19,7 +19,7 @@ - + From e42446faf644f99ca4cf37a824a7c4723cfe3367 Mon Sep 17 00:00:00 2001 From: CoJaques Date: Sat, 16 Aug 2025 16:09:48 +0200 Subject: [PATCH 9/9] feat: add warning if file not found --- .../Connector/FileConnector.cs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs b/src/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs index f3d46e1..5359ba2 100644 --- a/src/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs +++ b/src/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs @@ -4,6 +4,7 @@ using DataAggregator.Collector.Shared.Abstraction.Configuration; using DataAggregator.Collector.Shared.Models; using DataAggregator.Shared.Domain.DataType; +using Serilog; namespace DataAggregator.Collector.FileCollector.Connector; @@ -40,7 +41,10 @@ public Task ConnectAsync() string executablePath = AppContext.BaseDirectory; string filePath = Path.Combine(executablePath, file); if (!File.Exists(filePath)) + { + Log.Warning("File not found: {FilePath}", filePath); continue; + } var reader = new StreamReader(filePath); string? header = reader.ReadLine();