Add support for load tester device

- Modify the existing configuration operator since it had implementation
  issues
- Add frame reader and frame writer operators and support classes

Author: jonnew

OpenEphys.Onix1/ConfigureLoadTester.cs

@@ -1,27 +1,21 @@
 using System;
 using System.ComponentModel;
-using System.Linq;
-using System.Reactive.Disposables;
-using System.Reactive.Subjects;
 using Bonsai;
 
 namespace OpenEphys.Onix1
 {
-    // TODO: Add data IO operators, update XML comment to link to them (<see cref="LoadTesterData"/>)
     /// <summary>
-    /// Configures a load tester device.
+    /// Configures a load tester device for measuring system performance.
     /// </summary>
     /// <remarks>
-    /// This configuration operator can be linked to a data IO operator, such as LoadTesterData,
-    /// using a shared <c>DeviceName</c>. The load tester device can be configured
-    /// to produce data at user-settable size and rate to stress test various communication links and test
-    /// closed-loop response latency.
+    /// This configuration operator can be linked to a data IO operator, such as <see cref="LoadTesterData"/>,
+    /// using a shared <c>DeviceName</c>. The load tester device can be configured to produce and consume data
+    /// at user-defined sizes and rates to stress test various communication links and measure closed-loop
+    /// response latency using a high-resolution hardware timer.
     /// </remarks>
     [Description("Configures a load testing device.")]
     public class ConfigureLoadTester : SingleDeviceFactory
     {
-        readonly BehaviorSubject<uint> frameHz = new(1000);
-
         /// <summary>
         /// Initializes a new instance of the <see cref="ConfigureLoadTester"/> class.
         /// </summary>
@@ -38,31 +32,37 @@ public ConfigureLoadTester()
         public bool Enable { get; set; } = false;
 
         /// <summary>
-        /// Gets or sets the number of repetitions of the 16-bit unsigned integer 42 sent with each read-frame.
+        /// Gets or sets the number of repetitions incrementing, unsigned 16-bit integers sent with each read frame.
         /// </summary>
+        /// <remarks>
+        /// These data are produced by the controller and are used to impose a load the controller to host
+        /// communication. These data can be used in downstream computational operations that model the
+        /// computational load imposed by a closed-loop algorithm.
+        /// </remarks>
         [Category(ConfigurationCategory)]
         [Description("Number of repetitions of the 16-bit unsigned integer 42 sent with each read-frame.")]
         [Range(0, 10e6)]
         public uint ReceivedWords { get; set; }
 
         /// <summary>
-        /// Gets or sets the number of repetitions of the 32-bit integer 42 sent with each write frame.
+        /// Gets or sets the number of repetitions of the 32-bit integer dummy words sent with each write
+        /// frame.
         /// </summary>
+        /// <remarks>
+        /// These data are produced by the host and are used to impose a load on host to controller
+        /// commutation. They are discarded by the controller when they are received.
+        /// </remarks>
         [Category(ConfigurationCategory)]
         [Description("Number of repetitions of the 32-bit integer 42 sent with each write frame.")]
         [Range(0, 10e6)]
         public uint TransmittedWords { get; set; }
 
         /// <summary>
-        /// Gets or sets a value specifying the rate at which frames are produced, in Hz.
+        /// Gets or sets a value specifying the rate at which frames are produced in Hz.
         /// </summary>
-        [Category(AcquisitionCategory)]
+        [Category(ConfigurationCategory)]
         [Description("Specifies the rate at which frames are produced (Hz).")]
-        public uint FramesPerSecond
-        {
-            get { return frameHz.Value; }
-            set { frameHz.OnNext(value); }
-        }
+        public uint FramesPerSecond { get; set; }
 
         /// <summary>
         /// Configures a load testing device.
@@ -83,40 +83,38 @@ public override IObservable<ContextTask> Process(IObservable<ContextTask> source
             var deviceAddress = DeviceAddress;
             var receivedWords = ReceivedWords;
             var transmittedWords = TransmittedWords;
+            var framesPerSecond = FramesPerSecond;
+
             return source.ConfigureDevice((context, observer) =>
             {
                 var device = context.GetDeviceContext(deviceAddress, DeviceType);
                 device.WriteRegister(LoadTester.ENABLE, enable ? 1u : 0u);
 
                 var clockHz = device.ReadRegister(LoadTester.CLK_HZ);
 
-                // Assumes 8-byte timer
-                uint ValidSize()
+                const int OverheadCycles = 9; // 4 cycles to produce hub clock, and 5 state machine overhead per the datasheet
+
+                var maxFramesPerSecond = clockHz / OverheadCycles;
+                if (framesPerSecond > maxFramesPerSecond)
                 {
-                    var clkDiv = device.ReadRegister(LoadTester.CLK_DIV);
-                    return clkDiv - 4 - 10; // -10 is overhead hack
+                    throw new ArgumentOutOfRangeException(nameof(FramesPerSecond), $"{nameof(FramesPerSecond)} must be less than {maxFramesPerSecond}.");
                 }
 
-                var maxSize = ValidSize();
-                var bounded = receivedWords > maxSize ? maxSize : receivedWords;
-                device.WriteRegister(LoadTester.DT0H16_WORDS, bounded);
+                device.WriteRegister(LoadTester.CLK_DIV, clockHz / framesPerSecond);
 
-                var writeArray = Enumerable.Repeat((uint)42, (int)(transmittedWords + 2)).ToArray();
-                device.WriteRegister(LoadTester.HTOD32_WORDS, transmittedWords);
-                var frameHzSubscription = frameHz.SubscribeSafe(observer, newValue =>
+                var maxSize = device.ReadRegister(LoadTester.CLK_DIV) - OverheadCycles;
+
+                if (receivedWords > maxSize)
                 {
-                    device.WriteRegister(LoadTester.CLK_DIV, clockHz / newValue);
-                    var maxSize = ValidSize();
-                    if (receivedWords > maxSize)
-                    {
-                        receivedWords = maxSize;
-                    }
-                });
-
-                return new CompositeDisposable(
-                    DeviceManager.RegisterDevice(deviceName, device, DeviceType),
-                    frameHzSubscription
-                );
+                    throw new ArgumentOutOfRangeException(nameof(ReceivedWords), 
+                        $"{nameof(ReceivedWords)} must be less than {maxSize} for the requested frame rate of {framesPerSecond} Hz.");
+                }
+
+                device.WriteRegister(LoadTester.DT0H16_WORDS, receivedWords);
+                device.WriteRegister(LoadTester.HTOD32_WORDS, transmittedWords);
+                
+                var deviceInfo = new LoadTesterDeviceInfo(context, DeviceType, deviceAddress, ReceivedWords, TransmittedWords);
+                return DeviceManager.RegisterDevice(deviceName, deviceInfo);
             });
         }
     }
