Feature/rework processor

DataAggregator.sln

@@ -17,8 +17,6 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Collector.Te
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Registration.Tests", "tests\DataAggregator.Registration.Tests\DataAggregator.Registration.Tests.csproj", "{FB0B4F80-5801-407F-8425-54069C93DB60}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Integration.Tests", "tests\DataAggregator.Integration.Tests\DataAggregator.Integration.Tests.csproj", "{12FEEB75-A03A-4389-980B-6B735C550CAC}"
-EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{02EA681E-C7D8-13C7-8484-4AC65E1B71E8}"
 	ProjectSection(SolutionItems) = preProject
 		.editorconfig = .editorconfig
@@ -38,8 +36,6 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Collector.Op
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Processor", "src\DataAggregator.Processor\DataAggregator.Processor.csproj", "{039EC00D-5EDF-4C48-B449-29CFB6750232}"
 EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "processor", "processor", "{330C7B17-DB90-458C-B630-99F9C1B5EA45}"
-EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "processor", "processor", "{EB9A5576-3E00-4007-8C72-2235E0A1546D}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataAggregator.Processor.Tests", "tests\DataAggregator.Processor.Tests\DataAggregator.Processor.Tests.csproj", "{EFE47FE6-F41E-CDD6-0991-472080AD88B0}"
@@ -114,18 +110,6 @@ Global
 		{FB0B4F80-5801-407F-8425-54069C93DB60}.Release|x64.Build.0 = Release|x64
 		{FB0B4F80-5801-407F-8425-54069C93DB60}.Release|x86.ActiveCfg = Release|x86
 		{FB0B4F80-5801-407F-8425-54069C93DB60}.Release|x86.Build.0 = Release|x86
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|x64.ActiveCfg = Debug|x64
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|x64.Build.0 = Debug|x64
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|x86.ActiveCfg = Debug|x86
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Debug|x86.Build.0 = Debug|x86
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|Any CPU.Build.0 = Release|Any CPU
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|x64.ActiveCfg = Release|x64
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|x64.Build.0 = Release|x64
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|x86.ActiveCfg = Release|x86
-		{12FEEB75-A03A-4389-980B-6B735C550CAC}.Release|x86.Build.0 = Release|x86
 		{4A5708E5-B5D1-4DA4-A2E4-323D3A21B182}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{4A5708E5-B5D1-4DA4-A2E4-323D3A21B182}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{4A5708E5-B5D1-4DA4-A2E4-323D3A21B182}.Debug|x64.ActiveCfg = Debug|x64
@@ -184,7 +168,6 @@ Global
 		{E593EA7B-F713-4444-AE40-FDCEF462660A} = {AA521B32-6E75-46BD-ABDF-0450227A94C4}
 		{63D70F2C-9C7D-4397-8150-37F343949530} = {247EF7A2-1DFD-4B51-AC7D-0FD13827CAC2}
 		{FB0B4F80-5801-407F-8425-54069C93DB60} = {247EF7A2-1DFD-4B51-AC7D-0FD13827CAC2}
-		{12FEEB75-A03A-4389-980B-6B735C550CAC} = {247EF7A2-1DFD-4B51-AC7D-0FD13827CAC2}
 		{95F8787D-6FE2-4173-8450-6762EA6FAE1F} = {E1AD9667-4C40-4CAF-8096-5FA749EBB2B1}
 		{AA521B32-6E75-46BD-ABDF-0450227A94C4} = {E1AD9667-4C40-4CAF-8096-5FA749EBB2B1}
 		{F5250AC4-9CCC-432B-8725-DAC6AE01CCF0} = {95F8787D-6FE2-4173-8450-6762EA6FAE1F}

src/DataAggregator.Collector.Shared/Abstraction/CollectorService.cs

@@ -27,7 +27,7 @@ public class CollectorService(
 
     private readonly SemaphoreSlim _processingLock = new(1, 1);
     private readonly ConcurrentQueue<IMeasurementData> _dataQueue = new();
-    private readonly TimeSpan _flushInterval = TimeSpan.FromSeconds(1); // Flush every 1 second
+    private readonly TimeSpan _flushInterval = TimeSpan.FromMilliseconds(configuration.FlushIntervalMilliseconds); // Flush every 1 second
 
     private bool _isRunning;
     private CancellationTokenSource? _cancellationTokenSource;

src/DataAggregator.Collector.Shared/Abstraction/Configuration/CollectorConfiguration.cs

@@ -24,4 +24,9 @@ public class CollectorConfiguration
     /// Gets or sets the list of sensors configured for this collector.
     /// </summary>
     public List<SensorConfig> Sensors { get; set; } = [];
+
+    /// <summary>
+    /// Gets or sets the interval in milliseconds for flushing data to the time series database.
+    /// </summary>
+    public int FlushIntervalMilliseconds { get; set; } = 1000;
 }

src/DataAggregator.Collector/appsettings.json

@@ -20,6 +20,7 @@
     "DeviceName": "Micro5",
     "Location": "Test",
     "HealthCheckEndpoint": "http://localhost:5000/health",
