Feature/enhance deployement

DataAggregator.Collector.FileCollector/DataAggregator.Collector.FileCollector.csproj

@@ -0,0 +1,21 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net9.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <Content Include="configuration-example.json">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+      <ExcludeFromSingleFile>true</ExcludeFromSingleFile>
+      <CopyToPublishDirectory>PreserveNewest</CopyToPublishDirectory>
+    </Content>
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\src\DataAggregator.Collector.Shared\DataAggregator.Collector.Shared.csproj" />
+  </ItemGroup>
+
+</Project>

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}

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
 

src/DataAggregator.Collector.FileCollector/Configuration/FileConnectorConfiguration.cs

@@ -0,0 +1,19 @@
+using DataAggregator.Collector.Shared.Abstraction.Configuration;
+
+namespace DataAggregator.Collector.FileCollector.Configuration;
+
+/// <summary>
+/// Specifies the configuration for the File Connector.
+/// </summary>
+public class FileConnectorConfiguration : CollectorConfiguration
+{
+    /// <summary>
+    /// Gets or sets the list of file paths to read data from.
+    /// </summary>
+    public List<string> Files { get; set; } = [];
+
+    /// <summary>
+    /// Gets or sets the sampling rate (in Hz) for data provision.
+    /// </summary>
+    public double SamplingRate { get; set; } = 100;
+}

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;
+
+/// <summary>
+/// Define a connector for file-based data sources, used to read data from files.
+/// Useful for testing or when data is stored in files.
+/// </summary>
+public class FileConnector(FileConnectorConfiguration config) : IDataSourceConnector
+{
+    #region Private Fields
+    private List<StreamReader> _readers = [];
+    private List<string> _sensorNames = [];
+    private int _currentFileIndex = 0;
+    private DateTime _lastFetchTime = DateTime.MinValue;
+    private bool _initialized = false;
+    #endregion
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>Task.CompletedTask.</returns>
+    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;
+    }
+
+    /// <summary>
+    /// Closes and disposes all open files, and resets the connector state.
+    /// </summary>
+    /// <returns>Task.CompletedTask.</returns>
+    public Task DisconnectAsync()
+    {
+        DisposeReaders();
+        _readers = [];
+        _sensorNames = [];
+        _currentFileIndex = 0;
+        _lastFetchTime = DateTime.MinValue;
+        _initialized = false;
+
+        return Task.CompletedTask;
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A batch of sensor data, with correct timestamps.</returns>
+    public Task<IEnumerable<IMeasurementData>> FetchDataAsync()
+    {
+        if (!_initialized)
+            return Task.FromResult(Enumerable.Empty<IMeasurementData>());
+
+        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<IEnumerable<IMeasurementData>>([]);
+
+        List<IMeasurementData> 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<bool>(timestamp, sensor.Name, float.Parse(valueStr, CultureInfo.InvariantCulture) != 0),
+                    SensorDataType.Integer => new MeasurementData<int>(timestamp, sensor.Name, (int)float.Parse(valueStr, CultureInfo.InvariantCulture)),
+                    SensorDataType.Double or SensorDataType.Float => new MeasurementData<double>(timestamp, sensor.Name, double.Parse(valueStr, CultureInfo.InvariantCulture)),
+                    SensorDataType.String => new MeasurementData<string>(timestamp, sensor.Name, valueStr),
+                    _ => null,
+                };
+
+                if (measurement != null)
+                    result.Add(measurement);
+            }
+        }
+
+        _lastFetchTime = _lastFetchTime.AddMilliseconds(linesToProvide * intervalMs);
+        return Task.FromResult<IEnumerable<IMeasurementData>>(result);
+    }
+
+    /// <summary>
+    /// Checks that all files specified in the configuration exist and are accessible.
+    /// </summary>
+    /// <returns>True if all files exist, false otherwise.</returns>
+    public Task<bool> 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();
+        }
+    }
+}

src/DataAggregator.Collector.FileCollector/DataAggregator.Collector.FileCollector.csproj

@@ -0,0 +1,21 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net9.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <Content Include="configuration-example-FileConnector.json">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+      <ExcludeFromSingleFile>true</ExcludeFromSingleFile>
+      <CopyToPublishDirectory>PreserveNewest</CopyToPublishDirectory>
+    </Content>
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\DataAggregator.Collector.Shared\DataAggregator.Collector.Shared.csproj" />
+  </ItemGroup>
+
+</Project>