fix: fix USB streaming — AnalogInDataFloat support and USB stream interface init (#149)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: tylerkron

src/Daqifi.Core.Tests/Device/Protocol/ProtobufProtocolHandlerTests.cs

@@ -122,20 +122,55 @@ public async void HandleAsync_WithNonProtobufMessage_DoesNotCallHandlers()
         Assert.False(streamHandlerCalled);
     }
 
+    [Fact]
+    public async Task HandleAsync_WithFloatStreamMessage_CallsStreamHandler()
+    {
+        // Arrange - USB firmware sends pre-scaled float values (AnalogInDataFloat)
+        var streamHandlerCalled = false;
+        DaqifiOutMessage receivedMessage = null;
+
+        var handler = new ProtobufProtocolHandler(
+            streamMessageHandler: msg =>
+            {
+                streamHandlerCalled = true;
+                receivedMessage = msg;
+            });
+
+        var streamMessage = new DaqifiOutMessage
+        {
+            MsgTimeStamp = 99999
+        };
+        streamMessage.AnalogInDataFloat.Add(1.234f);
+        streamMessage.AnalogInDataFloat.Add(2.345f);
+
+        var inboundMessage = new GenericInboundMessage<object>(streamMessage);
+
+        // Act
+        await handler.HandleAsync(inboundMessage);
+
+        // Assert
+        Assert.True(streamHandlerCalled, "Stream handler should be called for AnalogInDataFloat messages");
+        Assert.NotNull(receivedMessage);
+        Assert.Equal(99999u, receivedMessage.MsgTimeStamp);
+        Assert.Equal(2, receivedMessage.AnalogInDataFloat.Count);
+    }
+
     [Theory]
-    [InlineData(8u, 0u, 0u, 0u, 0, 0, ProtobufMessageType.Status)]
-    [InlineData(0u, 16u, 0u, 0u, 0, 0, ProtobufMessageType.Status)]
-    [InlineData(0u, 0u, 2u, 0u, 0, 0, ProtobufMessageType.Status)]
-    [InlineData(0u, 0u, 0u, 12345u, 1, 0, ProtobufMessageType.Stream)]
-    [InlineData(0u, 0u, 0u, 12345u, 0, 1, ProtobufMessageType.Stream)]
-    [InlineData(0u, 0u, 0u, 0u, 0, 0, ProtobufMessageType.Unknown)]
+    [InlineData(8u, 0u, 0u, 0u, 0, 0, false, ProtobufMessageType.Status)]
+    [InlineData(0u, 16u, 0u, 0u, 0, 0, false, ProtobufMessageType.Status)]
+    [InlineData(0u, 0u, 2u, 0u, 0, 0, false, ProtobufMessageType.Status)]
+    [InlineData(0u, 0u, 0u, 12345u, 1, 0, false, ProtobufMessageType.Stream)]   // int data
+    [InlineData(0u, 0u, 0u, 12345u, 0, 1, false, ProtobufMessageType.Stream)]   // digital data
+    [InlineData(0u, 0u, 0u, 12345u, 0, 0, true, ProtobufMessageType.Stream)]    // float data (USB firmware)
+    [InlineData(0u, 0u, 0u, 0u, 0, 0, false, ProtobufMessageType.Unknown)]
     public void DetectMessageType_ReturnsCorrectType(
         uint analogInPortNum,
         uint digitalPortNum,
         uint analogOutPortNum,
         uint msgTimeStamp,
         int analogDataCount,
         int digitalDataLength,
+        bool hasFloatData,
         ProtobufMessageType expectedType)
     {
         // Arrange
@@ -157,6 +192,11 @@ public void DetectMessageType_ReturnsCorrectType(
             message.DigitalData = Google.Protobuf.ByteString.CopyFrom(new byte[digitalDataLength]);
         }
 
+        if (hasFloatData)
+        {
+            message.AnalogInDataFloat.Add(1.5f);
+        }
+
         // Act
         var result = ProtobufProtocolHandler.DetectMessageType(message);
 

src/Daqifi.Core/Communication/Consumers/StreamMessageConsumer.cs

@@ -101,6 +101,7 @@ public bool StopSafely(int timeoutMs = 1000)
         _isRunning = false;
         var stopped = _consumerThread?.Join(timeoutMs) ?? true;
         _consumerThread = null;
+        _messageBuffer.Clear();
 
         return stopped;
     }