+    "FlushIntervalMilliseconds": 1000,
     "SamplingRate": 100,
     "CapnProto": {
       "ServerAddress": "192.168.53.15",

src/DataAggregator.Processor/Configuration/MachinePredictionConfig.cs

@@ -1,4 +1,6 @@
-namespace DataAggregator.Processor.Configuration;
+using DataAggregator.Processor.Services.Processing.Factory;
+
+namespace DataAggregator.Processor.Configuration;
 
 /// <summary>
 /// Configuration for a specific machine prediction.
@@ -16,32 +18,27 @@ public class MachinePredictionConfig
     public bool Enabled { get; set; } = true;
 
     /// <summary>
-    /// Gets or sets the path to the ONNX model file.
-    /// </summary>
-    public string ModelPath { get; set; } = string.Empty;
-
-    /// <summary>
-    /// Gets or sets the preprocessing strategy name.
+    /// Gets or sets the list of input sensor names.
     /// </summary>
-    public string PreprocessingStrategy { get; set; } = string.Empty;
+    public List<string> InputSensors { get; set; } = [];
 
     /// <summary>
-    /// Gets or sets the list of input sensor names.
+    /// Gets or sets a value indicating whether Window size in seconds if true, otherwise in elements number.
     /// </summary>
-    public List<string> InputSensors { get; set; } = [];
+    public bool WindowSizeInSeconds { get; set; } = true;
 
     /// <summary>
-    /// Gets or sets the window size in seconds for data collection.
+    /// Gets or sets the window size unit depends on WindowSizeInSeconds property.
     /// </summary>
-    public int WindowSizeSeconds { get; set; } = 60;
+    public int WindowSize { get; set; } = 60;
 
     /// <summary>
     /// Gets or sets the cycle interval in seconds for this machine.
     /// </summary>
-    public int CycleIntervalSeconds { get; set; } = 1;
+    public double CycleIntervalSeconds { get; set; } = 1;
 
     /// <summary>
-    /// Gets or sets the preprocessing configuration for Z-score normalization.
+    /// Gets or sets the processing pipeline for this machine.
     /// </summary>
-    public PreprocessingConfig Preprocessing { get; set; } = new();
+    public List<ProcessorDescription> ProcessingPipeline { get; set; } = new();
 }

src/DataAggregator.Processor/Program.cs

@@ -1,8 +1,9 @@
+using System.Text.Json;
 using DataAggregator.Processor.Configuration;
 using DataAggregator.Processor.Services;
 using DataAggregator.Processor.Services.DataStorage;
 using DataAggregator.Processor.Services.Prediction;
-using DataAggregator.Processor.Services.PreProcessing;
+using DataAggregator.Processor.Services.Processing.Factory;
 using DataAggregator.Processor.Services.Registration;
 using Serilog;
 
@@ -25,8 +26,18 @@
 // Register health checks
 builder.Services.AddHealthChecks();
 
-// Configure prediction service
-builder.Services.Configure<PredictionServiceConfiguration>(builder.Configuration.GetSection("PredictionService"));
+// Configuration management
+string appSettingsJson = File.ReadAllText("appsettings.json");
+using var doc = JsonDocument.Parse(appSettingsJson);
+JsonElement predictionServiceElement = doc.RootElement.GetProperty("PredictionService");
+string json = predictionServiceElement.GetRawText();
+
+var options = new JsonSerializerOptions { PropertyNameCaseInsensitive = true };
+options.Converters.Add(new ProcessorDescriptionJsonConverter());
+PredictionServiceConfiguration? predictionConfig = JsonSerializer.Deserialize<PredictionServiceConfiguration>(json, options);
+if (predictionConfig == null)
+    throw new Exception("Failed to deserialize PredictionServiceConfiguration");
+builder.Services.AddSingleton(predictionConfig);
 
 // Configure HTTP clients
 builder.Services.AddHttpClient<IRegistrationServiceClient, RegistrationServiceClient>("RegistrationClient", client =>
@@ -38,8 +49,7 @@
 
 // Register services
 builder.Services.AddScoped<IDataRepository, InfluxV3Repository>();
-builder.Services.AddSingleton<IOnnxPredictionEngine, OnnxPredictionEngine>();
-builder.Services.AddSingleton<IPreprocessingStrategyFactory, PreprocessingStrategyFactory>();
+builder.Services.AddSingleton<IDataProcessorFactory, DataProcessorFactory>();
 builder.Services.AddScoped<IMachinePredictionProcessor, MachinePredictionProcessor>();
 
 // Register background service

src/DataAggregator.Processor/Services/DataStorage/IDataRepository.cs

@@ -13,7 +13,7 @@ public interface IDataRepository
     /// </summary>
     /// <param name="endpoint">The InfluxDB endpoint.</param>
     /// <param name="token">The authentication token.</param>
-    public void InitializeAsync(string endpoint, string token);
+    public void Initialize(string endpoint, string token);
 
     /// <summary>
     /// Queries measurements from InfluxDB for a specific time range and sensors with type information.
@@ -25,6 +25,15 @@ public interface IDataRepository
     /// <returns>A list of measurement data.</returns>
     public Task<List<IMeasurementData>> QueryMeasurementsAsync(string table, DateTime startTime, DateTime endTime, List<SensorInfoDto> sensors);
 
+    /// <summary>
+    /// Queries the last measurements from InfluxDB for a specific table and window size, filtering by sensors.
+    /// </summary>
+    /// <param name="table">The table name.</param>
+    /// <param name="windowSize">The size of the window.</param>
+    /// <param name="sensors">The list of sensor information with type data.</param>
+    /// <returns>A list of measurements data of windowSize size max.</returns>
+    public Task<List<IMeasurementData>> QueryLastMeasurements(string table, int windowSize, List<SensorInfoDto> sensors);
+
     /// <summary>
     /// Writes a single measurement to InfluxDB.
     /// </summary>