타임 스탬프를 기반으로 ReactiveX로 이동 평균을 구현합니다.

Question

저는 Rx.Net을 사용하고 있으며 Observable을 사용하여 시계열 점 (double, timestamp)을 발생시킵니다. 때마다 새로운 점은 내가 지난 30 초 말할 수에서 평균 값을 계산하려는 도착한다. count/timestamp를 기반으로하지 않는 Window/Buffer를 겹쳐 쓰는 것이 필요하다고 생각합니다.

나는 SlidingWindow의 구현 this 주제를 발견했습니다,하지만 난 내 문제에이 맞게하는 방법을 알아낼 수 없습니다.

편집 : this에

덕분에 내가 검색 연산자를 사용하여 내 포인트를 버퍼링 할 수 배웠습니다, 그래서 basicly이 문제를 해결합니다. 그러나 아마도 이것을 할 수있는 더 좋은 방법이 있을까요?

Answer 1

Buffer 및 Window는 다시 보이는 무언가를 원하는, 기대합니다. 당신이 BackBuffer, 또는 같은 것을 일단

public static IObservable<List<T>> BackBuffer<T>(this IObservable<T> source, TimeSpan ts) 
{ 
    return BackBuffer(source, ts, Scheduler.Default); 
} 
public static IObservable<List<T>> BackBuffer<T>(this IObservable<T> source, TimeSpan ts, IScheduler scheduler) 
{ 
    return source 
     .Timestamp() 
     .Scan(new List<Timestamped<T>>(), (list, element) => list 
      .Where(ti => scheduler.Now - ti.Timestamp <= ts) 
      .Concat(Enumerable.Repeat(element, 1)) 
      .ToList() 
     ) 
     .Select(list => list.Select(t => t.Value).ToList()); 
} 

, 그럼 나머지는 쉽게된다 :

source 
    .BackBuffer(TimeSpan.FromMilliseconds(70)) 
    .Select(list => list.Average()) 
    .Subscribe(average => Console.WriteLine(average)); 

Scan 최고의 출발점이 될 것입니다