src/Daqifi.Core/Device/DaqifiStreamingDevice.cs

@@ -1,3 +1,4 @@
+using Daqifi.Core.Communication;
 using Daqifi.Core.Communication.Messages;
 using Daqifi.Core.Communication.Producers;
 using Daqifi.Core.Communication.Transport;
@@ -101,6 +102,45 @@ public DaqifiStreamingDevice(string name, IStreamTransport transport) : base(nam
             StreamingFrequency = 100;
         }
 
+        /// <summary>
+        /// Initializes the device with the standard SCPI sequence and, for USB/serial connections,
+        /// additionally sets the streaming interface to USB so data is routed to the serial consumer
+        /// rather than to a previously-configured WiFi destination.
+        /// </summary>
+        /// <remarks>
+        /// The DAQiFi firmware persists the last configured stream interface across sessions.
+        /// If the device was previously set to stream to WiFi (<c>SYSTem:STReam:INTerface 1</c>),
+        /// it will continue sending data over WiFi even when connected via USB — causing the serial
+        /// consumer to receive nothing. Sending <c>SYSTem:STReam:INTerface 0</c> during USB
+        /// initialization ensures data flows to the serial port.
+        /// </remarks>
+        /// <returns>A task representing the asynchronous initialization operation.</returns>
+        public override async Task InitializeAsync()
+        {
+            // Capture pre-init state so we can skip the USB step on repeated calls
+            // (base.InitializeAsync() guards with _isInitialized and returns early on repeats).
+            var needsUsbInit = IsUsbConnection && State != DeviceState.Ready;
+
+            await base.InitializeAsync();
+
+            if (needsUsbInit)
+            {
+                // Direct streaming to the USB interface. Uses ExecuteTextCommandAsync so the
+                // command is sent in text mode (protobuf consumer temporarily stopped) and any
+                // SCPI error response is captured rather than garbling the protobuf stream.
+                var lines = await ExecuteTextCommandAsync(
+                    () => Send(ScpiMessageProducer.SetStreamInterface(StreamInterface.Usb)),
+                    responseTimeoutMs: 500,
+                    cancellationToken: CancellationToken.None);
+
+                if (ContainsScpiError(lines))
+                {
+                    throw new InvalidOperationException(
+                        "Device returned a SCPI error while setting stream interface to USB.");
+                }
+            }
+        }
+
         /// <summary>
         /// Starts streaming data from the device at the configured frequency.
         /// </summary>

src/Daqifi.Core/Device/Protocol/ProtobufProtocolHandler.cs

@@ -135,8 +135,12 @@ private static bool IsStatusMessage(DaqifiOutMessage message)
     /// <returns><c>true</c> if the message contains streaming data; otherwise, <c>false</c>.</returns>
     private static bool IsStreamMessage(DaqifiOutMessage message)
     {
-        // Stream messages contain timestamp and data
+        // Stream messages contain timestamp and data.
+        // USB firmware sends AnalogInDataFloat (pre-scaled floats) while WiFi firmware
+        // may send AnalogInData (raw integer ADC counts). Both must be detected.
         return message.MsgTimeStamp != 0 &&
-               (message.AnalogInData.Count > 0 || message.DigitalData.Length > 0);
+               (message.AnalogInData.Count > 0 ||
+                message.AnalogInDataFloat.Count > 0 ||
+                message.DigitalData.Length > 0);
     }
 }