diff --git a/src/Daqifi.Core.Tests/Channel/AnalogChannelTests.cs b/src/Daqifi.Core.Tests/Channel/AnalogChannelTests.cs index 50a7daa5..c10d84e1 100644 --- a/src/Daqifi.Core.Tests/Channel/AnalogChannelTests.cs +++ b/src/Daqifi.Core.Tests/Channel/AnalogChannelTests.cs @@ -139,6 +139,39 @@ public void SetActiveSample_RaisesSampleReceivedEvent() Assert.Equal(timestamp, receivedSample.Timestamp); } + [Fact] + public void SetActiveSample_EventArgsCarryRaisingChannel() + { + // Arrange + var channel = new AnalogChannel(3); + IChannel? eventChannel = null; + channel.SampleReceived += (_, args) => eventChannel = args.Channel; + + // Act + channel.SetActiveSample(1.0, DateTime.UtcNow); + + // Assert + Assert.Same(channel, eventChannel); + } + + [Fact] + public void SetActiveSample_WithFullSample_PreservesRawValueAndDeviceTimestamp() + { + // Arrange + var channel = new AnalogChannel(0); + IDataSample? received = null; + channel.SampleReceived += (_, args) => received = args.Sample; + var sample = new DataSample(DateTime.UtcNow, 2.5, rawValue: 128, deviceTimestamp: 555u); + + // Act + channel.SetActiveSample(sample); + + // Assert + Assert.Same(sample, channel.ActiveSample); + Assert.Equal(128, received!.RawValue); + Assert.Equal(555u, received.DeviceTimestamp); + } + [Fact] public async Task SetActiveSample_IsThreadSafe() { diff --git a/src/Daqifi.Core.Tests/Channel/DataSampleTests.cs b/src/Daqifi.Core.Tests/Channel/DataSampleTests.cs index 11b39802..123b6c3d 100644 --- a/src/Daqifi.Core.Tests/Channel/DataSampleTests.cs +++ b/src/Daqifi.Core.Tests/Channel/DataSampleTests.cs @@ -30,6 +30,29 @@ public void Constructor_WithParameters_SetsProvidedValues() Assert.Equal(value, sample.Value); } + [Fact] + public void Constructor_WithNoDecodeMetadata_DefaultsRawValueAndDeviceTimestamp() + { + // Samples not produced by the decode pipeline have no raw value or device timestamp. + var sample = new DataSample(DateTime.UtcNow, 1.0); + + Assert.Null(sample.RawValue); + Assert.Null(sample.DeviceTimestamp); + } + + [Fact] + public void Constructor_WithDecodeMetadata_SetsRawValueAndDeviceTimestamp() + { + var timestamp = new DateTime(2025, 10, 20, 12, 0, 0, DateTimeKind.Utc); + + var sample = new DataSample(timestamp, 1.25, rawValue: 4321, deviceTimestamp: 987654u); + + Assert.Equal(timestamp, sample.Timestamp); + Assert.Equal(1.25, sample.Value); + Assert.Equal(4321, sample.RawValue); + Assert.Equal(987654u, sample.DeviceTimestamp); + } + [Fact] public void Value_CanBeModified() { diff --git a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs new file mode 100644 index 00000000..35ec6258 --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs @@ -0,0 +1,381 @@ +using Daqifi.Core.Channel; +using Daqifi.Core.Communication.Messages; +using Daqifi.Core.Device; +using Google.Protobuf; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using Xunit; + +namespace Daqifi.Core.Tests.Device; + +/// +/// Unit tests for 's decoded per-frame sample pipeline: +/// stream frames are decoded into per-channel samples that drive . +/// +public class DaqifiStreamingDeviceDecodeTests +{ + #region Analog decoding + + [Fact] + public void Decode_UsbFloatPath_UsesFloatsDirectlyWithNoRawValue() + { + // Arrange: 3 analog channels, enable AI0 and AI2 (leaving a gap at AI1). + var device = CreateStreamingDevice(analogCount: 3); + var ai0 = AnalogChannel(device, 0); + var ai2 = AnalogChannel(device, 2); + ai0.IsEnabled = true; + ai2.IsEnabled = true; + device.StartStreaming(); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 4242 }; + frame.AnalogInDataFloat.Add(1.5f); + frame.AnalogInDataFloat.Add(2.5f); + + // Act + device.InvokeStreamMessage(frame); + + // Assert: values map to enabled channels in ascending channel-number order. + Assert.NotNull(ai0.ActiveSample); + Assert.Equal(1.5, ai0.ActiveSample!.Value); + Assert.Null(ai0.ActiveSample.RawValue); // pre-scaled float => no raw ADC count + Assert.Equal(4242u, ai0.ActiveSample.DeviceTimestamp); + + Assert.NotNull(ai2.ActiveSample); + Assert.Equal(2.5, ai2.ActiveSample!.Value); + Assert.Null(ai2.ActiveSample.RawValue); + + // The disabled channel between them received nothing. + Assert.Null(AnalogChannel(device, 1).ActiveSample); + } + + [Fact] + public void Decode_WifiRawPath_AppliesChannelCalibrationAndPreservesRawCount() + { + // Arrange: give the channels a non-identity port range so scaling is observable. + var device = CreateStreamingDevice(analogCount: 2, portRange: 10.0f, resolution: 65535); + var ai0 = AnalogChannel(device, 0); + var ai1 = AnalogChannel(device, 1); + ai0.IsEnabled = true; + ai1.IsEnabled = true; + device.StartStreaming(); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 7 }; + frame.AnalogInData.Add(1000); + frame.AnalogInData.Add(2000); + + // Act + device.InvokeStreamMessage(frame); + + // Assert: decode applied the channel's own calibration and preserved the raw count. + Assert.NotNull(ai0.ActiveSample); + Assert.Equal(ai0.GetScaledValue(1000), ai0.ActiveSample!.Value); + Assert.Equal(1000, ai0.ActiveSample.RawValue); + Assert.NotEqual(1000.0, ai0.ActiveSample.Value); // scaling actually happened + + Assert.NotNull(ai1.ActiveSample); + Assert.Equal(ai1.GetScaledValue(2000), ai1.ActiveSample!.Value); + Assert.Equal(2000, ai1.ActiveSample.RawValue); + } + + [Fact] + public void Decode_MapsValuesByChannelNumberNotEnableOrder() + { + // Enable the higher-numbered channel "first" to prove ordering is by channel number. + var device = CreateStreamingDevice(analogCount: 3); + var ai2 = AnalogChannel(device, 2); + var ai0 = AnalogChannel(device, 0); + ai2.IsEnabled = true; + ai0.IsEnabled = true; + device.StartStreaming(); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 1 }; + frame.AnalogInDataFloat.Add(10f); // first value -> lowest channel number (AI0) + frame.AnalogInDataFloat.Add(20f); // second value -> AI2 + + device.InvokeStreamMessage(frame); + + Assert.Equal(10.0, ai0.ActiveSample!.Value); + Assert.Equal(20.0, ai2.ActiveSample!.Value); + } + + [Fact] + public void Decode_RaisesSampleReceivedWithChannelReference() + { + var device = CreateStreamingDevice(analogCount: 1); + var ai0 = AnalogChannel(device, 0); + ai0.IsEnabled = true; + device.StartStreaming(); + + SampleReceivedEventArgs? captured = null; + ai0.SampleReceived += (_, e) => captured = e; + + var frame = new DaqifiOutMessage { MsgTimeStamp = 99 }; + frame.AnalogInDataFloat.Add(3.14f); + + device.InvokeStreamMessage(frame); + + Assert.NotNull(captured); + Assert.Same(ai0, captured!.Channel); + Assert.Equal(3.14, captured.Sample.Value, 5); + Assert.Equal(99u, captured.Sample.DeviceTimestamp); + } + + #endregion + + #region Digital decoding + + [Fact] + public void Decode_Digital_UnpacksBitsPerChannel() + { + var device = CreateStreamingDevice(analogCount: 0, digitalCount: 4); + var dio = Enumerable.Range(0, 4).Select(n => DigitalChannel(device, n)).ToList(); + foreach (var d in dio) d.IsEnabled = true; + device.StartStreaming(); + + // 0b1010 => DIO0=low, DIO1=high, DIO2=low, DIO3=high + var frame = new DaqifiOutMessage { MsgTimeStamp = 5 }; + frame.DigitalData = ByteString.CopyFrom(new byte[] { 0b1010 }); + + device.InvokeStreamMessage(frame); + + Assert.Equal(0.0, dio[0].ActiveSample!.Value); + Assert.Equal(1.0, dio[1].ActiveSample!.Value); + Assert.Equal(0.0, dio[2].ActiveSample!.Value); + Assert.Equal(1.0, dio[3].ActiveSample!.Value); + Assert.Equal(1, dio[1].ActiveSample!.RawValue); + Assert.Equal(5u, dio[1].ActiveSample!.DeviceTimestamp); + } + + [Fact] + public void Decode_Digital_SkipsOutputDirectionChannels() + { + var device = CreateStreamingDevice(analogCount: 0, digitalCount: 2); + var dio0 = DigitalChannel(device, 0); + var dio1 = DigitalChannel(device, 1); + dio0.IsEnabled = true; + dio1.IsEnabled = true; + dio1.Direction = ChannelDirection.Output; // output channels are not sampled + device.StartStreaming(); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 5 }; + frame.DigitalData = ByteString.CopyFrom(new byte[] { 0b11 }); + + device.InvokeStreamMessage(frame); + + Assert.NotNull(dio0.ActiveSample); + Assert.Equal(1.0, dio0.ActiveSample!.Value); + Assert.Null(dio1.ActiveSample); // output channel skipped + } + + [Fact] + public void Decode_Digital_BeyondTwoBytes_ReadsCorrectByteWithoutWrapping() + { + // Regression for Qodo #279: with >16 enabled digital channels / >2 payload bytes, bit + // position i must map to byte i/8, bit i%8 — not wrap byte 1 for i>=16. + var device = CreateStreamingDevice(analogCount: 0, digitalCount: 17); + var dio = Enumerable.Range(0, 17).Select(n => DigitalChannel(device, n)).ToList(); + foreach (var d in dio) d.IsEnabled = true; + device.StartStreaming(); + + // Only channel index 16 high: byte 2 bit 0. A wrapping decoder would read byte 1 bit 0 (low). + var frame = new DaqifiOutMessage { MsgTimeStamp = 5 }; + frame.DigitalData = ByteString.CopyFrom(new byte[] { 0x00, 0x00, 0b0000_0001 }); + + device.InvokeStreamMessage(frame); + + Assert.Equal(1.0, dio[16].ActiveSample!.Value); + for (var i = 0; i < 16; i++) + { + Assert.Equal(0.0, dio[i].ActiveSample!.Value); + } + } + + [Fact] + public void Decode_Digital_MoreChannelsThanPayloadBits_StopsInsteadOfForcingLow() + { + // With a single payload byte (8 bits) but more enabled channels, channels past the + // payload get no sample rather than a bogus "low" reading. + var device = CreateStreamingDevice(analogCount: 0, digitalCount: 10); + var dio = Enumerable.Range(0, 10).Select(n => DigitalChannel(device, n)).ToList(); + foreach (var d in dio) d.IsEnabled = true; + device.StartStreaming(); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 5 }; + frame.DigitalData = ByteString.CopyFrom(new byte[] { 0xFF }); // 8 bits, channels 0-7 + + device.InvokeStreamMessage(frame); + + for (var i = 0; i < 8; i++) + { + Assert.Equal(1.0, dio[i].ActiveSample!.Value); + } + Assert.Null(dio[8].ActiveSample); + Assert.Null(dio[9].ActiveSample); + } + + #endregion + + #region Gating and resilience + + [Fact] + public void Decode_WhenNotStreaming_DoesNotProduceSamples() + { + var device = CreateStreamingDevice(analogCount: 1); + var ai0 = AnalogChannel(device, 0); + ai0.IsEnabled = true; + // Note: StartStreaming intentionally NOT called. + + var raised = false; + ai0.SampleReceived += (_, _) => raised = true; + + var frame = new DaqifiOutMessage { MsgTimeStamp = 1 }; + frame.AnalogInDataFloat.Add(1f); + + device.InvokeStreamMessage(frame); + + Assert.Null(ai0.ActiveSample); + Assert.False(raised); + } + + [Fact] + public void Decode_StillReRaisesRawMessageReceived() + { + // Existing consumers that hand-demux the raw frame must keep working. + var device = CreateStreamingDevice(analogCount: 1); + AnalogChannel(device, 0).IsEnabled = true; + device.StartStreaming(); + + MessageReceivedEventArgs? raw = null; + device.MessageReceived += (_, e) => raw = e; + + var frame = new DaqifiOutMessage { MsgTimeStamp = 1 }; + frame.AnalogInDataFloat.Add(1f); + + device.InvokeStreamMessage(frame); + + Assert.NotNull(raw); + } + + [Fact] + public void Decode_MoreValuesThanChannels_MapsAvailableWithoutThrowing() + { + var device = CreateStreamingDevice(analogCount: 1); + var ai0 = AnalogChannel(device, 0); + ai0.IsEnabled = true; + device.StartStreaming(); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 1 }; + frame.AnalogInDataFloat.Add(1f); + frame.AnalogInDataFloat.Add(2f); // extra value with no channel to receive it + + var ex = Record.Exception(() => device.InvokeStreamMessage(frame)); + + Assert.Null(ex); + Assert.Equal(1.0, ai0.ActiveSample!.Value); + } + + [Fact] + public void Decode_FewerValuesThanChannels_MapsAvailableWithoutThrowing() + { + var device = CreateStreamingDevice(analogCount: 2); + var ai0 = AnalogChannel(device, 0); + var ai1 = AnalogChannel(device, 1); + ai0.IsEnabled = true; + ai1.IsEnabled = true; + device.StartStreaming(); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 1 }; + frame.AnalogInDataFloat.Add(1f); // only one value for two enabled channels + + var ex = Record.Exception(() => device.InvokeStreamMessage(frame)); + + Assert.Null(ex); + Assert.Equal(1.0, ai0.ActiveSample!.Value); + Assert.Null(ai1.ActiveSample); + } + + [Fact] + public void Decode_CarriesDeviceTimestampVerbatimAcrossFrames() + { + var device = CreateStreamingDevice(analogCount: 1); + var ai0 = AnalogChannel(device, 0); + ai0.IsEnabled = true; + device.StartStreaming(); + + var first = new DaqifiOutMessage { MsgTimeStamp = 1000 }; + first.AnalogInDataFloat.Add(1f); + device.InvokeStreamMessage(first); + var firstHost = ai0.ActiveSample!.Timestamp; + Assert.Equal(1000u, ai0.ActiveSample!.DeviceTimestamp); + + var second = new DaqifiOutMessage { MsgTimeStamp = 2000 }; + second.AnalogInDataFloat.Add(2f); + device.InvokeStreamMessage(second); + Assert.Equal(2000u, ai0.ActiveSample!.DeviceTimestamp); + + // Host timestamp advances monotonically as device ticks increase. + Assert.True(ai0.ActiveSample!.Timestamp >= firstHost); + } + + #endregion + + #region Helpers + + private static DecodableStreamingDevice CreateStreamingDevice( + int analogCount, + int digitalCount = 0, + float? portRange = null, + uint resolution = 65535) + { + var device = new DecodableStreamingDevice("TestDevice"); + device.Connect(); + + var status = new DaqifiOutMessage + { + AnalogInPortNum = (uint)analogCount, + DigitalPortNum = (uint)digitalCount, + AnalogInRes = resolution, + }; + + for (var i = 0; i < analogCount; i++) + { + status.AnalogInPortRange.Add(portRange ?? 1.0f); + } + + device.PopulateChannelsFromStatus(status); + return device; + } + + private static IAnalogChannel AnalogChannel(DaqifiStreamingDevice device, int number) => + (IAnalogChannel)device.Channels.First(c => c.Type == ChannelType.Analog && c.ChannelNumber == number); + + private static IChannel DigitalChannel(DaqifiStreamingDevice device, int number) => + device.Channels.First(c => c.Type == ChannelType.Digital && c.ChannelNumber == number); + + /// + /// A that captures sent SCPI commands (so streaming + /// setup does not require a real transport) and exposes the protected stream handler so a + /// frame can be injected directly. + /// + private sealed class DecodableStreamingDevice : DaqifiStreamingDevice + { + public DecodableStreamingDevice(string name, IPAddress? ipAddress = null) : base(name, ipAddress) + { + } + + public List> SentMessages { get; } = new(); + + public void InvokeStreamMessage(DaqifiOutMessage message) => OnStreamMessageReceived(message); + + public override void Send(IOutboundMessage message) + { + if (message is IOutboundMessage stringMessage) + { + SentMessages.Add(stringMessage); + } + } + } + + #endregion +} diff --git a/src/Daqifi.Core/Channel/AnalogChannel.cs b/src/Daqifi.Core/Channel/AnalogChannel.cs index 02d749fe..a277e44b 100644 --- a/src/Daqifi.Core/Channel/AnalogChannel.cs +++ b/src/Daqifi.Core/Channel/AnalogChannel.cs @@ -181,14 +181,24 @@ public double GetScaledValue(int rawValue) /// The timestamp when the sample was taken. public void SetActiveSample(double value, DateTime timestamp) { - var sample = new DataSample(timestamp, value); + SetActiveSample(new DataSample(timestamp, value)); + } + + /// + /// Sets the active sample for this channel to a fully-formed sample and triggers the + /// SampleReceived event. + /// + /// The sample to set as active. + public void SetActiveSample(IDataSample sample) + { + ArgumentNullException.ThrowIfNull(sample); lock (_lock) { _activeSample = sample; } - SampleReceived?.Invoke(this, new SampleReceivedEventArgs(sample)); + SampleReceived?.Invoke(this, new SampleReceivedEventArgs(this, sample)); } /// diff --git a/src/Daqifi.Core/Channel/DataSample.cs b/src/Daqifi.Core/Channel/DataSample.cs index fb6f2092..581f2d72 100644 --- a/src/Daqifi.Core/Channel/DataSample.cs +++ b/src/Daqifi.Core/Channel/DataSample.cs @@ -15,6 +15,18 @@ public class DataSample : IDataSample /// public double Value { get; set; } + /// + /// Gets the raw device value this sample was decoded from, or null when the device + /// supplied an already-scaled value or the sample was not produced by the decode pipeline. + /// + public int? RawValue { get; init; } + + /// + /// Gets the raw device timestamp (clock ticks) of the stream frame this sample was decoded + /// from, or null when the sample was not produced from a stream frame. + /// + public uint? DeviceTimestamp { get; init; } + /// /// Initializes a new instance of the class. /// @@ -35,6 +47,21 @@ public DataSample(DateTime timestamp, double value) Value = value; } + /// + /// Initializes a new instance of the class with decode metadata. + /// + /// The host timestamp when the sample was taken. + /// The scaled value of the sample. + /// The raw device value the sample was decoded from, or null if none. + /// The raw device timestamp (clock ticks) of the source stream frame, or null if none. + public DataSample(DateTime timestamp, double value, int? rawValue, uint? deviceTimestamp) + { + Timestamp = timestamp; + Value = value; + RawValue = rawValue; + DeviceTimestamp = deviceTimestamp; + } + /// /// Returns a string representation of the data sample. /// diff --git a/src/Daqifi.Core/Channel/DigitalChannel.cs b/src/Daqifi.Core/Channel/DigitalChannel.cs index 5d04ca20..8c769c4c 100644 --- a/src/Daqifi.Core/Channel/DigitalChannel.cs +++ b/src/Daqifi.Core/Channel/DigitalChannel.cs @@ -122,14 +122,24 @@ public DigitalChannel(int channelNumber) /// The timestamp when the sample was taken. public void SetActiveSample(double value, DateTime timestamp) { - var sample = new DataSample(timestamp, value); + SetActiveSample(new DataSample(timestamp, value)); + } + + /// + /// Sets the active sample for this channel to a fully-formed sample and triggers the + /// SampleReceived event. + /// + /// The sample to set as active. + public void SetActiveSample(IDataSample sample) + { + ArgumentNullException.ThrowIfNull(sample); lock (_lock) { _activeSample = sample; } - SampleReceived?.Invoke(this, new SampleReceivedEventArgs(sample)); + SampleReceived?.Invoke(this, new SampleReceivedEventArgs(this, sample)); } /// diff --git a/src/Daqifi.Core/Channel/IChannel.cs b/src/Daqifi.Core/Channel/IChannel.cs index faf2a724..31d94e6c 100644 --- a/src/Daqifi.Core/Channel/IChannel.cs +++ b/src/Daqifi.Core/Channel/IChannel.cs @@ -46,4 +46,12 @@ public interface IChannel /// The raw or scaled value. /// The timestamp when the sample was taken. void SetActiveSample(double value, DateTime timestamp); + + /// + /// Sets the active sample for this channel to a fully-formed sample and triggers the + /// SampleReceived event. Used by the decode pipeline to carry raw value and device + /// timestamp metadata that cannot. + /// + /// The sample to set as active. + void SetActiveSample(IDataSample sample); } diff --git a/src/Daqifi.Core/Channel/IDataSample.cs b/src/Daqifi.Core/Channel/IDataSample.cs index ac4c0581..f27a1102 100644 --- a/src/Daqifi.Core/Channel/IDataSample.cs +++ b/src/Daqifi.Core/Channel/IDataSample.cs @@ -6,12 +6,29 @@ namespace Daqifi.Core.Channel; public interface IDataSample { /// - /// Gets the timestamp when the sample was taken. + /// Gets the host (system) timestamp when the sample was taken. For streamed samples this is + /// reconstructed from the device clock (rollover-aware) rather than the arrival time. /// DateTime Timestamp { get; } /// - /// Gets or sets the value of the sample. + /// Gets or sets the scaled value of the sample (e.g. volts for an analog channel, or 0/1 for + /// a digital channel). /// double Value { get; set; } + + /// + /// Gets the raw device value this sample was decoded from, when one exists: the raw ADC count + /// for a calibration-scaled analog sample, or the 0/1 bit for a digital sample. It is + /// null when the device supplied an already-scaled value (e.g. the USB pre-scaled + /// float path) or when the sample was not produced by the decode pipeline. + /// + int? RawValue { get; } + + /// + /// Gets the raw device timestamp (clock ticks) of the stream frame this sample was decoded + /// from, taken verbatim from the device, or null for samples not produced from a + /// stream frame. Unlike this value is not rollover-adjusted. + /// + uint? DeviceTimestamp { get; } } diff --git a/src/Daqifi.Core/Channel/SampleReceivedEventArgs.cs b/src/Daqifi.Core/Channel/SampleReceivedEventArgs.cs index 2eac26be..e58eefc7 100644 --- a/src/Daqifi.Core/Channel/SampleReceivedEventArgs.cs +++ b/src/Daqifi.Core/Channel/SampleReceivedEventArgs.cs @@ -5,6 +5,12 @@ namespace Daqifi.Core.Channel; /// public class SampleReceivedEventArgs : EventArgs { + /// + /// Gets the channel the sample was received on. This lets a single handler subscribed to + /// multiple channels attribute each sample without capturing the channel per subscription. + /// + public IChannel Channel { get; } + /// /// Gets the data sample that was received. /// @@ -13,9 +19,11 @@ public class SampleReceivedEventArgs : EventArgs /// /// Initializes a new instance of the class. /// + /// The channel the sample was received on. /// The data sample that was received. - public SampleReceivedEventArgs(IDataSample sample) + public SampleReceivedEventArgs(IChannel channel, IDataSample sample) { + Channel = channel ?? throw new ArgumentNullException(nameof(channel)); Sample = sample ?? throw new ArgumentNullException(nameof(sample)); } } diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index 41810844..df7abe27 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -48,6 +48,18 @@ public class DaqifiStreamingDevice : DaqifiDevice, IStreamingDevice, INetworkCon private bool _isLoggingToSdCard; private IReadOnlyList _sdCardFiles = Array.Empty(); + /// + /// Reconstructs host timestamps from the device's rolling 32-bit tick counter during a + /// streaming session. Scoped to this device instance, so a single fixed key suffices. + /// + private readonly ITimestampProcessor _timestampProcessor = new TimestampProcessor(); + + /// + /// The per-device key used with . The processor is not + /// shared across devices, so the key only needs to be stable within this instance. + /// + private const string StreamTimestampKey = "stream"; + /// /// Gets a value indicating whether the device is currently streaming data. /// @@ -161,6 +173,13 @@ public void StartStreaming() if (IsStreaming) return; + // Re-anchor per-session timestamp reconstruction: the first frame of this session + // anchors to the current host time, and subsequent frames advance by the device-tick + // delta. Apply the device-reported tick frequency (falls back to the 50 MHz default + // when unreported, e.g. older firmware). + _timestampProcessor.Reset(StreamTimestampKey); + _timestampProcessor.SetTimestampFrequency(StreamTimestampKey, TimestampFrequency); + IsStreaming = true; Send(ScpiMessageProducer.StartStreaming(StreamingFrequency)); } @@ -182,6 +201,172 @@ public void StopStreaming() Send(ScpiMessageProducer.StopStreaming); } + /// + /// Handles a streaming data frame: re-raises it for raw-frame consumers (via the base + /// implementation) and, while streaming, decodes it into per-channel samples that drive + /// . + /// + /// The streaming message from the device. + protected override void OnStreamMessageReceived(DaqifiOutMessage message) + { + // Preserve the raw-frame MessageReceived event so existing consumers that hand-demux + // the protobuf frame keep working unchanged. + base.OnStreamMessageReceived(message); + + // Only decode into channel samples while an app-driven stream is active. A stray frame + // that arrives outside a streaming session is still re-raised above but not decoded. + if (!IsStreaming) + { + return; + } + + try + { + DecodeStreamFrame(message); + } + catch (Exception) + { + // A single malformed frame must never tear down the stream or starve other + // consumers; decoding is best-effort per frame. + } + } + + /// + /// Decodes a streaming frame into per-channel samples: selects the active channels in + /// device order, chooses the correct value source (USB pre-scaled float vs. WiFi raw ADC + /// count scaled via calibration), unpacks digital bits, and pushes a sample to each channel. + /// + /// The streaming message to decode. + private void DecodeStreamFrame(DaqifiOutMessage message) + { + var hasFloat = message.AnalogInDataFloat.Count > 0; + var hasRawAnalog = message.AnalogInData.Count > 0; + var hasDigital = message.DigitalData.Length > 0; + + if (!hasFloat && !hasRawAnalog && !hasDigital) + { + return; + } + + // Reconstruct a host timestamp from the device tick counter (rollover-aware) and carry + // the raw device tick value through to each decoded sample. + var deviceTimestamp = message.MsgTimeStamp; + var hostTimestamp = _timestampProcessor + .ProcessTimestamp(StreamTimestampKey, deviceTimestamp) + .Timestamp; + + // Snapshot channels once: the consumer thread that repopulates channels is the same + // thread that runs this decode, so the structure is stable for the duration of the call. + var channels = SnapshotChannels(); + + if (hasFloat || hasRawAnalog) + { + DecodeAnalog(message, channels, hostTimestamp, deviceTimestamp, hasFloat); + } + + if (hasDigital) + { + DecodeDigital(message, channels, hostTimestamp, deviceTimestamp); + } + } + + /// + /// Maps a frame's analog values to the enabled analog channels, in ascending channel order. + /// USB firmware streams pre-scaled floats (used directly); WiFi firmware streams raw ADC + /// counts (scaled per channel via ). + /// + private static void DecodeAnalog( + DaqifiOutMessage message, + IReadOnlyList channels, + DateTime hostTimestamp, + uint deviceTimestamp, + bool hasFloat) + { + // The device streams one value per enabled analog channel, ordered by channel number, + // not by activation order — so re-derive that ordering here. + var activeAnalog = new List(); + foreach (var channel in channels) + { + if (channel.IsEnabled && channel is IAnalogChannel analog) + { + activeAnalog.Add(analog); + } + } + activeAnalog.Sort((a, b) => a.ChannelNumber.CompareTo(b.ChannelNumber)); + + var dataCount = hasFloat ? message.AnalogInDataFloat.Count : message.AnalogInData.Count; + var count = Math.Min(dataCount, activeAnalog.Count); + + for (var i = 0; i < count; i++) + { + var channel = activeAnalog[i]; + double scaled; + int? raw; + + if (hasFloat) + { + // USB firmware already scaled to volts; no raw ADC count is available. + scaled = message.AnalogInDataFloat[i]; + raw = null; + } + else + { + // WiFi firmware sent a raw ADC count; apply this channel's calibration. + var rawValue = message.AnalogInData[i]; + scaled = channel.GetScaledValue(rawValue); + raw = rawValue; + } + + channel.SetActiveSample(new DataSample(hostTimestamp, scaled, raw, deviceTimestamp)); + } + } + + /// + /// Unpacks a frame's digital byte(s) into per-channel high/low samples for the enabled + /// digital input channels, ordered by channel number. Bit position corresponds to the + /// channel's position within the active list (LSB first: position i -> byte + /// i / 8, bit i % 8), matching the device's dense packing of enabled channels + /// across however many payload bytes are present. Output channels occupy a bit position but + /// are not sampled. Decoding stops once the payload runs out of bits rather than wrapping + /// or forcing later channels low. + /// + private static void DecodeDigital( + DaqifiOutMessage message, + IReadOnlyList channels, + DateTime hostTimestamp, + uint deviceTimestamp) + { + var digitalData = message.DigitalData; + var bitCount = digitalData.Length * 8; + + var activeDigital = new List(); + foreach (var channel in channels) + { + if (channel.IsEnabled && channel.Type == ChannelType.Digital) + { + activeDigital.Add(channel); + } + } + activeDigital.Sort((a, b) => a.ChannelNumber.CompareTo(b.ChannelNumber)); + + for (var i = 0; i < activeDigital.Count && i < bitCount; i++) + { + var channel = activeDigital[i]; + + // Only input-direction channels carry a meaningful streamed reading; an output + // channel still occupies its bit position (so i advances), but is not sampled. + if (channel.Direction != ChannelDirection.Input) + { + continue; + } + + var bit = (digitalData[i / 8] & (1 << (i % 8))) != 0; + + channel.SetActiveSample( + new DataSample(hostTimestamp, bit ? 1.0 : 0.0, bit ? 1 : 0, deviceTimestamp)); + } + } + /// /// The maximum analog channel number that can be encoded in the ADC enable bitmask. /// The mask is a 32-bit value (1u << ChannelNumber), so channel numbers must be 0-31.