@@ -132,12 +130,13 @@ static class LoadTester
         public const uint DT0H16_WORDS = 3; // Number of repetitions of 16-bit unsigned integer 42 sent with each frame. 
                                             // Note: max here depends of CLK_HZ and CLK_DIV. There needs to be enough clock
                                             // cycles to push the data at the requested CLK_HZ. Specifically,
-                                            // CLK_HZ / CLK_DIV >= TX16_WORDS + 9. Going above this will result in 
+                                            // CLK_HZ / CLK_DIV >= DT0H16_WORDS + 9. Going above this will result in 
                                             // decreased bandwidth as samples will be skipped.
         public const uint HTOD32_WORDS = 4; // Number of 32-bit words in a write-frame. All write frame data is ignored except
                                             // the first 64-bits, which are looped back into the device to host data frame for   
                                             // testing loop latency. This value must be at least 2.
 
+
         internal class NameConverter : DeviceNameConverter
         {
             public NameConverter()

OpenEphys.Onix1/LoadTesterData.cs

@@ -0,0 +1,42 @@
+using System;
+using System.ComponentModel;
+using System.Linq;
+using System.Reactive.Linq;
+using Bonsai;
+
+namespace OpenEphys.Onix1
+{
+    /// <summary>
+    /// Produces a sequence of heartbeat data frames.
+    /// </summary>
+    /// <remarks>
+    /// This data IO operator must be linked to an appropriate configuration, such as a <see
+    /// cref="ConfigureHeartbeat"/>, using a shared <c>DeviceName</c>.
+    /// </remarks>
+    [Description("Produces a sequence of load tester data frames.")]
+    public class LoadTesterData : Source<LoadTesterDataFrame>
+    {
+        /// <inheritdoc cref = "SingleDeviceFactory.DeviceName"/>
+        [TypeConverter(typeof(LoadTester.NameConverter))]
+        [Description(SingleDeviceFactory.DeviceNameDescription)]
+        [Category(DeviceFactory.ConfigurationCategory)]
+        public string DeviceName { get; set; }
+
+        /// <summary>
+        /// Generates a sequence of <see cref="LoadTesterDataFrame"/> objects, each of which contains period signal from the
+        /// acquisition system indicating that it is active.
+        /// </summary>
+        /// <returns>A sequence of <see cref="LoadTesterDataFrame"/> objects.</returns>
+        public override IObservable<LoadTesterDataFrame> Generate()
+        {
+            return DeviceManager.GetDevice(DeviceName).SelectMany(deviceInfo =>
+            {
+                var info = (LoadTesterDeviceInfo)deviceInfo;
+                var device = info.GetDeviceContext(typeof(LoadTester));
+                return deviceInfo.Context
+                    .GetDeviceFrames(device.Address)
+                    .Select(frame => new LoadTesterDataFrame(frame, info.ReceivedWords));
+            });
+        }
+    }
+}

OpenEphys.Onix1/LoadTesterDataFrame.cs

@@ -0,0 +1,61 @@
+using System;
+using System.Runtime.InteropServices;
+
+namespace OpenEphys.Onix1
+{
+    /// <summary>
+    /// A load tester data frame.
+    /// </summary>
+    public class LoadTesterDataFrame : DataFrame
+    {
+        /// <summary>
+        /// Gets the difference between the hub clock and the loopback value at the moment the loopback
+        /// value was received.
+        /// </summary>
+        /// <remarks>
+        /// This value is the result of subtracting the loopback value written by the <see
+        /// cref="LoadTesterLoopback"/> operator from the device's hub clock counter value at the moment that
+        /// the loop back value was received. Typically, the <see cref="DataFrame.HubClock"/> member is sent
+        /// to <see cref="LoadTesterLoopback"/> operator such that this value is a hardware-timed measurement
+        /// value of real-time latency. This value is only updated when a new loopback value is sent to the
+        /// device using the <see cref="LoadTesterLoopback"/> operator.
+        /// </remarks>
+        public ulong HubClockDelta { get; }
+
+        /// <summary>
+        /// Gets the counter array.
+        /// </summary>
+        /// <remarks>
+        /// This is a <see cref="ConfigureLoadTester.ReceivedWords"/>-long array of incrementing integers that
+        /// is used for simulating the bandwidth of physical data sources.
+        /// </remarks>
+        public ushort[] Counter { get; }
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="LoadTesterDataFrame"/> class.
+        /// </summary>
+        /// <param name="frame">A data frame produced by a load tester device.</param>
+        /// <param name="receivedWords">The number of counter words that appear at the end of the load test
+        /// data frame. This number is determined by the value of <see
+        /// cref="ConfigureLoadTester.ReceivedWords"/>.</param>
+        public unsafe LoadTesterDataFrame(oni.Frame frame, uint receivedWords)
+            : base(frame.Clock)
+        {
+            var payload = (LoadTesterPayload*)frame.Data.ToPointer();
+            var counterPtr = (ushort*)((byte*)payload + sizeof(LoadTesterPayload));
+
+            HubClock = payload->HubClock;
+            HubClockDelta = payload->HubClockDelta;
+            Counter = new Span<ushort>(counterPtr, (int)receivedWords).ToArray();
+        }
+    }
+
+    [StructLayout(LayoutKind.Sequential, Pack = 1)]
+    struct LoadTesterPayload
+    {
+        public ulong HubClock;
+        public ulong HubClockDelta;
+        // NB: The ushort Counter array may or may not reside here. Its size is determined by ReceivedWords.
+    }
+}
+

OpenEphys.Onix1/LoadTesterDeviceInfo.cs

@@ -0,0 +1,18 @@
+using System;
+
+namespace OpenEphys.Onix1
+{
+    class LoadTesterDeviceInfo : DeviceInfo
+    {
+        public LoadTesterDeviceInfo(ContextTask context, Type deviceType, uint deviceAddress, uint receivedWords, uint transmittedWords)
+            : base(context, deviceType, deviceAddress)
+        {
+            ReceivedWords = receivedWords;
+            TransmittedWords = transmittedWords;
+        }
+
+        public uint ReceivedWords { get; }
+
+        public uint TransmittedWords { get; }
+    }
+}

OpenEphys.Onix1/LoadTesterLoopback.cs

@@ -0,0 +1,68 @@
+using System;
+using System.ComponentModel;
+using System.Linq;
+using System.Reactive.Linq;
+using Bonsai;
+
+namespace OpenEphys.Onix1
+{
+
+    /// <summary>
+    /// Sends loopback data to the load testing device for closed-loop latency measurement.
+    /// </summary>
+    /// <remarks>
+    /// This data IO operator must be linked to an appropriate configuration, such as a <see
+    /// cref="ConfigureLoadTester"/>, using a shared <c>DeviceName</c>.
+    /// </remarks>
+    [Description("Sends analog output data to an ONIX breakout board.")]
+    public class LoadTesterLoopback : Sink<ulong>
+    {
+        /// <inheritdoc cref = "SingleDeviceFactory.DeviceName"/>
+        [TypeConverter(typeof(LoadTester.NameConverter))]
+        [Description(SingleDeviceFactory.DeviceNameDescription)]
+        [Category(DeviceFactory.ConfigurationCategory)]
+        public string DeviceName { get; set; }
+
+        /// <summary>
+        /// Creates and sends an loopback frame to the load testing device.
+        /// </summary>
+        /// <remarks>
+        /// A loopback frame consists the <c>ulong</c> loopback value provided by <paramref name="source"/>
+        /// that is prepended to a <see cref="ConfigureLoadTester.TransmittedWords"/>-element <c>ushort</c>
+        /// array of dummy data. When the frame is received by hardware, the loopback value is subtracted from
+        /// the current hub clock count on the load testing device and stored. Therefore, if the loopback
+        /// value is is that of a previous <see cref="DataFrame.HubClock"/> from the <see
+        /// cref="LoadTesterData"/> with the same <see cref="DeviceName"/> as this operator, this difference will provide a
+        /// hardware-timed measurement of real-time latency. The the variably-sized dummy data in the loopback
+        /// frame is used for testing the effect of increasing the frame size, and thus the write
+        /// communication bandwidth, on real-time latency.
+        /// </remarks>
+        /// <param name="source">A sequence of loopback values to send to the device</param>
+        /// <returns> A sequence of loopback values to send to the device.</returns>
+        public unsafe override IObservable<ulong> Process(IObservable<ulong> source)
+        {
+            return DeviceManager.GetDevice(DeviceName).SelectMany(deviceInfo =>
+            {
+                var info = (LoadTesterDeviceInfo)deviceInfo;
+                var device = info.GetDeviceContext(typeof(LoadTester));
+                var transmittedWords = info.TransmittedWords;
+
+                if (transmittedWords > 0)
+                {
+                    var payload = new uint[transmittedWords + 2];
+
+                    return source.Do(loopbackValue =>
+                    {
+                        payload[0] = (uint)loopbackValue;
+                        payload[1] = (uint)(loopbackValue >> 32);
+                        device.Write(payload);
+                    });
+                }
+                else
+                {
+                    return source.Do(device.Write);
+                }
+            });
+        }
+    }
+}

OpenEphys.Onix1/OpenEphys.Onix1.csproj

@@ -16,5 +16,6 @@
     <PackageReference Include="Microsoft.Bcl.Numerics" Version="9.0.7" />
     <PackageReference Include="OpenCV.Net" Version="3.4.2" />
     <PackageReference Include="OpenEphys.ProbeInterface.NET" Version="0.3.0" />
+    <PackageReference Include="System.Memory" Version="4.6.3" />
   </ItemGroup>
 </Project>