将 IObservable<Timestamped<T>> 转换为 IObservable<TimeInterval<T>>。

Question

我该如何将一个Timestamped<T>的可观察序列转换成TimeInterval<T>的序列，其中时间间隔是原始序列中时间戳之间的时间？

考虑到输入序列...

new Timestamped<int>(1, DateTime.Parse("2000-01-01 00:00:01"))
new Timestamped<int>(2, DateTime.Parse("2000-01-01 00:00:05"))
new Timestamped<int>(3, DateTime.Parse("2000-01-01 00:01:04"))

输出结果将是：

new TimeInterval<int>(1, TimeSpan.Parse("00:00:00"))
new TimeInterval<int>(2, TimeSpan.Parse("00:00:04"))
new TimeInterval<int>(3, TimeSpan.Parse("00:00:59"))

- chillitom

---

第二个序列需要具有相同数量的条目吗？ 您对这个隐式间隔需要什么行为？ （此外，是否结构化，以至于您无法只调用 originalSource.TimeInterval() ？） - user7116

时间戳和数值来自日志文件，我想要能够以不同的速度重放它们，并且希望从序列开始时获得相对时间，并按照这种方式进行安排。这些序列应该具有相同数量的元素。 - chillitom

Answer 1

我 认为 这样做可以。

var s = source.Publish().RefCount();
var sprev = s.Take(1).Concat(s);
var scurrent = s;

var converted = Observable.Zip(sprev, scurrent, (prev, current) =>
   new TimeInterval<int>(current.Value, current.Timestamp - prev.Timestamp));

我唯一不确定的是当任何一个序列结束时，Zip 是否也结束。我认为它会结束，但我尚未进行测试。

- Brandon

Answer 2

也许您可以使用简单的投影与 Do 结合使用：

static IObservable<TimeInterval<T>> ToTimeInterval<T>(
    this IObservable<Timestamped<T>> source)
{
    DateTimeOffset? previous = null;
    return source.Select(ts => 
        new
        {
            Timestamp = ts.Timestamp,
            Value = ts.Value,
            TimeSpan = previous.HasValue ? ts.Timestamp - previous
                                         : TimeSpan.FromSeconds(0)
        })
        .Do(xx => { previous = xx.Timestamp; })
        .Select(xx => new TimeInterval<T>(xx.Value, xx.TimeSpan));
}

用法如下：

var intervals = stampedData.ToTimeInterval();

- user7116

Answer 3

我对可观察对象不是很了解，但你能做到吗：

 myInputSequence.ToEnumerable().Select(t => 
    new TimeInterval<int>(
        t.Value,
        t.Value == 1 
            ? new TimeSpan(0) 
            : t.Timestamp - System.Reactive.Linq.Observable.ToEnumerable(myList).FirstOrDefault(t2 => t2.Value == t.Value - 1).Timestamp)
        ).ToObservable();

当然，这并不是尽可能高效的解决方法，特别是如果你知道日志语句是按顺序到达的。

- Ben Reich