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.sln b/DataAggregator.sln index 7c3b036..ded92da 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", "src\DataAggregator.Collector.FileCollector\DataAggregator.Collector.FileCollector.csproj", "{40BE9DBC-6DA7-3610-DF71-4C4DF64B0CD1}" +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 + {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 @@ -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} + {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/src/DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs b/src/DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs new file mode 100644 index 0000000..3385c73 --- /dev/null +++ b/src/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/src/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs b/src/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs new file mode 100644 index 0000000..5359ba2 --- /dev/null +++ b/src/DataAggregator.Collector.FileCollector/Connector/FileConnector.cs @@ -0,0 +1,195 @@ +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; +using Serilog; + +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) + { + 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(); + 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 => + string.Equals(n, sensor.Name, StringComparison.OrdinalIgnoreCase)); + + 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/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/src/DataAggregator.Collector.FileCollector/configuration-example-FileConnector.json b/src/DataAggregator.Collector.FileCollector/configuration-example-FileConnector.json new file mode 100644 index 0000000..39f92dd --- /dev/null +++ b/src/DataAggregator.Collector.FileCollector/configuration-example-FileConnector.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/src/DataAggregator.Collector.OpenCNCapnProtoConnector/DataAggregator.Collector.OpenCNCapnProtoConnector.csproj b/src/DataAggregator.Collector.OpenCNCapnProtoConnector/DataAggregator.Collector.OpenCNCapnProtoConnector.csproj index 9d101c0..a555b97 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-CapnpConnector.json b/src/DataAggregator.Collector.OpenCNCapnProtoConnector/configuration-example-CapnpConnector.json new file mode 100644 index 0000000..3dbac9b --- /dev/null +++ b/src/DataAggregator.Collector.OpenCNCapnProtoConnector/configuration-example-CapnpConnector.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..f299abc 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..e222400 100644 --- a/src/DataAggregator.Collector/Program.cs +++ b/src/DataAggregator.Collector/Program.cs @@ -1,6 +1,9 @@ +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; +using DataAggregator.Collector.Shared.Abstraction.Configuration; using DataAggregator.Collector.Shared.DataStorage; using DataAggregator.Collector.Shared.DataStorage.Influx; using DataAggregator.Collector.Shared.LocalStorage; @@ -77,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); }); @@ -93,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(); @@ -185,6 +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"); @@ -203,6 +213,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..305a3cc 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,11 @@ "MinValue": "-20", "MaxValue": "20" }, - "DataType": "float", - "PinName": "streamer.0.pin.5" + "DataType": "float" } + ], + "Files": [ + "complete.csv" ] } } 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..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 @@ -36,24 +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, "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, "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()); } @@ -67,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 @@ -95,76 +87,29 @@ 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; - - // 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; return [ - activeRatio, - changeDensity, - meanCorrelation, - maxCorrelation, - correlationVariance, - synchronization, loadBalance, + correlationVariance, + maxCorrelation, temporalStability, globalSkewness, - globalKurtosis, - trendSlope, - coeffVar, - normIqrMedian, - normIqrMean, ]; } @@ -254,23 +199,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) @@ -278,13 +206,7 @@ private float[] NormalizeFeaturesAsync(float[] features, PreprocessingConfig pre return features; } - string[] featureNames = - [ - "GlobalActivityRatio", "GlobalChangeDensity", "InterAxisMeanCorrelation", - "InterAxisMaxCorrelation", "InterAxisCorrelationVariance", "AxisSynchronization", - "AxisLoadBalance", "TemporalStability", "GlobalSkewness", "GlobalKurtosis", - "GlobalTrendSlope", "CoefficientOfVariation", "NormalizedIqrMedian", "NormalizedIqrMean", - ]; + string[] featureNames = preprocessing.NormalizationParameters.Keys.ToArray(); 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..5ab478c 100644 --- a/src/DataAggregator.Processor/appsettings.json +++ b/src/DataAggregator.Processor/appsettings.json @@ -46,20 +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 ], - "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 ] + "GlobalSkewness": [ -0.122091, 0.380314 ] } } }, 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 ceb8a75..c78be1f 100644 --- a/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs +++ b/tests/DataAggregator.Processor.Tests/Services/PreProcessing/ActuatorCurrentFeatureExtractorTests.cs @@ -5,32 +5,33 @@ namespace DataAggregator.Processor.Tests.Services.PreProcessing; public class ActuatorCurrentFeatureExtractorTests { + private const int _expectedFeatureCount = 5; + private readonly ActuatorCurrentFeatureExtractor _featureExtractor; 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(_expectedFeatureCount + 1, result.Count()); } [Fact] - public async Task ProcessAsync_ShouldReturnFourteenFeatures_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(14, 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(14, 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(14, 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(14, 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(14, 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/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; diff --git a/tests/DataAggregator.Processor.Tests/resources/opencn_model.onnx b/tests/DataAggregator.Processor.Tests/resources/opencn_model.onnx index 06c1a33..91a843f 100644 Binary files a/tests/DataAggregator.Processor.Tests/resources/opencn_model.onnx and b/tests/DataAggregator.Processor.Tests/resources/opencn_model.onnx differ