我是否可以使用反应式扩展来控制仪器测试周期的时间？

Question

我想解决的问题是..。

我在写一个控制测试设备的程序。该程序每秒钟通过串行接口对仪器进行轮询，并以一串测试数据作为回报。

连接速度为9600波特，仪器一次只发送100个字符，因此性能不成问题。

要求之一是“循环”模式，其工作方式如下：

1. 用户设置三个计时器间隔-测试前时间、测试时间和测试间时间-并单击UI上的“开始”按钮。
2. 该程序命令仪器启动测试周期，并显示仪器接收到的数据，直到测试前时间结束为止。
3. 然后，程序显示并记录数据，直到测试时间过期为止。
4. 然后，该程序将显示数据，直到测试时间过期为止。
5. 重复步骤2-4，直到用户单击UI上的“停止”按钮。

在过去，我为这类问题编写了简单的状态机。但是对于这个项目(在SO专家的帮助下)，我已经能够使用Rx来解析串行数据，并且对得到的代码非常满意。

我的直觉告诉我，这个问题也可能被Rx很好地处理，但是它根本不清楚如何继续。

那么，我的直觉是正确的，还是你会建议另一种方法？

如果Rx很适合，那么有人能提供一些快速而肮脏的示例代码来演示解决方案吗？

每一个谜团的Edit...as：

这是读取串行数据的代码。简而言之，它将串行数据流转换为单个数据包字符串。"A0“、"C00004”、"H0501100“等被传递给ProcessPacket方法，该方法将数据包解释为仪器状态、压力等。

    /// <summary>
    /// Create a Reactive Extensions subscription to the selected serial port
    /// </summary>
    private IDisposable SetupSerialDataSubscription()
    {
        // Create Observable stream of chars from serial port's DataRecieved event
        var serialData = Observable
                .FromEventPattern<SerialDataReceivedEventArgs>(_serialPort, "DataReceived")

                // Collapse groups of chars into stream of individual chars
                .SelectMany(_ =>
                {
                    int bytesToRead = _serialPort.BytesToRead;
                    byte[] bytes = new byte[bytesToRead];
                    int nbrBytesRead = _serialPort.Read(bytes, 0, bytesToRead);

                    // Empty array if no bytes were actually read
                    if (nbrBytesRead == 0)
                        return new char[0];

                    var chars = Encoding.ASCII.GetChars(bytes);

                    return chars;
                })

                // Strip out null chars, which can cause trouble elsewhere (e.g. Console.WriteLine)
               .SkipWhile(c => c == '\0');

        // Emits packets, one string per packet, e.g. "A0", "C00004", "H0501100", etc
        var packets = serialData
            .Scan(string.Empty, (prev, cur) => char.IsLetter(cur) ? cur.ToString() : prev + cur)
            .Where(IsCompletePacket);

        // Process each packet, on the UI thread
        var packetsSubscription = packets
            .ObserveOn(this)
            .Subscribe(packet => ProcessPacket(packet) );

        return packetsSubscription;
    }

Answer 1

你能做到的。使用助手方法发出无限的IEnumerable，它传递一个记录信号(真/假)和一个时隙：

public IEnumerable<Tuple<TimeSpan, bool>> GetTestPeriods()
{
    TimeSpan preTestTime = TimeSpan.FromSeconds(5);
    TimeSpan testTime = TimeSpan.FromSeconds(3);
    TimeSpan interTestTime = TimeSpan.FromSeconds(2);

    yield return Tuple.Create(preTestTime, false);
    while(true)
    {
        yield return Tuple.Create(testTime, true);
        yield return Tuple.Create(interTestTime, false);
    }
}

这可以用来创建一个循环流，在适当的时候输出一个记录信号：

// cycle outputs true at start of recording period
// false at start of a display period
var cycle = GetTestPeriods()
    .Select(s => Observable.Return(s.Item2)
        .Concat(Observable.Empty<bool>().Delay(s.Item1)))
    .Concat();

现在我们可以做一些随机数据：

// make some random data
var random = new Random();
var dummyData = Observable.Interval(TimeSpan.FromSeconds(1))
                          .Select(_ => random.Next(10));

设置一个信号停止主题(您可以使用FromEventPattern挂钩按钮点击这里)：

var stop = new Subject<Unit>();

然后将数据与记录信号组合，直到停止信号发出为止。这对每个数据点都有一个布尔值，指示它是否要显示或记录：

var output = cycle.CombineLatest(dummyData, (c,d) => Tuple.Create(c,d))
    .TakeUntil(stop);

这个流可以使用Where过滤器发布和订阅几个流，以便将数据与要显示的数据分开记录。下面是一个测试，用于时间戳流并将数据转储出去：

// timestamp data and "record" or "display" it.
output.Timestamp()
    .Subscribe(x => {
        if(x.Value.Item1)
            Console.WriteLine(x.Timestamp + ": Recording " + x.Value.Item2);
        else
            Console.WriteLine(x.Timestamp + ": Displaying " + x.Value.Item2);
    },
    () => Console.WriteLine("Stopped")); 

最后，发出停止信号的代码：

Console.ReadLine();
stop.OnNext(Unit.Default);