From 317f4755e501a09430c00ba60f8dcf495677b266 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Wed, 5 Aug 2026 13:24:37 -0600 Subject: [PATCH 1/3] refactor(device): extract the stream frame decode path into a collaborator (part of #344) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Moves the streaming hot path — the two frame guards, timestamp reconstruction, gap detection, and the analog/digital unpacking — out of DaqifiStreamingDevice into a new internal StreamFrameDecoder, the item #344 names as the next (and riskiest) extraction. The device keeps the public events and delegates. DaqifiStreamingDevice: 1,756 -> 1,368 lines. No public API or behavior change; zero edits to existing tests. The three events stay on the device and are reached through IDeviceOperationHost, because their sender has to remain the device a subscriber attached to, and the raw re-raise has to run through the device's base OnStreamMessageReceived so a subclass override still sees the frame. Co-Authored-By: Claude Opus 5 --- .../Internal/StreamFrameDecoderTests.cs | 471 ++++++++++++++++ .../Device/DaqifiStreamingDevice.cs | 447 ++------------- .../Device/Internal/IDeviceOperationHost.cs | 38 ++ .../Device/Internal/StreamFrameDecoder.cs | 508 ++++++++++++++++++ 4 files changed, 1055 insertions(+), 409 deletions(-) create mode 100644 src/Daqifi.Core.Tests/Device/Internal/StreamFrameDecoderTests.cs create mode 100644 src/Daqifi.Core/Device/Internal/StreamFrameDecoder.cs diff --git a/src/Daqifi.Core.Tests/Device/Internal/StreamFrameDecoderTests.cs b/src/Daqifi.Core.Tests/Device/Internal/StreamFrameDecoderTests.cs new file mode 100644 index 0000000..d643ea7 --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/Internal/StreamFrameDecoderTests.cs @@ -0,0 +1,471 @@ +using Daqifi.Core.Channel; +using Daqifi.Core.Communication.Messages; +using Daqifi.Core.Device; +using Daqifi.Core.Device.Internal; +using Daqifi.Core.Device.SdCard; +using Google.Protobuf; +using System; +using System.Collections.Generic; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace Daqifi.Core.Tests.Device.Internal; + +/// +/// Unit tests for , the streaming hot path extracted from +/// (#344). +/// +/// +/// +/// The end-to-end behavior of this pipeline is already pinned through the device by +/// DaqifiStreamingDeviceDecodeTests, and those tests are deliberately untouched — they are +/// the evidence that the extraction changed nothing. What these add is the part only a direct test +/// can see: the order and multiplicity of the calls back into the host. Through the device +/// those callbacks are invisible — a raw frame re-raised twice, or a discard counted after the +/// event rather than before, looks identical from the outside until a consumer trips over it. +/// +/// +public class StreamFrameDecoderTests +{ + private const uint TicksPerSecond = 1000; + + #region Raw-frame re-raise + + [Fact] + public void FrameArrivingWhileNotStreaming_IsReRaisedOnce_AndNotDecoded() + { + var host = new FakeHost { IsStreaming = false }; + var ai0 = host.AddAnalog(0, enabled: true); + var decoder = new StreamFrameDecoder(host); + + decoder.ProcessFrame(AnalogFrame(1000, 1.5f)); + + Assert.Equal(new[] { "raw" }, host.Calls); + Assert.Null(ai0.ActiveSample); + } + + [Fact] + public void DeliveredFrame_IsReRaisedExactlyOnce() + { + // The raw re-raise and the decode are two separate consumer paths off one frame. Re-raising + // twice would double-count for every consumer that hand-demuxes the protobuf frame. + var host = new FakeHost { IsStreaming = true }; + var ai0 = host.AddAnalog(0, enabled: true); + var decoder = new StreamFrameDecoder(host); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + decoder.ProcessFrame(AnalogFrame(1000, 2.5f)); + + Assert.Equal(new[] { "raw" }, host.Calls); + Assert.Equal(2.5, ai0.ActiveSample!.Value); + } + + [Fact] + public void SuppressedWarmupFrame_IsNotReRaised_ButItsDigitalPayloadStillDecodes() + { + // A short-analog leading frame is unusable to raw consumers (they read AnalogInData straight + // off it), so it is withheld — but withholding the whole frame would lose digital edges. + var host = new FakeHost { IsStreaming = true }; + var ai0 = host.AddAnalog(0, enabled: true); + var ai1 = host.AddAnalog(1, enabled: true); + var di0 = host.AddDigital(0, enabled: true); + var decoder = new StreamFrameDecoder(host); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 1000, DigitalData = ByteString.CopyFrom(0x01) }; + frame.AnalogInDataFloat.Add(9f); // one value for two enabled analog channels + decoder.ProcessFrame(frame); + + Assert.Equal(new[] { "discard" }, host.Calls); // no "raw" + Assert.Null(ai0.ActiveSample); + Assert.Null(ai1.ActiveSample); + Assert.Equal(1.0, di0.ActiveSample!.Value); + } + + [Fact] + public void StaleLeftoverFrame_IsNeitherReRaisedNorDecoded() + { + var host = new FakeHost { IsStreaming = false }; + var ai0 = host.AddAnalog(0, enabled: true); + var decoder = new StreamFrameDecoder(host); + + // Seed the gate's counter reference with the trailing frame the device emits after the stop + // command lands, then open a session whose first frame falls inside the leftover window. + decoder.ProcessFrame(AnalogFrame(10_000, 1f)); + host.Calls.Clear(); + host.IsStreaming = true; + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + decoder.ProcessFrame(AnalogFrame(10_500, 99f)); + + Assert.Equal(new[] { "discard" }, host.Calls); + Assert.Null(ai0.ActiveSample); + Assert.Equal(1, decoder.DiscardedStreamFrameCount); + } + + #endregion + + #region Discard counting + + [Fact] + public void DiscardIsCountedBeforeTheEventIsRaised() + { + // Documented guarantee on DiscardedStreamFrameCount: read inside a StreamFrameDiscarded + // handler, the count already includes the frame being reported. The device raises the event, + // so only a direct test can pin that the decoder increments first. + var host = new FakeHost { IsStreaming = true }; + host.AddAnalog(0, enabled: true); + host.AddAnalog(1, enabled: true); + StreamFrameDecoder? decoder = null; + long countSeenByHandler = -1; + host.OnDiscard = _ => countSeenByHandler = decoder!.DiscardedStreamFrameCount; + + decoder = new StreamFrameDecoder(host); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 1000 }; + frame.AnalogInDataFloat.Add(9f); + decoder.ProcessFrame(frame); + + Assert.Equal(1, countSeenByHandler); + } + + [Fact] + public void DiscardIsCountedEvenWhenTheHostRaisesNothing() + { + var host = new FakeHost { IsStreaming = true }; + host.AddAnalog(0, enabled: true); + host.AddAnalog(1, enabled: true); + var decoder = new StreamFrameDecoder(host); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 1000 }; + frame.AnalogInDataFloat.Add(9f); + decoder.ProcessFrame(frame); + + Assert.Equal(1, decoder.DiscardedStreamFrameCount); + } + + [Fact] + public void DiscardEventCarriesTheCountsTheDecisionWasMadeOn() + { + var host = new FakeHost { IsStreaming = true }; + host.AddAnalog(0, enabled: true); + host.AddAnalog(1, enabled: true); + host.AddAnalog(2, enabled: true); + StreamFrameDiscardedEventArgs? reported = null; + host.OnDiscard = e => reported = e; + + var decoder = new StreamFrameDecoder(host); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 4242 }; + frame.AnalogInDataFloat.Add(1f); + decoder.ProcessFrame(frame); + + Assert.NotNull(reported); + Assert.Equal(StreamFrameDiscardReason.PartialAnalogFrame, reported!.Reason); + Assert.Equal(4242u, reported.DeviceTimestamp); + Assert.Equal(1, reported.AnalogValueCount); + Assert.Equal(3, reported.EnabledAnalogChannelCount); + } + + #endregion + + #region Decode-failure isolation + + [Fact] + public void DecodeFailure_IsCountedAndReported_AndTheFrameStillReachedRawConsumers() + { + // Best-effort per frame (#378): a throwing decode must not tear down the stream, and the + // raw re-raise has already happened by then — the ordering is what keeps a bad decode from + // starving the other consumer path. + var host = new FakeHost { IsStreaming = true }; + var channel = host.AddAnalog(0, enabled: true); + var thrower = AttachThrowingSubscriber(channel); + var decoder = new StreamFrameDecoder(host); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + decoder.ProcessFrame(AnalogFrame(1000, 1f)); + + Assert.Equal(new[] { "raw", "decode-failure" }, host.Calls); + Assert.Equal(1, decoder.DecodeFailureCount); + Assert.Same(thrower.Error, host.LastDecodeFailure); + } + + [Fact] + public void DecodeFailure_DoesNotStopTheNextFrame() + { + var host = new FakeHost { IsStreaming = true }; + var channel = host.AddAnalog(0, enabled: true); + var thrower = AttachThrowingSubscriber(channel); + var decoder = new StreamFrameDecoder(host); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + decoder.ProcessFrame(AnalogFrame(1000, 1f)); + thrower.Error = null; + decoder.ProcessFrame(AnalogFrame(2000, 7f)); + + Assert.Equal(1, decoder.DecodeFailureCount); + Assert.Equal(7.0, channel.ActiveSample!.Value); + } + + #endregion + + #region Session reset + + [Fact] + public void BeginSession_ResetsBothCounters() + { + var host = new FakeHost { IsStreaming = true }; + var channel = host.AddAnalog(0, enabled: true); + host.AddAnalog(1, enabled: true); + var decoder = new StreamFrameDecoder(host); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + var shortFrame = new DaqifiOutMessage { MsgTimeStamp = 1000 }; + shortFrame.AnalogInDataFloat.Add(1f); + decoder.ProcessFrame(shortFrame); + + var thrower = AttachThrowingSubscriber(channel); + decoder.ProcessFrame(FullAnalogFrame(2000, 1f, 2f)); + thrower.Error = null; + + Assert.Equal(1, decoder.DiscardedStreamFrameCount); + Assert.Equal(1, decoder.DecodeFailureCount); + + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + Assert.Equal(0, decoder.DiscardedStreamFrameCount); + Assert.Equal(0, decoder.DecodeFailureCount); + } + + [Fact] + public void BeginSession_LeavesTheWarmupGuardDisarmedForADigitalOnlyStart() + { + // No analog channel is enabled at session start, so a short analog frame arriving later + // (analog enabled mid-stream) must not be mistaken for the firmware's warmup frame. + var host = new FakeHost { IsStreaming = true }; + var decoder = new StreamFrameDecoder(host); + host.AddDigital(0, enabled: true); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + var ai0 = host.AddAnalog(0, enabled: true); + host.AddAnalog(1, enabled: true); + + var frame = new DaqifiOutMessage { MsgTimeStamp = 1000 }; + frame.AnalogInDataFloat.Add(3f); + decoder.ProcessFrame(frame); + + Assert.Equal(new[] { "raw" }, host.Calls); + Assert.Equal(0, decoder.DiscardedStreamFrameCount); + Assert.Equal(3.0, ai0.ActiveSample!.Value); + } + + [Fact] + public void WarmupSuppression_IsBounded() + { + // A genuinely short stream must never be withheld forever: after the cap the frames flow. + var host = new FakeHost { IsStreaming = true }; + host.AddAnalog(0, enabled: true); + var ai1 = host.AddAnalog(1, enabled: true); + var decoder = new StreamFrameDecoder(host); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + for (uint ts = 1000; ts <= 10_000; ts += 1000) + { + var frame = new DaqifiOutMessage { MsgTimeStamp = ts }; + frame.AnalogInDataFloat.Add(1f); + decoder.ProcessFrame(frame); + } + + Assert.Equal(5, decoder.DiscardedStreamFrameCount); + Assert.Null(ai1.ActiveSample); // still only one value per frame + } + + #endregion + + #region Gap detection + + [Fact] + public void GapDetected_IsRaisedThroughTheHost_OnADeviceClockJump() + { + var host = new FakeHost { IsStreaming = true }; + host.AddAnalog(0, enabled: true); + var decoder = new StreamFrameDecoder(host); + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + for (uint ts = 1000; ts <= 11_000; ts += 1000) + { + decoder.ProcessFrame(AnalogFrame(ts, 1f)); + } + Assert.Null(host.LastGap); + + decoder.ProcessFrame(AnalogFrame(16_000, 1f)); + + Assert.NotNull(host.LastGap); + Assert.Equal(16_000u, host.LastGap!.DeviceTimestamp); + } + + [Fact] + public void BeginSession_ResetsTheGapDetector() + { + var host = new FakeHost { IsStreaming = true }; + host.AddAnalog(0, enabled: true); + var decoder = new StreamFrameDecoder(host); + + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + for (uint ts = 1000; ts <= 11_000; ts += 1000) + { + decoder.ProcessFrame(AnalogFrame(ts, 1f)); + } + + // A slower session: were the EMA carried over, the first 3000-tick delta would false-trip. + decoder.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + host.LastGap = null; + decoder.ProcessFrame(AnalogFrame(100_000, 1f)); + decoder.ProcessFrame(AnalogFrame(103_000, 1f)); + decoder.ProcessFrame(AnalogFrame(106_000, 1f)); + + Assert.Null(host.LastGap); + } + + #endregion + + #region Guards + + [Fact] + public void Constructor_RejectsANullHost() + { + Assert.Throws(() => new StreamFrameDecoder(null!)); + } + + #endregion + + #region Helpers + + private static ThrowSwitch AttachThrowingSubscriber(IChannel channel) + { + var thrower = new ThrowSwitch(); + channel.SampleReceived += (_, _) => + { + if (thrower.Error != null) + { + throw thrower.Error; + } + }; + return thrower; + } + + private static DaqifiOutMessage AnalogFrame(uint timestamp, float value) + { + var frame = new DaqifiOutMessage { MsgTimeStamp = timestamp }; + frame.AnalogInDataFloat.Add(value); + return frame; + } + + private static DaqifiOutMessage FullAnalogFrame(uint timestamp, params float[] values) + { + var frame = new DaqifiOutMessage { MsgTimeStamp = timestamp }; + foreach (var value in values) + { + frame.AnalogInDataFloat.Add(value); + } + return frame; + } + + /// + /// An that records the decoder's calls back into the device + /// in order. Only the members the decoder uses are implemented; the rest throw, so a future + /// change that makes the decoder reach for device I/O fails loudly instead of quietly. + /// + private sealed class FakeHost : IDeviceOperationHost + { + private readonly List _channels = new(); + + public List Calls { get; } = new(); + + public Action? OnDiscard { get; set; } + + public TimestampGapEventArgs? LastGap { get; set; } + + public Exception? LastDecodeFailure { get; private set; } + + public bool IsStreaming { get; set; } + + public AnalogChannel AddAnalog(int number, bool enabled) + { + var channel = new AnalogChannel(number) { IsEnabled = enabled }; + _channels.Add(channel); + return channel; + } + + public IChannel AddDigital(int number, bool enabled) + { + var channel = new DigitalChannel(number) { IsEnabled = enabled, Direction = ChannelDirection.Input }; + _channels.Add(channel); + return channel; + } + + public IReadOnlyList SnapshotChannels() => _channels.ToArray(); + + public void RaiseStreamFrameDiscarded(StreamFrameDiscardedEventArgs e) + { + Calls.Add("discard"); + OnDiscard?.Invoke(e); + } + + public void RaiseGapDetected(TimestampGapEventArgs e) + { + Calls.Add("gap"); + LastGap = e; + } + + public void RaiseRawStreamFrame(DaqifiOutMessage message) => Calls.Add("raw"); + + public void RaiseStreamDecodeFailure(Exception error) + { + Calls.Add("decode-failure"); + LastDecodeFailure = error; + } + + // Not part of the decode path. + public bool IsConnected => throw new NotSupportedException(); + public bool IsUsbConnection => throw new NotSupportedException(); + public int StreamingFrequency => throw new NotSupportedException(); + public TimeSpan SdCardDownloadTimeout => throw new NotSupportedException(); + public TimeSpan SdCardTransferIdleTimeout => throw new NotSupportedException(); + public void StopStreaming() => throw new NotSupportedException(); + public void Send(IOutboundMessage message) => throw new NotSupportedException(); + public void WithChannelsLock(Action action) => throw new NotSupportedException(); + public Task> ExecuteTextCommandAsync( + Action setupAction, + int responseTimeoutMs = 1000, + int completionTimeoutMs = 250, + CancellationToken cancellationToken = default, + Func? prepareAsync = null, + Func? finalizeAsync = null) => throw new NotSupportedException(); + public Task ExecuteRawCaptureAsync( + Func rawAction, + CancellationToken cancellationToken = default) => throw new NotSupportedException(); + public void EnsureSupported(DeviceFeature feature) => throw new NotSupportedException(); + public FeatureNotSupportedException CreateFeatureNotSupportedException(DeviceFeature feature) + => throw new NotSupportedException(); + public void RaiseLowSdSpaceWarning(LowSdSpaceWarningEventArgs e) => throw new NotSupportedException(); + } + + /// + /// A switch for a subscriber that throws while + /// is set. A throwing subscriber is the realistic shape of a decode that + /// fails — it propagates out of the per-channel push and into the frame's catch — and is the + /// same lever DeviceErrorSurfaceTests uses. + /// + private sealed class ThrowSwitch + { + public Exception? Error { get; set; } = new InvalidOperationException("decode consumer is broken"); + } + + #endregion +} diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index 3473d35..5766b3d 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -44,62 +44,6 @@ public class DaqifiStreamingDevice : DaqifiDevice, IStreamingDevice, INetworkCon /// private const int UsbStreamInterfaceRetryDelayMs = 150; - /// - /// 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"; - - /// - /// Detects dropped samples from the device-clock delta between frames. Reset at the start of - /// every streaming session alongside . Drives . - /// - private readonly TimestampGapDetector _gapDetector = new(); - - /// - /// Keeps frames the device latched from the previous streaming session out of this one - /// (daqifi-nyquist-firmware #533). Re-armed by . - /// - private readonly StreamFrameGate _frameGate = new(); - - /// - /// The maximum number of leading short-analog frames suppressed at stream start - /// (see ). Bounds the warmup-frame guard so a - /// genuinely short stream can never be withheld indefinitely. - /// - private const int MaxSuppressedWarmupFrames = 5; - - /// - /// True from the start of a streaming session that begins with analog channels enabled, - /// until the first analog-bearing frame carrying the full enabled-channel complement has - /// been decoded (disarmed for a digital-only start). Guards the malformed warmup frame - /// the firmware emits at stream start (issue #351): its fast streaming encoder can emit a - /// leading frame with fewer analog values than the enabled channel mask, which would - /// otherwise reach every consumer as a partial (silently corrupting - /// first-value baselining, gap detection, and export). For such leading short frames only - /// the malformed analog decode is skipped — a combined frame's digital payload is still - /// decoded and the raw frame is still re-raised — until the first full frame arrives, - /// bounded by . - /// - private bool _awaitingFirstFullAnalogFrame; - - /// - /// Count of leading short-analog frames suppressed in the current session; capped by - /// . - /// - private int _suppressedWarmupFrameCount; - - /// - /// Backing counter for . - /// - private long _discardedStreamFrameCount; - /// /// Raised when a stream frame was withheld from consumers because the device should not have /// sent it — a malformed leading frame, or one latched from the previous session. @@ -137,12 +81,7 @@ public class DaqifiStreamingDevice : DaqifiDevice, IStreamingDevice, INetworkCon /// reported. /// /// - public long DiscardedStreamFrameCount => Interlocked.Read(ref _discardedStreamFrameCount); - - /// - /// Backing counter for . - /// - private long _decodeFailureCount; + public long DiscardedStreamFrameCount => _frameDecoder.DiscardedStreamFrameCount; /// /// Gets the number of streaming frames whose decode threw and was discarded since the @@ -161,7 +100,7 @@ public class DaqifiStreamingDevice : DaqifiDevice, IStreamingDevice, INetworkCon /// stream leaves it at zero. /// /// - public long DecodeFailureCount => Interlocked.Read(ref _decodeFailureCount); + public long DecodeFailureCount => _frameDecoder.DecodeFailureCount; /// /// Gets a value indicating whether the device is currently streaming data. @@ -259,6 +198,7 @@ private void InitializeStreamingDevice() // Built here rather than in field initializers because each needs `this` as its host, // which a field initializer cannot reference. Every constructor routes through this // method, so they are always in place before the device is handed to a caller. + _frameDecoder = new StreamFrameDecoder(this); _channelControl = new ChannelControlOperations(this); _networkOperations = new NetworkConfigurationOperations(this); _sdCardOperations = new SdCardOperations(this); @@ -398,31 +338,8 @@ public void StartStreaming() /// previous session's anchor and re-use its gap detector, producing samples stamped with /// times that never happened — silently. /// - private void BeginStreamingSession() - { - // 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); - _gapDetector.Reset(); - - // Arm the warmup-frame guard only when analog channels are enabled at stream start — - // the reproduced failure mode (issue #351) is the firmware's leading partial-analog - // frame at the start of an *analog* stream. A digital-only start needs no guard; leaving - // it disarmed there also avoids suppressing short analog frames that could arrive far - // from session start if analog channels are enabled mid-stream (a scenario with no - // observed warmup frame). - _awaitingFirstFullAnalogFrame = CountEnabledAnalogChannels(SnapshotChannels()) > 0; - _suppressedWarmupFrameCount = 0; - Interlocked.Exchange(ref _decodeFailureCount, 0); - - // Re-arm the cross-session leftover guard against the counter value this session - // inherits, and at this session's rate — the window is measured in sample periods. - _frameGate.BeginSession(TimestampFrequency, StreamingFrequency); - Interlocked.Exchange(ref _discardedStreamFrameCount, 0); - } + private void BeginStreamingSession() => + _frameDecoder.BeginSession(TimestampFrequency, StreamingFrequency); /// /// Stops streaming data from the device. @@ -867,10 +784,10 @@ void OnSample(object? sender, SampleReceivedEventArgs e) => } /// - /// Handles a streaming data frame: screens out the frames the device should not have sent, - /// then re-raises the frame for raw-frame consumers (via the base implementation) and, while - /// streaming, decodes it into per-channel samples that drive - /// . + /// Handles a streaming data frame by handing it to , which + /// screens out the frames the device should not have sent, then re-raises what survives for + /// raw-frame consumers (via the base implementation) and, while streaming, decodes it into + /// per-channel samples that drive . /// /// /// Screening covers both consumer paths, which is the whole point of issue #425. The @@ -879,167 +796,26 @@ void OnSample(object? sender, SampleReceivedEventArgs e) => /// the frame verbatim — and that is the path most callers actually use, including the /// example CLI, whose offline export inferred a channel count of one from it and truncated /// every sample that followed. Whatever is unfit for the decoded path is unfit for the raw - /// one; both are now gated together, and every drop is reported through + /// one; both are gated together, and every drop is reported through /// . /// /// The streaming message from the device. - protected override void OnStreamMessageReceived(DaqifiOutMessage message) - { - if (IsStreaming && _frameGate.IsValidating && _frameGate.IsLeftoverFromPreviousSession(message)) - { - RaiseStreamFrameDiscarded( - StreamFrameDiscardReason.StaleLeftoverFrame, - message, - CountAnalogValues(message), - CountEnabledAnalogChannels(SnapshotChannels())); - return; - } - - _frameGate.TrackFrame(message.MsgTimeStamp); - - // 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 but not decoded. - if (!IsStreaming) - { - base.OnStreamMessageReceived(message); - return; - } - - EmitStreamFrame(message); - } + protected override void OnStreamMessageReceived(DaqifiOutMessage message) => + _frameDecoder.ProcessFrame(message); /// - /// Delivers a frame that has cleared : re-raises it for raw-frame - /// consumers and decodes it into per-channel samples. + /// Raises for a frame the decoder withheld. Subscriber + /// exceptions are isolated, mirroring . /// /// - /// The firmware's malformed leading frame (issue #351) is caught here rather than by the - /// gate, because it needs the enabled-channel count and because what it costs is narrower: - /// only the analog payload is unusable. Such a frame is withheld from raw consumers — they - /// read AnalogInData straight off it, so there is no way to hand it over safely — but - /// its digital payload is still decoded and its timestamp still anchors the session clock, - /// exactly as before. + /// Kept on the device rather than moved into so the event's + /// sender stays the device a subscriber attached to. The decoder counts the discard + /// before calling this, so read inside a handler + /// still already includes the frame being reported. /// - /// The frame to deliver. - private void EmitStreamFrame(DaqifiOutMessage message) - { - var suppressAnalog = false; - var analogValueCount = 0; - var enabledAnalogChannelCount = 0; - - if (_awaitingFirstFullAnalogFrame) - { - suppressAnalog = ShouldSuppressPartialAnalog( - message, out analogValueCount, out enabledAnalogChannelCount); - } - - if (suppressAnalog) - { - // The counts reported here are the very ones the suppression decision was made on, - // not a fresh reading: channel enablement can change from another thread, and a - // discard whose reported numbers disagree with the reason it was discarded would - // make the telemetry harder to trust than no telemetry at all. - RaiseStreamFrameDiscarded( - StreamFrameDiscardReason.PartialAnalogFrame, - message, - analogValueCount, - enabledAnalogChannelCount); - } - else - { - // Preserve the raw-frame MessageReceived event so existing consumers that hand-demux - // the protobuf frame keep working unchanged. - base.OnStreamMessageReceived(message); - } - - try - { - DecodeStreamFrame(message, suppressAnalog); - } - catch (Exception ex) - { - // A single malformed frame must never tear down the stream or starve other - // consumers; decoding is best-effort per frame. That isolation stays exactly as it - // was — the frame is dropped and the loop continues — but it is no longer silent - // (issue #378): a decode that throws on every frame yields no samples, which used - // to be indistinguishable from a device sending nothing at all. Both the counter - // and the (throttled) event are observation only; neither changes what happens to - // this frame or the next one. - Interlocked.Increment(ref _decodeFailureCount); - RaiseDeviceError(DeviceErrorSource.StreamDecode, ex); - } - } - - /// - /// Decides whether a leading frame's analog payload is the firmware's malformed warmup frame - /// (issue #351): fewer analog values than there are enabled analog channels. Disarms the - /// guard on the first analog-bearing frame that is not short, or once - /// have been suppressed. - /// - /// The frame about to be delivered. - /// The number of analog values the frame carried. - /// - /// The number of enabled analog channels the decision was made against. Handed back so the - /// discard event reports the same numbers the decision used rather than re-reading channel - /// state that another thread may have changed in between. - /// - /// true when the frame's analog values must be withheld. - private bool ShouldSuppressPartialAnalog( - DaqifiOutMessage message, - out int analogValueCount, - out int enabledAnalogChannelCount) + /// The discard to report. + private void RaiseStreamFrameDiscarded(StreamFrameDiscardedEventArgs e) { - analogValueCount = CountAnalogValues(message); - enabledAnalogChannelCount = 0; - - // A frame with no analog payload says nothing about the warmup frame either way, so the - // guard stays armed for the first analog-bearing frame. - if (analogValueCount == 0) - { - return false; - } - - enabledAnalogChannelCount = CountEnabledAnalogChannels(SnapshotChannels()); - if (enabledAnalogChannelCount > 0 - && analogValueCount < enabledAnalogChannelCount - && _suppressedWarmupFrameCount < MaxSuppressedWarmupFrames) - { - _suppressedWarmupFrameCount++; - return true; - } - - _awaitingFirstFullAnalogFrame = false; - return false; - } - - /// - /// The number of analog values a frame carries, from whichever payload the transport used — - /// USB streams pre-scaled floats, WiFi streams raw ADC counts. - /// - private static int CountAnalogValues(DaqifiOutMessage message) => - message.AnalogInDataFloat.Count > 0 - ? message.AnalogInDataFloat.Count - : message.AnalogInData.Count; - - /// - /// Raises for a frame that was withheld, and counts it. - /// Subscriber exceptions are isolated, mirroring . - /// - /// Why the frame was withheld. - /// The frame that was withheld. - /// The number of analog values the frame carried. - /// - /// The number of enabled analog channels to report. Passed in rather than re-derived so it - /// is the same reading the discard decision was made against. - /// - private void RaiseStreamFrameDiscarded( - StreamFrameDiscardReason reason, - DaqifiOutMessage frame, - int analogValueCount, - int enabledAnalogChannelCount) - { - Interlocked.Increment(ref _discardedStreamFrameCount); - var handler = StreamFrameDiscarded; if (handler == null) { @@ -1048,8 +824,7 @@ private void RaiseStreamFrameDiscarded( try { - handler(this, new StreamFrameDiscardedEventArgs( - reason, frame.MsgTimeStamp, analogValueCount, enabledAnalogChannelCount)); + handler(this, e); } catch (Exception ex) { @@ -1082,60 +857,6 @@ private static void SafeTrace(string message) } } - /// - /// 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. - /// - /// When true, the frame's analog payload is the firmware's malformed warmup frame - /// (issue #351) and is skipped. Only the analog values are withheld — a combined frame's - /// digital payload is still decoded, and the frame's (normal one-period) timestamp still - /// anchors the session clock, so digital state and edges are not lost. - /// - private void DecodeStreamFrame(DaqifiOutMessage message, bool suppressAnalog) - { - var hasFloat = message.AnalogInDataFloat.Count > 0; - var hasRawAnalog = message.AnalogInData.Count > 0; - var hasDigital = message.DigitalData.Length > 0; - - if (!hasFloat && !hasRawAnalog && !hasDigital) - { - return; - } - - // 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(); - - // 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 timestampResult = _timestampProcessor.ProcessTimestamp(StreamTimestampKey, deviceTimestamp); - var hostTimestamp = timestampResult.Timestamp; - - // Flag dropped samples from the device-clock delta (immune to host arrival jitter). - // Isolate subscriber exceptions (see RaiseGapDetected) so a throwing GapDetected handler - // cannot skip the per-channel decode below — which the caller's broad catch would then - // silently drop. - if (_gapDetector.IsGap(timestampResult.SecondsBetweenMessages)) - { - RaiseGapDetected(new TimestampGapEventArgs( - hostTimestamp, timestampResult.SecondsBetweenMessages, deviceTimestamp)); - } - - if ((hasFloat || hasRawAnalog) && !suppressAnalog) - { - DecodeAnalog(message, channels, hostTimestamp, deviceTimestamp, hasFloat); - } - - if (hasDigital) - { - DecodeDigital(message, channels, hostTimestamp, deviceTimestamp); - } - } - /// /// Raises , isolating the decode pipeline from a subscriber /// exception so a throwing handler cannot skip this frame's per-channel decode (which the @@ -1160,115 +881,6 @@ private void RaiseGapDetected(TimestampGapEventArgs args) } } - /// - /// 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 int CountEnabledAnalogChannels(IReadOnlyList channels) - { - var count = 0; - foreach (var channel in channels) - { - if (channel.IsEnabled && channel is IAnalogChannel) - { - count++; - } - } - return count; - } - - 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. The firmware streams the whole DIO port as a raw pin-state - /// snapshot (the wire-level DIO enable is global, not per pin), so a channel's bit - /// position is its channel number — bit n lives at byte n / 8, bit - /// n % 8 (LSB first) — independent of which channels the client has enabled. - /// Output-direction channels are not sampled (their state is client-driven via - /// ). Channels whose number lies beyond the payload get no - /// sample rather than a bogus "low" reading. - /// - private static void DecodeDigital( - DaqifiOutMessage message, - IReadOnlyList channels, - DateTime hostTimestamp, - uint deviceTimestamp) - { - var digitalData = message.DigitalData; - var bitCount = digitalData.Length * 8; - - foreach (var channel in channels) - { - if (!channel.IsEnabled || channel.Type != ChannelType.Digital) - { - continue; - } - - // Only input-direction channels carry a meaningful streamed reading. - if (channel.Direction != ChannelDirection.Input) - { - continue; - } - - var bitIndex = channel.ChannelNumber; - if (bitIndex >= bitCount) - { - continue; - } - - var bit = (digitalData[bitIndex / 8] & (1 << (bitIndex % 8))) != 0; - - channel.SetActiveSample( - new DataSample(hostTimestamp, bit ? 1.0 : 0.0, bit ? 1 : 0, deviceTimestamp)); - } - } - /// public void EnableChannel(IChannel channel) => _channelControl.EnableChannel(channel); @@ -1523,6 +1135,9 @@ public void LoadVoltagePrecision() // virtual members and any subclass override of them. // ----------------------------------------------------------------- + /// The streaming hot path: frame screening, timestamps, gaps, per-channel decode. + private StreamFrameDecoder _frameDecoder = null!; + /// Channel enable/disable, DIO, PWM and analog output (). private ChannelControlOperations _channelControl = null!; @@ -1751,6 +1366,20 @@ FeatureNotSupportedException IDeviceOperationHost.CreateFeatureNotSupportedExcep void IDeviceOperationHost.RaiseLowSdSpaceWarning(LowSdSpaceWarningEventArgs e) => OnLowSdSpaceWarning(e); + void IDeviceOperationHost.RaiseStreamFrameDiscarded(StreamFrameDiscardedEventArgs e) + => RaiseStreamFrameDiscarded(e); + + void IDeviceOperationHost.RaiseGapDetected(TimestampGapEventArgs e) => RaiseGapDetected(e); + + // Deliberately base.OnStreamMessageReceived and not the override: the override is what hands + // the frame to the decoder in the first place, so calling it here would recurse. This is the + // same base call the decode block made before it moved out. + void IDeviceOperationHost.RaiseRawStreamFrame(DaqifiOutMessage message) + => base.OnStreamMessageReceived(message); + + void IDeviceOperationHost.RaiseStreamDecodeFailure(Exception error) + => RaiseDeviceError(DeviceErrorSource.StreamDecode, error); + #endregion } } diff --git a/src/Daqifi.Core/Device/Internal/IDeviceOperationHost.cs b/src/Daqifi.Core/Device/Internal/IDeviceOperationHost.cs index c5ce767..114891f 100644 --- a/src/Daqifi.Core/Device/Internal/IDeviceOperationHost.cs +++ b/src/Daqifi.Core/Device/Internal/IDeviceOperationHost.cs @@ -110,5 +110,43 @@ Task ExecuteRawCaptureAsync( /// a silent, compile-clean behavior change. /// void RaiseLowSdSpaceWarning(LowSdSpaceWarningEventArgs e); + + /// + /// Raises the device's event, + /// isolating a throwing subscriber. + /// + /// + /// Same reasoning as : the event is part of the device's + /// public surface, so its sender must stay the device. + /// + void RaiseStreamFrameDiscarded(StreamFrameDiscardedEventArgs e); + + /// + /// Raises the device's event, isolating a + /// throwing subscriber so it cannot skip the rest of the frame's decode. + /// + void RaiseGapDetected(TimestampGapEventArgs e); + + /// + /// Re-raises a streaming frame for raw-frame consumers, through the device's base + /// . + /// + /// + /// This is the MessageReceived path most callers actually use. It has to run through + /// the device rather than from a collaborator so the base implementation — and any subclass + /// sitting between it and the streaming device — still sees the frame. + /// + void RaiseRawStreamFrame(DaqifiOutMessage message); + + /// + /// Reports a frame whose decode threw through the device's + /// surface as + /// (issue #378). + /// + /// + /// Observation only — the frame is dropped either way, and the device's throttling decides + /// whether the event is actually raised. + /// + void RaiseStreamDecodeFailure(Exception error); } } diff --git a/src/Daqifi.Core/Device/Internal/StreamFrameDecoder.cs b/src/Daqifi.Core/Device/Internal/StreamFrameDecoder.cs new file mode 100644 index 0000000..625aa7d --- /dev/null +++ b/src/Daqifi.Core/Device/Internal/StreamFrameDecoder.cs @@ -0,0 +1,508 @@ +using Daqifi.Core.Channel; +using Daqifi.Core.Communication.Messages; +using System; +using System.Collections.Generic; +using System.Threading; + +#nullable enable + +namespace Daqifi.Core.Device.Internal +{ + /// + /// The streaming hot path extracted from (#344): everything + /// that happens to a frame between the transport handing it over and the per-channel + /// samples it becomes — the two frame guards, timestamp + /// reconstruction, gap detection, and the analog/digital unpacking. + /// + /// + /// + /// The session-scoped state this owns — the timestamp anchor, the gap detector, the + /// cross-session leftover gate, the warmup-frame guard, and the two per-session counters — is + /// exactly the set that resets together. Keeping it in one object is + /// the point: the failure mode this guards against is a partial reset, where frames are decoded + /// against the previous session's anchor and stamped with times that never happened. + /// + /// + /// The and + /// events stay on the device, raised through + /// rather than owned here, because their sender has to + /// remain the device a subscriber attached to. So does re-raising the raw frame, which must go + /// through the device's base.OnStreamMessageReceived so a subclass that overrides it + /// still intercepts it. + /// + /// + /// Not thread-safe by design, matching the code it was moved from: frames arrive on the single + /// message-consumer thread, which is also the thread that repopulates channels, so the + /// unsynchronized session fields are only ever touched from there. The two counters are + /// interlocked because they are read from arbitrary threads through the device's public + /// properties. + /// + /// + internal sealed class StreamFrameDecoder + { + /// + /// The maximum number of leading short-analog frames suppressed at stream start + /// (see ). Bounds the warmup-frame guard so a + /// genuinely short stream can never be withheld indefinitely. + /// + private const int MaxSuppressedWarmupFrames = 5; + + /// + /// 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"; + + private readonly IDeviceOperationHost _host; + + /// + /// 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(); + + /// + /// Detects dropped samples from the device-clock delta between frames. Reset at the start of + /// every streaming session alongside . Drives + /// . + /// + private readonly TimestampGapDetector _gapDetector = new(); + + /// + /// Keeps frames the device latched from the previous streaming session out of this one + /// (daqifi-nyquist-firmware #533). Re-armed by . + /// + private readonly StreamFrameGate _frameGate = new(); + + /// + /// True from the start of a streaming session that begins with analog channels enabled, + /// until the first analog-bearing frame carrying the full enabled-channel complement has + /// been decoded (disarmed for a digital-only start). Guards the malformed warmup frame + /// the firmware emits at stream start (issue #351): its fast streaming encoder can emit a + /// leading frame with fewer analog values than the enabled channel mask, which would + /// otherwise reach every consumer as a partial (silently corrupting + /// first-value baselining, gap detection, and export). For such leading short frames only + /// the malformed analog decode is skipped — a combined frame's digital payload is still + /// decoded and the raw frame is still re-raised — until the first full frame arrives, + /// bounded by . + /// + private bool _awaitingFirstFullAnalogFrame; + + /// + /// Count of leading short-analog frames suppressed in the current session; capped by + /// . + /// + private int _suppressedWarmupFrameCount; + + /// + /// Backing counter for . + /// + private long _discardedStreamFrameCount; + + /// + /// Backing counter for . + /// + private long _decodeFailureCount; + + internal StreamFrameDecoder(IDeviceOperationHost host) + { + _host = host ?? throw new ArgumentNullException(nameof(host)); + } + + /// + internal long DiscardedStreamFrameCount => Interlocked.Read(ref _discardedStreamFrameCount); + + /// + internal long DecodeFailureCount => Interlocked.Read(ref _decodeFailureCount); + + /// + /// Resets everything that is scoped to one streaming session, so the frames that follow are + /// decoded against this session rather than the last one. + /// + /// + /// Called for and for a start-streaming + /// command sent directly through Send. It is one method precisely so the two cannot + /// drift: a raw-started stream that skipped this would reconstruct timestamps from the + /// previous session's anchor and re-use its gap detector, producing samples stamped with + /// times that never happened — silently. + /// + /// + /// The device-reported tick frequency for this session (the device's 50 MHz fallback when + /// unreported). + /// + /// The rate this session is starting at. + internal void BeginSession(uint timestampFrequency, int streamingFrequencyHz) + { + // 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); + _gapDetector.Reset(); + + // Arm the warmup-frame guard only when analog channels are enabled at stream start — + // the reproduced failure mode (issue #351) is the firmware's leading partial-analog + // frame at the start of an *analog* stream. A digital-only start needs no guard; leaving + // it disarmed there also avoids suppressing short analog frames that could arrive far + // from session start if analog channels are enabled mid-stream (a scenario with no + // observed warmup frame). + _awaitingFirstFullAnalogFrame = CountEnabledAnalogChannels(_host.SnapshotChannels()) > 0; + _suppressedWarmupFrameCount = 0; + Interlocked.Exchange(ref _decodeFailureCount, 0); + + // Re-arm the cross-session leftover guard against the counter value this session + // inherits, and at this session's rate — the window is measured in sample periods. + _frameGate.BeginSession(timestampFrequency, streamingFrequencyHz); + Interlocked.Exchange(ref _discardedStreamFrameCount, 0); + } + + /// + /// Handles a streaming data frame: screens out the frames the device should not have sent, + /// then re-raises the frame for raw-frame consumers and, while streaming, decodes it into + /// per-channel samples that drive . + /// + /// + /// Screening covers both consumer paths, which is the whole point of issue #425. The + /// per-channel decode has guarded against the firmware's malformed leading frame since + /// issue #351, but the raw event was still handed + /// the frame verbatim — and that is the path most callers actually use, including the + /// example CLI, whose offline export inferred a channel count of one from it and truncated + /// every sample that followed. Whatever is unfit for the decoded path is unfit for the raw + /// one; both are now gated together, and every drop is reported through + /// . + /// + /// The streaming message from the device. + internal void ProcessFrame(DaqifiOutMessage message) + { + if (_host.IsStreaming && _frameGate.IsValidating && _frameGate.IsLeftoverFromPreviousSession(message)) + { + RaiseStreamFrameDiscarded( + StreamFrameDiscardReason.StaleLeftoverFrame, + message, + CountAnalogValues(message), + CountEnabledAnalogChannels(_host.SnapshotChannels())); + return; + } + + _frameGate.TrackFrame(message.MsgTimeStamp); + + // 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 but not decoded. + if (!_host.IsStreaming) + { + _host.RaiseRawStreamFrame(message); + return; + } + + EmitStreamFrame(message); + } + + /// + /// Delivers a frame that has cleared : re-raises it for raw-frame + /// consumers and decodes it into per-channel samples. + /// + /// + /// The firmware's malformed leading frame (issue #351) is caught here rather than by the + /// gate, because it needs the enabled-channel count and because what it costs is narrower: + /// only the analog payload is unusable. Such a frame is withheld from raw consumers — they + /// read AnalogInData straight off it, so there is no way to hand it over safely — but + /// its digital payload is still decoded and its timestamp still anchors the session clock, + /// exactly as before. + /// + /// The frame to deliver. + private void EmitStreamFrame(DaqifiOutMessage message) + { + var suppressAnalog = false; + var analogValueCount = 0; + var enabledAnalogChannelCount = 0; + + if (_awaitingFirstFullAnalogFrame) + { + suppressAnalog = ShouldSuppressPartialAnalog( + message, out analogValueCount, out enabledAnalogChannelCount); + } + + if (suppressAnalog) + { + // The counts reported here are the very ones the suppression decision was made on, + // not a fresh reading: channel enablement can change from another thread, and a + // discard whose reported numbers disagree with the reason it was discarded would + // make the telemetry harder to trust than no telemetry at all. + RaiseStreamFrameDiscarded( + StreamFrameDiscardReason.PartialAnalogFrame, + message, + analogValueCount, + enabledAnalogChannelCount); + } + else + { + // Preserve the raw-frame MessageReceived event so existing consumers that hand-demux + // the protobuf frame keep working unchanged. + _host.RaiseRawStreamFrame(message); + } + + try + { + DecodeStreamFrame(message, suppressAnalog); + } + catch (Exception ex) + { + // A single malformed frame must never tear down the stream or starve other + // consumers; decoding is best-effort per frame. That isolation stays exactly as it + // was — the frame is dropped and the loop continues — but it is no longer silent + // (issue #378): a decode that throws on every frame yields no samples, which used + // to be indistinguishable from a device sending nothing at all. Both the counter + // and the (throttled) event are observation only; neither changes what happens to + // this frame or the next one. + Interlocked.Increment(ref _decodeFailureCount); + _host.RaiseStreamDecodeFailure(ex); + } + } + + /// + /// Decides whether a leading frame's analog payload is the firmware's malformed warmup frame + /// (issue #351): fewer analog values than there are enabled analog channels. Disarms the + /// guard on the first analog-bearing frame that is not short, or once + /// have been suppressed. + /// + /// The frame about to be delivered. + /// The number of analog values the frame carried. + /// + /// The number of enabled analog channels the decision was made against. Handed back so the + /// discard event reports the same numbers the decision used rather than re-reading channel + /// state that another thread may have changed in between. + /// + /// true when the frame's analog values must be withheld. + private bool ShouldSuppressPartialAnalog( + DaqifiOutMessage message, + out int analogValueCount, + out int enabledAnalogChannelCount) + { + analogValueCount = CountAnalogValues(message); + enabledAnalogChannelCount = 0; + + // A frame with no analog payload says nothing about the warmup frame either way, so the + // guard stays armed for the first analog-bearing frame. + if (analogValueCount == 0) + { + return false; + } + + enabledAnalogChannelCount = CountEnabledAnalogChannels(_host.SnapshotChannels()); + if (enabledAnalogChannelCount > 0 + && analogValueCount < enabledAnalogChannelCount + && _suppressedWarmupFrameCount < MaxSuppressedWarmupFrames) + { + _suppressedWarmupFrameCount++; + return true; + } + + _awaitingFirstFullAnalogFrame = false; + return false; + } + + /// + /// The number of analog values a frame carries, from whichever payload the transport used — + /// USB streams pre-scaled floats, WiFi streams raw ADC counts. + /// + private static int CountAnalogValues(DaqifiOutMessage message) => + message.AnalogInDataFloat.Count > 0 + ? message.AnalogInDataFloat.Count + : message.AnalogInData.Count; + + /// + /// Counts a withheld frame and asks the device to raise + /// for it. + /// + /// + /// The increment happens before the event is raised, so a handler that reads + /// sees a total that already + /// includes the frame it is being told about — a documented guarantee. + /// + /// Why the frame was withheld. + /// The frame that was withheld. + /// The number of analog values the frame carried. + /// + /// The number of enabled analog channels to report. Passed in rather than re-derived so it + /// is the same reading the discard decision was made against. + /// + private void RaiseStreamFrameDiscarded( + StreamFrameDiscardReason reason, + DaqifiOutMessage frame, + int analogValueCount, + int enabledAnalogChannelCount) + { + Interlocked.Increment(ref _discardedStreamFrameCount); + + _host.RaiseStreamFrameDiscarded(new StreamFrameDiscardedEventArgs( + reason, frame.MsgTimeStamp, analogValueCount, enabledAnalogChannelCount)); + } + + /// + /// 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. + /// + /// When true, the frame's analog payload is the firmware's malformed warmup frame + /// (issue #351) and is skipped. Only the analog values are withheld — a combined frame's + /// digital payload is still decoded, and the frame's (normal one-period) timestamp still + /// anchors the session clock, so digital state and edges are not lost. + /// + private void DecodeStreamFrame(DaqifiOutMessage message, bool suppressAnalog) + { + var hasFloat = message.AnalogInDataFloat.Count > 0; + var hasRawAnalog = message.AnalogInData.Count > 0; + var hasDigital = message.DigitalData.Length > 0; + + if (!hasFloat && !hasRawAnalog && !hasDigital) + { + return; + } + + // 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 = _host.SnapshotChannels(); + + // 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 timestampResult = _timestampProcessor.ProcessTimestamp(StreamTimestampKey, deviceTimestamp); + var hostTimestamp = timestampResult.Timestamp; + + // Flag dropped samples from the device-clock delta (immune to host arrival jitter). + // Isolate subscriber exceptions (the device does that) so a throwing GapDetected handler + // cannot skip the per-channel decode below — which the caller's broad catch would then + // silently drop. + if (_gapDetector.IsGap(timestampResult.SecondsBetweenMessages)) + { + _host.RaiseGapDetected(new TimestampGapEventArgs( + hostTimestamp, timestampResult.SecondsBetweenMessages, deviceTimestamp)); + } + + if ((hasFloat || hasRawAnalog) && !suppressAnalog) + { + DecodeAnalog(message, channels, hostTimestamp, deviceTimestamp, hasFloat); + } + + if (hasDigital) + { + DecodeDigital(message, channels, hostTimestamp, deviceTimestamp); + } + } + + /// + /// The number of enabled analog channels in a channel snapshot. + /// + private static int CountEnabledAnalogChannels(IReadOnlyList channels) + { + var count = 0; + foreach (var channel in channels) + { + if (channel.IsEnabled && channel is IAnalogChannel) + { + count++; + } + } + return count; + } + + /// + /// 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. The firmware streams the whole DIO port as a raw pin-state + /// snapshot (the wire-level DIO enable is global, not per pin), so a channel's bit + /// position is its channel number — bit n lives at byte n / 8, bit + /// n % 8 (LSB first) — independent of which channels the client has enabled. + /// Output-direction channels are not sampled (their state is client-driven via + /// ). Channels whose number lies beyond the + /// payload get no sample rather than a bogus "low" reading. + /// + private static void DecodeDigital( + DaqifiOutMessage message, + IReadOnlyList channels, + DateTime hostTimestamp, + uint deviceTimestamp) + { + var digitalData = message.DigitalData; + var bitCount = digitalData.Length * 8; + + foreach (var channel in channels) + { + if (!channel.IsEnabled || channel.Type != ChannelType.Digital) + { + continue; + } + + // Only input-direction channels carry a meaningful streamed reading. + if (channel.Direction != ChannelDirection.Input) + { + continue; + } + + var bitIndex = channel.ChannelNumber; + if (bitIndex >= bitCount) + { + continue; + } + + var bit = (digitalData[bitIndex / 8] & (1 << (bitIndex % 8))) != 0; + + channel.SetActiveSample( + new DataSample(hostTimestamp, bit ? 1.0 : 0.0, bit ? 1 : 0, deviceTimestamp)); + } + } + } +} From 6a2e657eb0017bd94581134c6a582825ed609553 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Wed, 5 Aug 2026 13:26:02 -0600 Subject: [PATCH 2/3] docs: log the frame-decode extraction fire (#435) Co-Authored-By: Claude Opus 5 --- SESSION_LOG.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/SESSION_LOG.md b/SESSION_LOG.md index 99ee098..8c105f8 100644 --- a/SESSION_LOG.md +++ b/SESSION_LOG.md @@ -28,3 +28,15 @@ - Tests: +8 (message wording for both diagnostics, update-flow wording regression guard, Reset disposed-guard symmetry, cancellation for both, callback-reentrancy rejection, concurrent-call-waits). FULL suite green net9 + net10 (1783 total, 1781 passed, 2 skipped), 0 warnings. - Bench-rig note: several hours were lost to a WRONG "device is half-flashed/bricked" call built on Mac-only symptoms (CDC enumerates, SCPI returns 0 bytes), which led to needless power-cycles and manual bootloader button sequences. The unit worked fine on Windows and over WiFi; a reconnect fixed it. Triage next time by reading the port from the shell first (`stty ... -crtscts; cat &; printf 'SYSTem:SYSInfoPB?\r\n' > `) to settle device-vs-host before touching hardware. - Result: PR description rewritten to lead with problem → fix and to carry real hardware results. Not merging — awaiting user review. + +## 2026-08-05 — Fire: ready-notes on #433/#434 + implemented #344 item 2 (frame decode) → PR #435 +- State at start: **#432 MERGED** (so the cap freed a slot), 2 open loop PRs (#433, #434) — under the 3-PR cap. Priority 1: both had 0 unresolved `qodo-code-review` threads (checked via GraphQL `reviewThreads`, not the summary comment). Priority 2: #433 was Qodo-clean + CI green with no note → added a one-line ready-for-review note. #434's Qodo review was still running at that moment (PR Summary posted <1 min earlier), so it wasn't actionable yet; it came back clean mid-fire and got its ready-note too. Then priority 4. +- Backlog is still the same 5 open issues; 4 remain skips under the loop rules (#333 breaking API; #269/#271 destructive WINC flash — the non-destructive slice of #269 was already taken by #434; #183 needs an unshipped firmware dep + a new NuGet package). #344 is the only eligible one. +- PICKED #344 item 2 (**frame decode**), which the PREVIOUS fire explicitly deferred because it needs `SnapshotChannels()` on the seam and #432 was still unmerged and owned exactly that. #432 merging is what unblocked it — this is the clean follow-up that fire predicted, taken as soon as it became available rather than left on the floor. +- Extracted the whole hot path into `StreamFrameDecoder` (internal, `Device/Internal/`): the cross-session leftover gate, the #351 warmup guard, timestamp reconstruction, gap detection, analog/digital unpacking, the session state `BeginSession` resets together, and the two per-session counters. `DaqifiStreamingDevice` 1,756 → 1,368. Public API unchanged. +- **What deliberately did NOT move, and why:** `StreamFrameDiscarded`/`GapDetected` stay on the device (their `sender` must remain the device — same class of silent, compile-clean bug #419 needed a guard test for), and the raw re-raise must be `base.OnStreamMessageReceived` so a subclass override still sees the frame (calling the override would recurse; commented at the call site). `SafeTrace` stays because the session-command tracking still uses it. Four new `IDeviceOperationHost` members carry those callbacks. The decoder increments the discard counter *before* calling back, preserving the documented "count already includes the frame being reported" guarantee. +- Verified as a move mechanically, same as #419/#422/#432/#433: normalized statement multiset diff leaves only structural residues (new-file scaffolding, the 4 seam forwarders, 2 expression-bodied delegations, and the split of `RaiseStreamFrameDiscarded(reason, frame, counts)` into decoder-counts-and-builds-args / device-raises). No decode statement lost or altered. Also fixed a pre-existing doc/member mismatch the move surfaced: `DecodeAnalog`'s `` was attached to `CountEnabledAnalogChannels`. +- Tests: **zero edits to existing tests** (the 38 `DaqifiStreamingDeviceDecodeTests` cases still drive the same pipeline through the device — that's the evidence). +15 new cases against the collaborator directly, aimed at what only a direct test can see: the **order and multiplicity of the host callbacks**. Mutation-verified — counting the discard after the event fails 1 test; re-raising the raw frame unconditionally fails 2 (one of them a pre-existing device-level test); skipping the gap-detector reset in `BeginSession` fails 1. FULL suite green net9 + net10 (2,544 passed, 2 skipped; was 2,529) + Daqifi.Mcp.Tests 23. Release solution build 0 warnings both TFMs. +- BENCH (real Nq1, fw 3.7.2, USB, non-destructive) via a scratch harness, because the example CLI surfaces none of these counters. **Session 1 (ch 0,1,2 @200 Hz, 3 s):** `rawFrames=475` and `decodedCh0=475` — every delivered frame reached BOTH consumer paths exactly once (the re-raise multiplicity contract, on hardware); `discarded=1` with `PartialAnalogFrame[an=1/en=3]` — the firmware's malformed leading frame caught by the moved guard and withheld from raw consumers; all 3 channels decoded 475 samples each in ascending order; `decodeFailures=0`, `gaps=0`; every event's `sender` asserted to be the device. **Session 2 (ch 0 only @100 Hz, same instance):** `discarded=0`/`failures=0` (BeginSession reset both, no leftover tripped the gate); `rawFrames=238` vs `decodedCh0=237` — one post-stop frame re-raised but NOT decoded, exactly the `if (!IsStreaming)` branch on hardware; `ch1=ch2=0` decoded, so the disable reached the device and the snapshot the decode maps against is still right. Only channel enable/disable + stream start/stop; no NVM write, no reboot, no SD. +- Bench-rig note: the example CLI's `--channels` takes a **bitmask** (`7` = ch 0,1,2), not a comma list, and `--format` accepts only `text|csv|jsonl` (no `json`). +- Result: PR #435 opened (base main, part of #344, "not merging — for review"), /agentic_review requested. Now **3 loop PRs awaiting review (#433, #434, #435) — at the concurrency cap**, so the next fire should shepherd only, not start a new ticket. From 0e119468657ec57e634704d53f68c05139ce7a7b Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Wed, 5 Aug 2026 13:37:05 -0600 Subject: [PATCH 3/3] docs(device): state StreamFrameDecoder's real threading contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The remarks claimed the unsynchronized session fields "are only ever touched from" the message-consumer thread. That is not true of BeginSession, which runs on whichever thread called StartStreaming() or sent a raw start-streaming command through Send() — so the contract as written invited a future change to assume a thread confinement the API surface does not provide. Replaced with what actually holds: the decode path is consumer-thread only, BeginSession is the exception, and what makes it sound is the session boundary rather than synchronization — StartStreaming resets before sending the command, and the raw-Send path leaves IsStreaming false until its reset completes, so a frame landing in that window is re-raised as a stray and never decoded. Documentation only; no behavior change. Co-Authored-By: Claude Opus 5 --- .../Device/Internal/StreamFrameDecoder.cs | 26 +++++++++++++++---- 1 file changed, 21 insertions(+), 5 deletions(-) diff --git a/src/Daqifi.Core/Device/Internal/StreamFrameDecoder.cs b/src/Daqifi.Core/Device/Internal/StreamFrameDecoder.cs index 625aa7d..1fbaaa4 100644 --- a/src/Daqifi.Core/Device/Internal/StreamFrameDecoder.cs +++ b/src/Daqifi.Core/Device/Internal/StreamFrameDecoder.cs @@ -31,11 +31,27 @@ namespace Daqifi.Core.Device.Internal /// still intercepts it. /// /// - /// Not thread-safe by design, matching the code it was moved from: frames arrive on the single - /// message-consumer thread, which is also the thread that repopulates channels, so the - /// unsynchronized session fields are only ever touched from there. The two counters are - /// interlocked because they are read from arbitrary threads through the device's public - /// properties. + /// Not thread-safe, matching the code it was moved from. The decode path — + /// and everything under it — runs solely on the single + /// message-consumer thread, which is also the thread that repopulates channels. + /// is the exception, and the reason this is stated rather than + /// assumed: it runs on whichever thread called + /// or sent a raw start-streaming command through Send, and it writes the session fields + /// with no synchronization against the consumer thread reading them. + /// + /// + /// What makes that sound is the session boundary, not synchronization: a session's frames only + /// exist once the device has been told to start, and the two starts bracket the reset so that + /// no frame is decoded as part of a session whose reset has not run. StartStreaming + /// resets before it sends the command. The raw-Send path can only reset after the + /// command has gone out, but it leaves false + /// until the reset completes, so a frame landing in that window is re-raised as a stray frame + /// and never decoded. Starting a stream concurrently with another thread's in-flight session is + /// outside the device's contract, exactly as it was before this was extracted. + /// + /// + /// The two counters are interlocked because they are read from arbitrary threads through the + /// device's public properties. /// /// internal sealed class StreamFrameDecoder