diff --git a/Ares.Device.Drivers.slnx b/Ares.Device.Drivers.slnx index 9afe033..67386e3 100644 --- a/Ares.Device.Drivers.slnx +++ b/Ares.Device.Drivers.slnx @@ -4,6 +4,8 @@ + + diff --git a/SerialStreamingSensor/Connection/ISerialSteramingConnection.cs b/SerialStreamingSensor/Connection/ISerialSteramingConnection.cs new file mode 100644 index 0000000..5ac1265 --- /dev/null +++ b/SerialStreamingSensor/Connection/ISerialSteramingConnection.cs @@ -0,0 +1,9 @@ +using Ares.Toolkit.Serial; + +namespace SerialStreamingSensor.Connection +{ + public interface ISerialSteramingConnection: IAresSerialConnection + { + + } +} diff --git a/SerialStreamingSensor/Connection/SerialStreamingConnection.cs b/SerialStreamingSensor/Connection/SerialStreamingConnection.cs new file mode 100644 index 0000000..251a591 --- /dev/null +++ b/SerialStreamingSensor/Connection/SerialStreamingConnection.cs @@ -0,0 +1,19 @@ +using Ares.Toolkit.Serial; +using Ares.Toolkit.Serial.Commands; +using System.IO.Ports; + +namespace SerialStreamingSensor.Connection +{ + public class SerialStreamingConnection : AresHardwareConnection, ISerialSteramingConnection + { + public SerialStreamingConnection(SerialPortConnectionInfo connectionInfo, string portName, SerialConnectionOptions? connectionOptions = null) : base(connectionInfo, portName, connectionOptions) + { + } + + public SerialStreamingConnection(string portName, int baudRate = 115200, Parity parity = Parity.None, int dataBits = 8, StopBits stopBits = StopBits.One, SerialConnectionOptions? options = null) : base(portName, baudRate, parity, dataBits, stopBits, options) + { + + } + + } +} diff --git a/SerialStreamingSensor/DelimitedStreamSensor.cs b/SerialStreamingSensor/DelimitedStreamSensor.cs new file mode 100644 index 0000000..18d65f0 --- /dev/null +++ b/SerialStreamingSensor/DelimitedStreamSensor.cs @@ -0,0 +1,466 @@ +using Ares.Datamodel; +using Ares.Datamodel.Device; +using Ares.Datamodel.Extensions; +using Ares.Datamodel.Factories; +using Ares.Device; +using Microsoft.AspNetCore.Components.Web; +using Microsoft.AspNetCore.Mvc.Diagnostics; +using SerialStreamingSensor.Connection; +using SerialStreamingSensor.Models; +using StreamHelper; +using System.Linq.Expressions; +using System.Reactive.Disposables; +using System.Reactive.Linq; +using System.Reactive.Subjects; + +namespace SerialStreamingSensor +{ + public class DelimitedStreamSensor : AresDevice, IDelimitedSteamSensor + { + private readonly DeviceConnectionInfo _connectionInfo; + private readonly ILogger _logger; + private readonly string _dataFormat; + private readonly string[] _fields; + private readonly StreamingField[] _streamingFields; + private readonly IReadOnlyDictionary _streamingFieldsByName; + private readonly System.IO.Ports.SerialPort _serialConnection; + private CancellationTokenSource _stateGetterLoopTokenSource = new(); + //private CompositeDisposable _stateWatchers = new(); + private Task _stateUpdater = Task.CompletedTask; + + private readonly BehaviorSubject _stateSubject = new(new AresStruct()); + + public DelimitedStreamSensor(DeviceConnectionInfo info, ILogger logger) : base(info) + { + + _connectionInfo = info; + _logger = logger; + _dataFormat = info.DeviceSettings.Fields["DataFormat"].StringValue; + StateStream = _stateSubject.AsObservable(); + + _logger.LogInformation($"Parsing data format '{_dataFormat}'"); + _serialConnection = new System.IO.Ports.SerialPort() + { + PortName = info.SerialConnectionInfo.PortName, + BaudRate = 115200, + Parity = System.IO.Ports.Parity.None, + StopBits = System.IO.Ports.StopBits.One, + DataBits = 8, + DtrEnable = true, + RtsEnable = true, + ReadTimeout=1000, + }; + + + + _fields = _dataFormat.Split(":,\t".ToCharArray(), StringSplitOptions.None).Select(field => field.Trim()).ToArray(); + + _streamingFields = _fields + .Select((fieldName, dataIndex) => new + { + FieldName = fieldName, + DataIndex = dataIndex + }) + .Where(field => !string.IsNullOrWhiteSpace(field.FieldName)) + .Select(field => new StreamingField + { + Name = field.FieldName, + DataIndex = field.DataIndex, + StatsActive = false, + Value = null + }) + .ToArray(); + + _logger.LogInformation($"{_streamingFields.Length} fields parsed"); + + _streamingFieldsByName = _streamingFields.ToDictionary(field => field.Name, field => field, StringComparer.OrdinalIgnoreCase); + + StateSchema = AresSchemaBuilder.Empty() + .AddEntry("Name", AresSchemaBuilder.StringEntry().Build()) + .AddEntry("DataFormat", AresSchemaBuilder.StringEntry().Build()) + .AddEntry("LiveData", AresSchemaBuilder.Entry(AresDataType.Struct) + .WithStructSchema(liveData => + { + foreach (var field in _streamingFields) + { + liveData.Fields.Add(field.Name, AresSchemaBuilder.NumberEntry().AsOptional().Build()); + } + }) + .Build()) + .Build(); + + //_stateWatchers = new CompositeDisposable + // { + // _serialConnection.GetTransactionStream().Select(transaction => transaction.Response).Subscribe(UpdateLiveData) + // }; + + _logger.LogInformation($" Streaming device {Name} initialization completed"); + } + + private void UpdateLiveData(string response) + { + // TODO: move parsing into parser class (deliver key-value pairs instead of raw line) + //_logger.LogInformation($"Received line: {response}"); + var fields = response.Split(":,\t".ToCharArray(), StringSplitOptions.None).Select(field => field.Trim()).ToArray(); + foreach (var field in _streamingFields) + { + double value; + if (double.TryParse(fields[field.DataIndex], out value)) + { + field.Value = value; + if (field.StatsActive) field.Stats.AddValue(value); + } + else + { + field.Value = null; + } + } + + var next = AresStateBuilder + .From(_stateSubject.Value) + .AddStruct("LiveData", b => + { + foreach (var field in _streamingFields) + { + b.Add(field.Name, field.Value ?? 0.0); + } + }) + .Build(); + + _stateSubject.OnNext(next); + } + + private AresStruct BuildInitialState() + { + return AresStateBuilder.Create() + .Add("Name", Name) + .Add("DataFormat", _dataFormat) + .AddStruct("LiveData", liveData => + { + foreach (var field in _streamingFields) + { + /* + * If AresStateBuilder does not support null numeric values, + * use 0 initially and distinguish validity separately. + */ + liveData.Add(field.Name, field.Value ?? 0.0); + } + }) + .Build(); + } + + public override IObservable StateStream { get; } + + + public async override Task Activate(CancellationToken ct) + { + bool activated = false; + _logger.LogInformation($"Activating streaming device {Name}..."); + try + { + await Initialize(); + activated = true; + Status = new DeviceOperationalStatus { OperationalState = OperationalState.Active, Message = $"Streaming device {Name} is active!" }; + _logger.LogInformation($"Device {Name} activated"); + } + catch (Exception e) + { + Status = new DeviceOperationalStatus { OperationalState = OperationalState.Error, Message = $"Failed to initialize: {e.Message}" }; + _logger.LogError(e, $"Device {Name} activation failed"); + } + + return activated; + } + + private async Task Initialize() + { + if (_serialConnection is null) + { + _logger.LogError($"Serial device {Name}: Initialize was called, but connection was not set!"); + return; + } + + await StopUpdateLoop(); + + _stateSubject.OnNext(BuildInitialState()); + + _ = Start(); + + } + + public async Task Start() + { + await StopUpdateLoop(); + await StartUpdateLoop(TimeSpan.FromMilliseconds(500)); + } + + public async Task StartUpdateLoop(TimeSpan interval) + { + await StopUpdateLoop(); + await Task.Delay(150); + _stateGetterLoopTokenSource = new CancellationTokenSource(); + _stateUpdater = Task.Factory.StartNew(async _ => + { + Thread.CurrentThread.Name = $"Serial Device {Name} State Update Loop Thread"; + try + { + while (!_stateGetterLoopTokenSource.IsCancellationRequested) + { + try + { + if (!_serialConnection.IsOpen) _serialConnection.Open(); + //_logger.LogInformation($"Requesting live data at {DateTime.Now}"); + var liveData = _serialConnection.ReadLine(); + UpdateLiveData(liveData); + } + catch (TimeoutException) + { + _logger.LogError($"Get Live Data timed out at {DateTime.Now}"); + Status = new DeviceOperationalStatus { OperationalState = OperationalState.Active, Message = $"Get Live Data timed out at {DateTime.Now}" }; + } + catch (Exception e) + { + _logger.LogError(e, $"Get Live Data failed at {DateTime.Now}"); + await Task.Delay(150); + } + } + } + catch (ObjectDisposedException) + { + } + catch (Exception e) + { + Status = new DeviceOperationalStatus { OperationalState = OperationalState.Error, Message = $"{e.Message}" }; + } + }, + _stateGetterLoopTokenSource.Token); + } + + private async Task StopUpdateLoop() + { + _stateGetterLoopTokenSource?.Cancel(); + await _stateUpdater; + } + + public Task BeginCollectingStats(string fieldName) + { + var field = getStreamingField(fieldName); + field.Stats.ResetStats(); + field.StatsActive = true; + return Task.CompletedTask; + } + + public async ValueTask DisposeAsync() + { + //_stateWatchers.Dispose(); + await _stateGetterLoopTokenSource.CancelAsync(); + await _stateUpdater; + _stateGetterLoopTokenSource.Dispose(); + _stateSubject.OnCompleted(); + } + + public Task EndCollectingStats(string fieldName) + { + var field = getStreamingField(fieldName); + field.StatsActive = false; + return Task.CompletedTask; + } + + public override Task EnterSafeMode(CancellationToken ct) + { + // passive device. No action required + return Task.CompletedTask; + } + + public async override Task ExecuteCommand(string command, List arguments, CancellationToken token) + { + var result = new CommandResult { Success = true }; + string? fieldName = null; + + try + { + switch (command) + { + case nameof(BeginCollectingStats): + fieldName = arguments.FirstOrDefault(a => a.ArgName == "fieldName")?.ArgValue.StringValue; + if (fieldName == null) return ArgumentError(nameof(BeginCollectingStats), "fieldName", "string"); + await BeginCollectingStats(fieldName); + break; + + case nameof(EndCollectingStats): + fieldName = arguments.FirstOrDefault(a => a.ArgName == "fieldName")?.ArgValue.StringValue; + if (fieldName == null) return ArgumentError(nameof(EndCollectingStats), "fieldName", "string"); + await EndCollectingStats(fieldName); + break; + + case nameof(ResetStatistics): + fieldName = arguments.FirstOrDefault(a => a.ArgName == "fieldName")?.ArgValue.StringValue; + if (fieldName == null) return ArgumentError(nameof(ResetStatistics), "fieldName", "string"); + await ResetStatistics(fieldName); + break; + + case nameof(getCollectionCount): + fieldName = arguments.FirstOrDefault(a => a.ArgName == "fieldName")?.ArgValue.StringValue; + if (fieldName == null) return ArgumentError(nameof(getCollectionCount), "fieldName", "string"); + var collectionCount = await getCollectionCount(fieldName); + result.Result=AresValueHelper.CreateNumber(collectionCount); + break; + + case nameof(getVariance): + fieldName = arguments.FirstOrDefault(a => a.ArgName == "fieldName")?.ArgValue.StringValue; + if (fieldName == null) return ArgumentError(nameof(getVariance), "fieldName", "string"); + var variance = await getVariance(fieldName); + result.Result = AresValueHelper.CreateNumber(variance); + break; + + case nameof(getMean): + fieldName = arguments.FirstOrDefault(a => a.ArgName == "fieldName")?.ArgValue.StringValue; + if (fieldName == null) return ArgumentError(nameof(getMean), "fieldName", "string"); + var mean = await getMean(fieldName) ; + result.Result = AresValueHelper.CreateNumber(mean); + break; + + default: + result.Success = false; + result.Error = $"Unrecognized command '{command}'"; + break; + } + } + catch (Exception err) + { + result.Success = false; + result.Error = err.Message; + } + + return result; + } + + private static CommandResult ArgumentError(string commandName, string paramName, string expectedType) + { + return new CommandResult + { + Success = false, + Error = $"The {commandName} command requires a valid {expectedType} for '{paramName}', but none was provided or the type was incorrect." + }; + } + + public Task getCollectionCount(string fieldName) + { + var field = getStreamingField(fieldName); + var count = field.Stats.GetCount(); + return Task.FromResult(count); + } + + public Task getMean(string fieldName) + { + var field = getStreamingField(fieldName); + var mean = field.Stats.GetMean(); + return Task.FromResult(mean); + } + + public override Task GetSettings() + { + var response = new AresStruct().AddString("Mode", "Streaming senor"); + return Task.FromResult(response); + } + + public override Task GetState() => Task.FromResult(_stateSubject.Value); + + public Task getVariance(string fieldName) + { + var field = getStreamingField(fieldName); + var variance = field.Stats.GetSampleVariance(); + return Task.FromResult(variance); + } + + public Task ResetStatistics(string fieldName) + { + var field = getStreamingField(fieldName); + field.Stats.ResetStats(); + return Task.CompletedTask; + } + + StreamingField getStreamingField(string fieldName) + { + if (!_streamingFieldsByName.ContainsKey(fieldName)) + { + throw new ArgumentOutOfRangeException($"'{fieldName}' not a recognized data field"); + } + return _streamingFieldsByName[fieldName]; + } + + + public override Task UpdateSettings(AresStruct settings) + { + return Task.CompletedTask; + } + + protected override Task> BuildCommandDescriptorsAsync() + { + + var descriptors = new List + { + new() + { + Name = nameof(BeginCollectingStats), + Description = "Starts calculating stats on streaming data", + InputSchema = AresSchemaBuilder.Empty() + .AddEntry("fieldName", AresSchemaBuilder.StringEntry().Build()) + .Build() + }, + new() + { + Name = nameof(EndCollectingStats), + Description = "Ends calculating stats on streaming data", + InputSchema = AresSchemaBuilder.Empty() + .AddEntry("fieldName", AresSchemaBuilder.StringEntry().Build()) + .Build() + }, + new() + { + Name = nameof(getVariance), + Description = "gets the calculated variance", + InputSchema = AresSchemaBuilder.Empty() + .AddEntry("fieldName", AresSchemaBuilder.StringEntry().Build()) + .Build(), + OutputSchema = AresSchemaBuilder.NumberEntry() + .WithDescription("variance") + .Build() + }, + new() + { + Name = nameof(getMean), + Description = "gets the calculated mean", + InputSchema = AresSchemaBuilder.Empty() + .AddEntry("fieldName", AresSchemaBuilder.StringEntry().Build()) + .Build(), + OutputSchema = AresSchemaBuilder.NumberEntry() + .WithDescription("mean") + .Build() + }, + new() + { + Name = nameof(getCollectionCount), + Description = "gets the number of samples in stats calculation", + InputSchema = AresSchemaBuilder.Empty() + .AddEntry("fieldName", AresSchemaBuilder.StringEntry().Build()) + .Build(), + OutputSchema = AresSchemaBuilder.NumberEntry() + .WithDescription("count") + .Build() + }, + new() + { + Name = nameof(ResetStatistics), + Description = "clears current stats data", + InputSchema = AresSchemaBuilder.Empty() + .AddEntry("fieldName", AresSchemaBuilder.StringEntry().Build()) + .Build() + }, + }; + + return Task.FromResult(descriptors); + + } + } +} diff --git a/SerialStreamingSensor/IDelimitedSteamSensor.cs b/SerialStreamingSensor/IDelimitedSteamSensor.cs new file mode 100644 index 0000000..bd3120e --- /dev/null +++ b/SerialStreamingSensor/IDelimitedSteamSensor.cs @@ -0,0 +1,15 @@ +namespace SerialStreamingSensor +{ + public interface IDelimitedSteamSensor: IAsyncDisposable + { + + + Task BeginCollectingStats(string fieldName); + Task EndCollectingStats(string fieldName); + Task getCollectionCount(string fieldName); + Task getMean(string fieldName); + Task getVariance(string fieldName); + Task ResetStatistics(string fieldName); + + } +} diff --git a/SerialStreamingSensor/Models/StreamingField.cs b/SerialStreamingSensor/Models/StreamingField.cs new file mode 100644 index 0000000..add5fcf --- /dev/null +++ b/SerialStreamingSensor/Models/StreamingField.cs @@ -0,0 +1,28 @@ +using StreamHelper; + +namespace SerialStreamingSensor.Models +{ + public sealed class StreamingField + { + public required string Name { get; init; } + + /// + /// Position of this field in the incoming delimited data line. + /// Empty entries in the data format are counted when assigning + /// this index, even though they are not represented as fields. + /// + public int DataIndex { get; init; } + + /// + /// Most recently parsed value. + /// + public double? Value { get; set; } + + /// + /// Determines whether incoming values are added to Stats. + /// + public bool StatsActive { get; set; } + + public SimpleStreamingStats Stats { get; } = new(); + } +} \ No newline at end of file diff --git a/SerialStreamingSensor/SerialStreamingSensor.csproj b/SerialStreamingSensor/SerialStreamingSensor.csproj new file mode 100644 index 0000000..3ee1c0e --- /dev/null +++ b/SerialStreamingSensor/SerialStreamingSensor.csproj @@ -0,0 +1,66 @@ + + + + net10.0 + enable + enable + Library + + + + + + + + + Always + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + True + False + + + + + + $(ProjectDir)obj\AresStaging\ + $(ProjectDir)..\plugins\$(AssemblyName).ares + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/SerialStreamingSensor/UI/DelimitedStreamSensorControl.razor b/SerialStreamingSensor/UI/DelimitedStreamSensorControl.razor new file mode 100644 index 0000000..9b3c3eb --- /dev/null +++ b/SerialStreamingSensor/UI/DelimitedStreamSensorControl.razor @@ -0,0 +1,160 @@ +@using ReactiveUI +@using SerialStreamingSensor.UI.State +@implements IDisposable + +@if (ViewModel is null) +{ +
+ No sensor view model is available. +
+} +else +{ +
+
+
+ @GetDisplayName() +
+ +
+ @GetStatusText() +
+
+ + @if (ViewModel.SensorState.LiveData.Count == 0) + { +
+ No live-data fields are defined. +
+ } + else + { + + + @foreach (var entry in ViewModel.SensorState.LiveData) + { + + + + + + } + +
+ @FormatName(entry.Name) + + @FormatValue(entry.Value) +
+ } +
+} + +@code { + private IDisposable? _viewModelSubscription; + + [Parameter] + public DelimitedStreamSensorControlViewModel? ViewModel + { + get; + set; + } + + protected override void OnParametersSet() + { + _viewModelSubscription?.Dispose(); + _viewModelSubscription = null; + + if (ViewModel is null) + return; + + _viewModelSubscription = ViewModel + .Changed + .Subscribe( + _ => InvokeAsync(StateHasChanged)); + } + + private string GetDisplayName() + { + if (ViewModel is null) + return "Delimited Stream Sensor"; + + if (!string.IsNullOrWhiteSpace( + ViewModel.SensorState.Name)) + { + return ViewModel.SensorState.Name; + } + + return ViewModel.DeviceName; + } + + private string GetStatusText() + { + if (ViewModel?.CapturingLiveData == true) + return "LIVE"; + + if (ViewModel?.HasValidData == true) + return "CONNECTED"; + + return "INITIALIZING"; + } + + private string GetStatusClass() + { + if (ViewModel?.CapturingLiveData == true) + return "device-state connected"; + + if (ViewModel?.HasValidData == true) + return "device-state connected"; + + return "device-state initializing"; + } + + private static string GetValueClass( + DelimitedStreamSensorLiveDataEntry entry) + { + return entry.Value.HasValue + ? "parameter-value" + : "parameter-value unavailable"; + } + + private static string FormatValue(double? value) + { + return value.HasValue + ? value.Value.ToString("G6") + : "—"; + } + + private static string FormatName(string name) + { + if (string.IsNullOrWhiteSpace(name)) + return "Value"; + + var result = + new System.Text.StringBuilder( + name.Length + 8); + + for (var index = 0; + index < name.Length; + index++) + { + var current = name[index]; + + if (index > 0 && + char.IsUpper(current) && + !char.IsUpper(name[index - 1])) + { + result.Append(' '); + } + + result.Append(current); + } + + return result.ToString(); + } + + public void Dispose() + { + _viewModelSubscription?.Dispose(); + _viewModelSubscription = null; + } +} diff --git a/SerialStreamingSensor/UI/DelimitedStreamSensorControl.razor.css b/SerialStreamingSensor/UI/DelimitedStreamSensorControl.razor.css new file mode 100644 index 0000000..17971f2 --- /dev/null +++ b/SerialStreamingSensor/UI/DelimitedStreamSensorControl.razor.css @@ -0,0 +1,68 @@ +.device-container { + display: grid; + grid-template: + "header" + "state" + / minmax(0, 1fr); + justify-content: center; +} + +.title-row { + grid-area: header; + display: flex; + justify-content: space-between; + align-items: center; + width: 100%; + min-width: 0; + padding: 8px 10px; + gap: 10px; + border-radius: 6px; + box-sizing: border-box; +} + +.device-name { + flex: 1 1 auto; + min-width: 0; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + font-size:larger; +} + +.device-state { + flex: 0 0 auto; +} + +.status-connected { + background-color: #28a745; +} + +.status-initializing { + background-color: #f0ad4e; +} + +.status-error { + background-color: #dc3545; +} + +.parameter-table { + grid-area: state; + width: 95%; + min-width: 0; + border-collapse: collapse; + table-layout: auto; +} + +.parameter-name { + min-width: 80px; + white-space: normal; + overflow-wrap: anywhere; + font-size: smaller; +} + +.parameter-value { + width: 120px; + min-width: 90px; + white-space: nowrap; + font-size: smaller; +} diff --git a/SerialStreamingSensor/UI/DelimitedStreamSensorControlViewModel.cs b/SerialStreamingSensor/UI/DelimitedStreamSensorControlViewModel.cs new file mode 100644 index 0000000..3ff034f --- /dev/null +++ b/SerialStreamingSensor/UI/DelimitedStreamSensorControlViewModel.cs @@ -0,0 +1,119 @@ +using Ares.Toolkit.Device.UI; +using Microsoft.Extensions.Logging; +using ReactiveUI; +using SerialStreamingSensor.UI.State; +using System.Reactive.Linq; + +namespace SerialStreamingSensor.UI +{ + public sealed class DelimitedStreamSensorControlViewModel + : DeviceUnitControlViewModel, + IAsyncDisposable + { + private readonly ILogger< + DelimitedStreamSensorControlViewModel> _logger; + + private IDisposable? _stateSubscription; + + private DelimitedStreamSensorState _sensorState = new(); + private bool _hasValidData; + private bool _capturingLiveData; + private bool _disposed; + + public DelimitedStreamSensorControlViewModel( + DelimitedStreamSensor sensor, + ILogger logger) + : base(sensor) + { + _logger = logger; + + ViewType = typeof(DelimitedStreamSensorControl); + DefaultWidth = 19; + + _stateSubscription = sensor.StateStream + .Select( + DelimitedStreamSensorStateMapper.FromAresStruct) + .Subscribe( + UpdateState, + HandleStateStreamError, + HandleStateStreamCompleted); + } + + public DelimitedStreamSensorState SensorState + { + get => _sensorState; + + private set => + this.RaiseAndSetIfChanged( + ref _sensorState, + value); + } + + public bool HasValidData + { + get => _hasValidData; + + private set => + this.RaiseAndSetIfChanged( + ref _hasValidData, + value); + } + + public bool CapturingLiveData + { + get => _capturingLiveData; + + private set => + this.RaiseAndSetIfChanged( + ref _capturingLiveData, + value); + } + + private void UpdateState( + DelimitedStreamSensorState? state) + { + if (state is null) + { + HasValidData = false; + CapturingLiveData = false; + return; + } + + SensorState = state; + HasValidData = true; + CapturingLiveData = true; + } + + private void HandleStateStreamError( + Exception exception) + { + _logger.LogError( + exception, + "State stream failed for {Name}", + DeviceName); + + HasValidData = false; + CapturingLiveData = false; + } + + private void HandleStateStreamCompleted() + { + CapturingLiveData = false; + } + + public ValueTask DisposeAsync() + { + if (_disposed) + return ValueTask.CompletedTask; + + _disposed = true; + + _stateSubscription?.Dispose(); + _stateSubscription = null; + + GC.SuppressFinalize(this); + + return ValueTask.CompletedTask; + } + } +} \ No newline at end of file diff --git a/SerialStreamingSensor/UI/State/DelimitedStreamSensorState.cs b/SerialStreamingSensor/UI/State/DelimitedStreamSensorState.cs new file mode 100644 index 0000000..3b1d93a --- /dev/null +++ b/SerialStreamingSensor/UI/State/DelimitedStreamSensorState.cs @@ -0,0 +1,20 @@ +namespace SerialStreamingSensor.UI.State +{ + public sealed class DelimitedStreamSensorState + { + public string Name { get; init; } = string.Empty; + + public IReadOnlyList LiveData + { + get; + init; + } = Array.Empty(); + } + + public sealed class DelimitedStreamSensorLiveDataEntry + { + public string Name { get; init; } = string.Empty; + + public double? Value { get; init; } + } +} \ No newline at end of file diff --git a/SerialStreamingSensor/UI/State/DelimitedStreamSensorStateMapper.cs b/SerialStreamingSensor/UI/State/DelimitedStreamSensorStateMapper.cs new file mode 100644 index 0000000..9e38949 --- /dev/null +++ b/SerialStreamingSensor/UI/State/DelimitedStreamSensorStateMapper.cs @@ -0,0 +1,47 @@ +using Ares.Datamodel; +using Ares.Datamodel.Extensions; + +namespace SerialStreamingSensor.UI.State +{ + public static class DelimitedStreamSensorStateMapper + { + public static DelimitedStreamSensorState FromAresStruct( + AresStruct state) + { + var model = new DelimitedStreamSensorState + { + + Name = state.Fields + .GetValueOrDefault("Name")? + .StringValue ?? string.Empty + }; + + if (state.Fields.TryGetValue( + "LiveData", + out var liveDataValue) && + liveDataValue.StructValue is not null) + { + model = new DelimitedStreamSensorState + { + Name = model.Name, + + LiveData = liveDataValue.StructValue.Fields + .Where(field => + !string.IsNullOrWhiteSpace(field.Key)) + .Select(field => + new DelimitedStreamSensorLiveDataEntry + { + Name = field.Key, + + Value = field.Value.HasNumberValue + ? field.Value.NumberValue + : null + }) + .ToArray() + }; + } + + return model; + } + } +} \ No newline at end of file diff --git a/SerialStreamingSensor/manifest.yaml b/SerialStreamingSensor/manifest.yaml new file mode 100644 index 0000000..5bd464b --- /dev/null +++ b/SerialStreamingSensor/manifest.yaml @@ -0,0 +1,23 @@ +# ARES Device Driver Manifest +# Target: Steaming Serial Sensor + +# Basic Metadata +device_type_name: "Delimited Stream Sensor" +driver_id: "delimited_stream_sensor_v1" +connection_type: "Serial" + +# The DLL that contains the MassFlowController and ViewModel classes +assembly_name: "SerialStreamingSensor.dll" + +# Serial connection specifics (Parsed only if connection_type is "Serial") +serial_settings: + variable_baud_rate: true + allowed_baud_rates: [9600, 19200, 38400, 57600, 115200] + default_baud_rate: 115200 # Keeps the UI friendly by pre-filling + +# Settings required when adding the device through the ARES UI +settings: + - key: "DataFormat" + display_name: "Data Format" + type: "string" + description: "streaming data line format." diff --git a/StreamHelper/SimpleStreamingStats.cs b/StreamHelper/SimpleStreamingStats.cs new file mode 100644 index 0000000..851a7de --- /dev/null +++ b/StreamHelper/SimpleStreamingStats.cs @@ -0,0 +1,431 @@ +using System; + +namespace StreamHelper +{ + /// + /// Calculates descriptive statistics incrementally without buffering + /// individual samples. + /// + /// Regression is performed against the zero-based sample index: + /// x = 0, 1, 2, ... + /// + /// This class uses constant memory regardless of the number of samples. + /// + public class SimpleStreamingStats + { + private readonly object _syncRoot = new(); + + private long _count; + + // Online central moments. + private double _mean; + private double _m2; + private double _m3; + private double _m4; + + // Sum of squares of raw values, used for RMS. + private double _sumSquares; + + private double _minimum; + private double _maximum; + + // Running regression values. + private double _meanX; + private double _meanY; + private double _sxx; + private double _syy; + private double _sxy; + + /// + /// Adds one sample to the running statistics. + /// + public void AddValue(double value) + { + if (!double.IsFinite(value)) + throw new ArgumentOutOfRangeException( + nameof(value), + "The sample must be a finite number."); + + lock (_syncRoot) + { + long previousCount = _count; + _count++; + + if (_count == 1) + { + _minimum = value; + _maximum = value; + } + else + { + if (value < _minimum) + _minimum = value; + + if (value > _maximum) + _maximum = value; + } + + _sumSquares += value * value; + + UpdateMoments(value, previousCount); + UpdateRegression(value, previousCount); + } + } + + /// + /// Clears all accumulated statistics. + /// + public void ResetStats() + { + lock (_syncRoot) + { + _count = 0; + + _mean = 0.0; + _m2 = 0.0; + _m3 = 0.0; + _m4 = 0.0; + + _sumSquares = 0.0; + + _minimum = 0.0; + _maximum = 0.0; + + _meanX = 0.0; + _meanY = 0.0; + _sxx = 0.0; + _syy = 0.0; + _sxy = 0.0; + } + } + + public long GetCount() + { + lock (_syncRoot) + return _count; + } + + public double GetMean() + { + lock (_syncRoot) + return _count > 0 ? _mean : double.NaN; + } + + /// + /// Returns variance using N as the denominator. + /// + public double GetPopulationVariance() + { + lock (_syncRoot) + return _count > 0 + ? _m2 / _count + : double.NaN; + } + + /// + /// Returns variance using N - 1 as the denominator. + /// + public double GetSampleVariance() + { + lock (_syncRoot) + return _count > 1 + ? _m2 / (_count - 1) + : double.NaN; + } + + public double GetPopulationStandardDeviation() + { + lock (_syncRoot) + { + return _count > 0 + ? Math.Sqrt(Math.Max(0.0, _m2 / _count)) + : double.NaN; + } + } + + public double GetSampleStandardDeviation() + { + lock (_syncRoot) + { + return _count > 1 + ? Math.Sqrt(Math.Max(0.0, _m2 / (_count - 1))) + : double.NaN; + } + } + + public double GetMinimum() + { + lock (_syncRoot) + return _count > 0 ? _minimum : double.NaN; + } + + public double GetMaximum() + { + lock (_syncRoot) + return _count > 0 ? _maximum : double.NaN; + } + + public double GetRange() + { + lock (_syncRoot) + return _count > 0 + ? _maximum - _minimum + : double.NaN; + } + + public double GetRootMeanSquare() + { + lock (_syncRoot) + { + return _count > 0 + ? Math.Sqrt(_sumSquares / _count) + : double.NaN; + } + } + + /// + /// Returns adjusted sample skewness. + /// Requires at least three samples. + /// + public double GetSkewness() + { + lock (_syncRoot) + { + if (_count < 3 || _m2 <= 0.0) + return double.NaN; + + double n = _count; + + return Math.Sqrt(n * (n - 1.0)) + / (n - 2.0) + * (_m3 / Math.Pow(_m2, 1.5)); + } + } + + /// + /// Returns bias-corrected excess kurtosis. + /// + /// A normal distribution has an excess kurtosis near zero. + /// Requires at least four samples. + /// + public double GetExcessKurtosis() + { + lock (_syncRoot) + { + if (_count < 4 || _m2 <= 0.0) + return double.NaN; + + double n = _count; + + double rawKurtosis = + n * _m4 / (_m2 * _m2); + + return ((n - 1.0) / + ((n - 2.0) * (n - 3.0))) + * + ((n + 1.0) * (rawKurtosis - 3.0) + 6.0); + } + } + + /// + /// Returns the slope per sample interval. + /// + /// For example, if samples are one second apart, this is the slope + /// per second. + /// + public double GetSlope() + { + lock (_syncRoot) + { + return _count > 1 && _sxx > 0.0 + ? _sxy / _sxx + : double.NaN; + } + } + + /// + /// Returns the slope converted to a specified time unit. + /// + /// Example: + /// sampleIntervalSeconds = 0.1 + /// returns units per second. + /// + public double GetSlopePerTime(double sampleInterval) + { + if (!double.IsFinite(sampleInterval) || sampleInterval <= 0.0) + throw new ArgumentOutOfRangeException( + nameof(sampleInterval), + "Sample interval must be finite and greater than zero."); + + lock (_syncRoot) + { + return _count > 1 && _sxx > 0.0 + ? (_sxy / _sxx) / sampleInterval + : double.NaN; + } + } + + public double GetIntercept() + { + lock (_syncRoot) + { + if (_count < 2 || _sxx <= 0.0) + return double.NaN; + + double slope = _sxy / _sxx; + + return _meanY - slope * _meanX; + } + } + + public double GetCorrelation() + { + lock (_syncRoot) + { + if (_count < 2 || _sxx <= 0.0 || _syy <= 0.0) + return double.NaN; + + double denominator = Math.Sqrt(_sxx * _syy); + + if (denominator <= 0.0) + return double.NaN; + + return Math.Clamp( + _sxy / denominator, + -1.0, + 1.0); + } + } + + public double GetRSquared() + { + lock (_syncRoot) + { + if (_count < 2 || _sxx <= 0.0 || _syy <= 0.0) + return double.NaN; + + double denominator = _sxx * _syy; + + if (denominator <= 0.0) + return double.NaN; + + double rSquared = + (_sxy * _sxy) / denominator; + + return Math.Clamp(rSquared, 0.0, 1.0); + } + } + + /// + /// Returns the residual standard error around the linear fit. + /// Requires at least three samples. + /// + public double GetRegressionStandardError() + { + lock (_syncRoot) + { + if (_count < 3 || _sxx <= 0.0) + return double.NaN; + + double explainedSumSquares = + (_sxy * _sxy) / _sxx; + + double residualSumSquares = + Math.Max(0.0, _syy - explainedSumSquares); + + return Math.Sqrt( + residualSumSquares / (_count - 2)); + } + } + + /// + /// Predicts the value at a zero-based sample index. + /// + public double GetPredictedValue(double sampleIndex) + { + if (!double.IsFinite(sampleIndex)) + throw new ArgumentOutOfRangeException( + nameof(sampleIndex)); + + lock (_syncRoot) + { + if (_count < 2 || _sxx <= 0.0) + return double.NaN; + + double slope = _sxy / _sxx; + double intercept = _meanY - slope * _meanX; + + return intercept + slope * sampleIndex; + } + } + + private void UpdateMoments( + double value, + long previousCount) + { + /* + * Online update of the first four central moments. + * + * This is an extension of Welford's algorithm and avoids + * subtracting two large, nearly equal raw sums. + */ + double n = _count; + double delta = value - _mean; + double deltaOverN = delta / n; + double deltaOverNSquared = + deltaOverN * deltaOverN; + + double term1 = + delta * deltaOverN * previousCount; + + _mean += deltaOverN; + + _m4 += + term1 + * deltaOverNSquared + * (n * n - 3.0 * n + 3.0) + + + 6.0 + * deltaOverNSquared + * _m2 + - + 4.0 + * deltaOverN + * _m3; + + _m3 += + term1 + * deltaOverN + * (n - 2.0) + - + 3.0 + * deltaOverN + * _m2; + + _m2 += term1; + } + + private void UpdateRegression( + double value, + long sampleIndex) + { + double x = sampleIndex; + double y = value; + double n = _count; + + double deltaX = x - _meanX; + double deltaY = y - _meanY; + + _meanX += deltaX / n; + _meanY += deltaY / n; + + /* + * Use the updated means for the second part of the online + * covariance update. + */ + _sxx += deltaX * (x - _meanX); + _syy += deltaY * (y - _meanY); + _sxy += deltaX * (y - _meanY); + } + } +} \ No newline at end of file diff --git a/StreamHelper/StreamHelper.csproj b/StreamHelper/StreamHelper.csproj new file mode 100644 index 0000000..a0bed63 --- /dev/null +++ b/StreamHelper/StreamHelper.csproj @@ -0,0 +1,10 @@ + + + + net10.0 + enable + enable + Library + + + diff --git a/plugins/SerialStreamingSensor.ares.archive b/plugins/SerialStreamingSensor.ares.archive new file mode 100644 index 0000000..bf966ba Binary files /dev/null and b/plugins/SerialStreamingSensor.ares.archive differ