Sqooba / scala-timeseries-lib / 105

pending completion

Build # 105

Build Type

push

travis-ci-com

Committed by

Shastick

Commit Message

<a href="https://github.com/Sqooba/scala-timeseries-lib/commit/5716f9601">Bump scalafmt version and configuration

GitOrigin-RevId: </a><a class="double-link" href="https://github.com/Sqooba/scala-timeseries-lib/commit/3123c92f4">3123c92f4</a>

Run Details

9 of 9 new or added lines in 3 files covered. (100.0%)

961 of 1065 relevant lines covered (90.23%)

1.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.55

/src/main/scala/io/sqooba/oss/timeseries/window/WindowSlider.scala

package io.sqooba.oss.timeseries.window

import io.sqooba.oss.timeseries.TimeSeries
import io.sqooba.oss.timeseries.immutable.TSEntry

import scala.collection.immutable.Queue

object WindowSlider {

  type Window[T] = TSEntry[Queue[TSEntry[T]]]

  /** See [[WindowSlider.window[T, A]()]]
    *
    * @param in the entries over which to slide a window
    * @param windowWidth width of the window
    * @return a stream of entries representing the content of the window in a time-interval
    */
  def window[T](
      in: Stream[TSEntry[T]],
      windowWidth: Long
  ): Stream[Window[T]] =
    window(
      in,
      windowWidth,
      DoNothingAggregator.asInstanceOf[TimeUnawareReversibleAggregator[T, Nothing]]
    ).map(_._1)

  /** Slides a window of size 'windowWidth' on the entries present in 'in'. And
    * calculate some aggregate that does not depend on the time of validity of the
    * entries.
    *
    * Each returned entry E contains the entries of the original time series that
    * intersect with any window that ends in the domain of E. The returned stream can
    * be seen as a time series that, when queried for time 't', answers the question
    * "All the entries that have a domain that is at least partly contained in 't -
    * window' and 't', along with the aggregated value for those entries.
    *
    * @param in the entries over which to slide a window
    * @param windowWidth width of the window
    * @param aggregator a reversible aggregator to efficiently compute aggregations over the window
    * @return a stream of entries representing the content of the window in a time-interval, along with
    *         the aggregated value
    */
  def window[T, A](
      in: Stream[TSEntry[T]],
      windowWidth: Long,
      aggregator: TimeUnawareReversibleAggregator[T, A]
  ): Stream[(Window[T], Option[A])] = {
    require(windowWidth > 0, "Needs a strictly positive window size")

    if (in.isEmpty) Stream.empty
    else {
      windowRec(
        in,
        Queue.empty,
        in.head.timestamp,
        windowWidth,
        aggregator
      )
    }
  }

  /** See [[window()]] above. This function slides a window and uses a time-aware
    * aggregator. Therefore it samples the entries first.
    *
    * @param sampleRate to resample the entries
    * @param useClosestInWindow whether to sample strictly or not (see [[TimeSeries.sample()]])
    * @return a stream of entries representing the content of the window in a time-interval, along with
    *         the aggregated value
    */
  def window[T, A](
      in: Stream[TSEntry[T]],
      windowWidth: Long,
      aggregator: TimeAwareReversibleAggregator[T, A],
      sampleRate: Long,
      useClosestInWindow: Boolean = true
  ): Stream[(Window[T], Option[A])] = {
    require(windowWidth > 0, "Needs a strictly positive window size")

    if (in.isEmpty) Stream.empty
    else {
      windowRec(
        TimeSeries.sample(in, in.head.timestamp, sampleRate, useClosestInWindow).toStream,
        Queue.empty,
        in.head.timestamp,
        windowWidth,
        aggregator
      )
    }
  }

  /**
    * Rough outline of the way the windowing works:
    *   - initialise the recursion with the entire remaining stream,
    *     an empty 'previous' queue and the 'cursor' pointing
    *     to the first entry
    *
    *   - determine whether we need to:
    *     - add an entry to the window content, and/or
    *     - remove an entry from the window content
    *   - determine:
    *     - the time until the next addition, if any
    *     - the time until the next removal
    *   (If the current window is empty, the next removal may be the value that will be added.)
    *   - the next cursor will be the minimum between the two values above
    *   - recurse
    *
    * @param remaining the entries over which the window still needs to pass
    * @param previousEntryContent the content of the window at the last iteration
    * @param timeCursor points to 'where we are' on the timeline'
    * @param windowWidth width of the window
    * @return a stream of entries that contain the content of the window for their domain of definition.
    */
  private def windowRec[T, A](
      remaining: Stream[TSEntry[T]],
      previousEntryContent: Queue[TSEntry[T]],
      timeCursor: Long,
      windowWidth: Long,
      aggregator: ReversibleAggregator[T, A]
  ): Stream[(Window[T], Option[A])] = {

    val (fromRemaining, dropFromWindow, advance) =
      whatToUpdate(
        remaining,
        previousEntryContent,
        timeCursor,
        windowWidth
      )

    if (advance == 0) {
      // This signals the end of the recursion
      return Stream.empty
    }

    val nextCursor  = timeCursor + advance
    val newValidity = nextCursor - timeCursor

    (fromRemaining, dropFromWindow) match {
      case (true, true) =>
        aggregator.addAndDrop(remaining.head, previousEntryContent)
        val newWindow = TSEntry(
          timeCursor,
          previousEntryContent.tail :+ remaining.head,
          newValidity
        )
        (newWindow, aggregator.currentValue) #::
          windowRec(remaining.tail, newWindow.value, nextCursor, windowWidth, aggregator)
      case (true, false) =>
        aggregator.addEntry(remaining.head, previousEntryContent)
        val newWindow = TSEntry(
          timeCursor,
          previousEntryContent :+ remaining.head,
          newValidity
        )
        (newWindow, aggregator.currentValue) #::
          windowRec(remaining.tail, newWindow.value, nextCursor, windowWidth, aggregator)
      case (false, true) =>
        aggregator.dropHead(previousEntryContent)
        val newWindow = TSEntry(
          timeCursor,
          previousEntryContent.tail,
          newValidity
        )
        (newWindow, aggregator.currentValue) #::
          windowRec(remaining, newWindow.value, nextCursor, windowWidth, aggregator)
      case _ =>
        throw new IllegalStateException(
          "Something went very wrong. Please file a bug report. Would you fancy a cup of tea while we fix this?"
        )
    }

  }

  /**
    * Tells us if we can take from the remaining, if we need to remove something from the window, or both,
    * and by how much we can advance the cursor.
    *
    * The cursor should always be located either on the timestamp of the next entry to be added,
    * or on (e_definedUntil + windowWidth) with 'e' being the next entry to be removed.
    * For the last iteration, the cursor will be set to the end of the domain of the last entry of
    * the series being windowed.
    *
    * Notes:
    *  - windows are begin-inclusive and end-exclusive: [begin, end[
    *  - once there are no more remaining entries, and the cursor has reached the end of the last added entry,
    *    'advance-by' will always be 0. This is because we do not slide the window into the 'future' beyond
    *    the last entry's domain of definition.
    *
    * @param remaining      the stream of remaining entries over which to slide a window
    * @param previousBucket the content of the previous bucket
    * @param timeCursor         the location of the current cursor on the timeline
    * @param windowWidth    width of the sliding window.
    * @return a tuple ('take-from-remaining', 'remove-from-window', 'advance-by')
    */
  private[timeseries] def whatToUpdate(
      remaining: Stream[TSEntry[_]],
      previousBucket: Queue[TSEntry[_]],
      timeCursor: Long,
      windowWidth: Long
  ): (Boolean, Boolean, Long) = {

    // Using an assert, as this triggering would indicate a bug in the recursion logic.
    // TODO: use relevant build/run flags to not have asserts at runtime ;)
    assert(remaining.nonEmpty || previousBucket.nonEmpty, "Can't accept empty remaining and empty previous window.")
    assert(
      remaining.headOption
        .fold(false)(e => e.timestamp == timeCursor)
        || previousBucket.headOption.fold(false)(_.definedUntil == timeCursor - windowWidth)
        || previousBucket.lastOption.fold(false)(_.definedUntil == timeCursor),
      "cursor MUST be sitting on the timestamp of the first remaining," +
          "the end of validity of the first element in the queue, or the end of validity of the last entry in the queue"
    )

    // Cursor reached the end of the domain of definition of the timeseries.
    if (remaining.isEmpty && previousBucket.last.definedUntil == timeCursor) {
      // See notes in doc. Currently we don't want to slide the window further
      // when there are no more to be added.
      return (false, false, 0)
    }

    // Determine what kind of updates we need to do to the window content
    val takeFromRemaining =
      remaining.headOption.exists(_.timestamp - timeCursor == 0)

    // We need to remove an entry from the elements in the sliding window once their 'definedUntil' is about to be
    // equal to the window tail
    val removeFromWindow =
      previousBucket.headOption.exists(_.definedUntil - (timeCursor - windowWidth) == 0)

    // Determine by how much we can advance the window:
    // (what we must add to the cursor so it is equal to the next addition's timestamp)
    val (spaceUntilNextAddition) = remaining match {
      case head +: tail =>
        if (takeFromRemaining) {
          // If we are taking from the remaining now, the next one to be added is the first of the remaining entries.
          // If no more entries are remaining, the cursor should only be advanced to the end of the last entry's domain.
          // (This allows the termination condition to trigger, and ensure we don't slide the window beyond the domain
          // of definition of the input.)
          tail.headOption.fold(head.validity)(_.timestamp - timeCursor)
        } else {
          head.timestamp - timeCursor
        }
      case _ => Long.MaxValue // Nothing to add until infinity...
    }

    // Time until the next removal, from the present cursor
    // (what we must add to the cursor so (cursor - windowWidth) is equal to the next removal's 'definedUntil')
    val spaceUntilNextRemoval = if (removeFromWindow) {
      previousBucket.tail.headOption
        .orElse(remaining.headOption)
        .fold(Long.MaxValue)(_.definedUntil) - (timeCursor - windowWidth)
    } else {
      previousBucket.headOption
        .fold(remaining.head.definedUntil)(_.definedUntil) - (timeCursor - windowWidth)
    }

    val advance = Math.min(spaceUntilNextAddition, spaceUntilNextRemoval)

    // Kinda cumbersome way to check that we don't advance the cursor beyond the last entry's validity after
    // we added the last element to it but still need to remove previous entries while we approach the end of the series.
    val nextAdvance = if (remaining.nonEmpty) {
      advance
    } else {
      Math.min(advance, previousBucket.last.definedUntil - timeCursor)
    }

    assert(takeFromRemaining || removeFromWindow, "Looks like a bug, mate...")
    (takeFromRemaining, removeFromWindow, nextAdvance)
  }

  /** Slides windows over the given TSEntries that are dynamically defined by a start
    * and a stop condition. Once the start condition is true on an entry, the window
    * extends from this entry up to but not including the entry where the stop
    * condition returns true. (If both conditions are true, no window is started.)
    *
    * The returned entries contain the value of the aggregator after the last entry
    * of the window has been added. Their validity is given by the loose domain of
    * the entries in the window. A new aggregator is taken for each window.
    *
    * @param entries over which to slide windows
    * @param start defines the start of the windows
    * @param stop defines the end of the windows
    * @param aggregator a by-name aggregator to use for each window
    * @return the value of the aggregator for each window
    */
  def dynamicWindow[T, A](
      entries: Stream[TSEntry[T]],
      start: TSEntry[T] => Boolean,
      stop: TSEntry[T] => Boolean,
      aggregator: => Aggregator[T, A]
  ): Stream[TSEntry[A]] =
    // To start a window, the starting condition needs to be true and the stopping
    // condition needs to be false: dropWhile takes the negation of that:
    // !(start && !stop) === !start || stop
    entries.dropWhile(e => !start(e) || stop(e)) match {
      // Nothing to do
      case Stream() => Stream()

      case started =>
        // Window contains all entries _before_ stop is true.
        val (window, remaining) = started.span(!stop(_))
        // As stop is true on the first element, the window is guaranteed to have at
        // least one entry.

        // Add each entry of the window to the aggregator. Because the aggregator
        // also takes the window as an argument, we fold that as well.
        val (finalAgg, _) = window.foldLeft((aggregator, Queue.empty[TSEntry[T]])) {
          case ((agg, prevWindow), entry) =>
            agg.addEntry(entry, prevWindow)
            (agg, prevWindow :+ entry)
        }

        val aggregatedEntry = finalAgg.currentValue.map(
          TSEntry(window.head.timestamp, _, window.last.definedUntil - window.head.timestamp)
        )

        // lazily evaluate the rest of the stream
        aggregatedEntry.toStream #::: dynamicWindow(remaining, start, stop, aggregator)
    }

}

1	package io.sqooba.oss.timeseries.window
2
3	import io.sqooba.oss.timeseries.TimeSeries
4	import io.sqooba.oss.timeseries.immutable.TSEntry
5
6	import scala.collection.immutable.Queue
7
8	object WindowSlider {
9
10	type Window[T] = TSEntry[Queue[TSEntry[T]]]
11
12	/** See [[WindowSlider.window[T, A]()]]
13	*
14	* @param in the entries over which to slide a window
15	* @param windowWidth width of the window
16	* @return a stream of entries representing the content of the window in a time-interval
17	*/
18	def window[T](
19	in: Stream[TSEntry[T]],
20	windowWidth: Long
21	): Stream[Window[T]] =
22	window(
23	in,
24	windowWidth,
25	DoNothingAggregator.asInstanceOf[TimeUnawareReversibleAggregator[T, Nothing]]	2✔
26	).map(_._1)	2✔
27
28	/** Slides a window of size 'windowWidth' on the entries present in 'in'. And
29	* calculate some aggregate that does not depend on the time of validity of the
30	* entries.
31	*
32	* Each returned entry E contains the entries of the original time series that
33	* intersect with any window that ends in the domain of E. The returned stream can
34	* be seen as a time series that, when queried for time 't', answers the question
35	* "All the entries that have a domain that is at least partly contained in 't -
36	* window' and 't', along with the aggregated value for those entries.
37	*
38	* @param in the entries over which to slide a window
39	* @param windowWidth width of the window
40	* @param aggregator a reversible aggregator to efficiently compute aggregations over the window
41	* @return a stream of entries representing the content of the window in a time-interval, along with
42	* the aggregated value
43	*/
44	def window[T, A](
45	in: Stream[TSEntry[T]],
46	windowWidth: Long,
47	aggregator: TimeUnawareReversibleAggregator[T, A]
48	): Stream[(Window[T], Option[A])] = {
49	require(windowWidth > 0, "Needs a strictly positive window size")	2✔
50
51	if (in.isEmpty) Stream.empty	2✔
52	else {
53	windowRec(	2✔
54	in,
55	Queue.empty,	2✔
56	in.head.timestamp,	2✔
57	windowWidth,
58	aggregator
59	)
60	}
61	}
62
63	/** See [[window()]] above. This function slides a window and uses a time-aware
64	* aggregator. Therefore it samples the entries first.
65	*
66	* @param sampleRate to resample the entries
67	* @param useClosestInWindow whether to sample strictly or not (see [[TimeSeries.sample()]])
68	* @return a stream of entries representing the content of the window in a time-interval, along with
69	* the aggregated value
70	*/
71	def window[T, A](
72	in: Stream[TSEntry[T]],
73	windowWidth: Long,
74	aggregator: TimeAwareReversibleAggregator[T, A],
75	sampleRate: Long,
76	useClosestInWindow: Boolean = true
77	): Stream[(Window[T], Option[A])] = {
78	require(windowWidth > 0, "Needs a strictly positive window size")	2✔
79
80	if (in.isEmpty) Stream.empty	2✔
81	else {
82	windowRec(	2✔
83	TimeSeries.sample(in, in.head.timestamp, sampleRate, useClosestInWindow).toStream,	2✔
84	Queue.empty,	2✔
85	in.head.timestamp,	2✔
86	windowWidth,
87	aggregator
88	)
89	}
90	}
91
92	/**
93	* Rough outline of the way the windowing works:
94	* - initialise the recursion with the entire remaining stream,
95	* an empty 'previous' queue and the 'cursor' pointing
96	* to the first entry
97	*
98	* - determine whether we need to:
99	* - add an entry to the window content, and/or
100	* - remove an entry from the window content
101	* - determine:
102	* - the time until the next addition, if any
103	* - the time until the next removal
104	* (If the current window is empty, the next removal may be the value that will be added.)
105	* - the next cursor will be the minimum between the two values above
106	* - recurse
107	*
108	* @param remaining the entries over which the window still needs to pass
109	* @param previousEntryContent the content of the window at the last iteration
110	* @param timeCursor points to 'where we are' on the timeline'
111	* @param windowWidth width of the window
112	* @return a stream of entries that contain the content of the window for their domain of definition.
113	*/
114	private def windowRec[T, A](
115	remaining: Stream[TSEntry[T]],
116	previousEntryContent: Queue[TSEntry[T]],
117	timeCursor: Long,
118	windowWidth: Long,
119	aggregator: ReversibleAggregator[T, A]
120	): Stream[(Window[T], Option[A])] = {
121
122	val (fromRemaining, dropFromWindow, advance) =	2✔
123	whatToUpdate(
124	remaining,
125	previousEntryContent,
126	timeCursor,
127	windowWidth
128	)
129
130	if (advance == 0) {	2✔
131	// This signals the end of the recursion
132	return Stream.empty	2✔
133	}
134
135	val nextCursor = timeCursor + advance	2✔
136	val newValidity = nextCursor - timeCursor	2✔
137
138	(fromRemaining, dropFromWindow) match {
139	case (true, true) =>
140	aggregator.addAndDrop(remaining.head, previousEntryContent)	2✔
141	val newWindow = TSEntry(	2✔
142	timeCursor,
143	previousEntryContent.tail :+ remaining.head,	2✔
144	newValidity
145	)
146	(newWindow, aggregator.currentValue) #::	2✔
147	windowRec(remaining.tail, newWindow.value, nextCursor, windowWidth, aggregator)
148	case (true, false) =>
149	aggregator.addEntry(remaining.head, previousEntryContent)	2✔
150	val newWindow = TSEntry(	2✔
151	timeCursor,
152	previousEntryContent :+ remaining.head,	2✔
153	newValidity
154	)
155	(newWindow, aggregator.currentValue) #::	2✔
156	windowRec(remaining.tail, newWindow.value, nextCursor, windowWidth, aggregator)
157	case (false, true) =>
158	aggregator.dropHead(previousEntryContent)	2✔
159	val newWindow = TSEntry(	2✔
160	timeCursor,
161	previousEntryContent.tail,	2✔
162	newValidity
163	)
164	(newWindow, aggregator.currentValue) #::	2✔
165	windowRec(remaining, newWindow.value, nextCursor, windowWidth, aggregator)
166	case _ =>
167	throw new IllegalStateException(	×
168	"Something went very wrong. Please file a bug report. Would you fancy a cup of tea while we fix this?"
169	)
170	}
171
172	}
173
174	/**
175	* Tells us if we can take from the remaining, if we need to remove something from the window, or both,
176	* and by how much we can advance the cursor.
177	*
178	* The cursor should always be located either on the timestamp of the next entry to be added,
179	* or on (e_definedUntil + windowWidth) with 'e' being the next entry to be removed.
180	* For the last iteration, the cursor will be set to the end of the domain of the last entry of
181	* the series being windowed.
182	*
183	* Notes:
184	* - windows are begin-inclusive and end-exclusive: [begin, end[
185	* - once there are no more remaining entries, and the cursor has reached the end of the last added entry,
186	* 'advance-by' will always be 0. This is because we do not slide the window into the 'future' beyond
187	* the last entry's domain of definition.
188	*
189	* @param remaining the stream of remaining entries over which to slide a window
190	* @param previousBucket the content of the previous bucket
191	* @param timeCursor the location of the current cursor on the timeline
192	* @param windowWidth width of the sliding window.
193	* @return a tuple ('take-from-remaining', 'remove-from-window', 'advance-by')
194	*/
195	private[timeseries] def whatToUpdate(
196	remaining: Stream[TSEntry[_]],
197	previousBucket: Queue[TSEntry[_]],
198	timeCursor: Long,
199	windowWidth: Long
200	): (Boolean, Boolean, Long) = {
201
202	// Using an assert, as this triggering would indicate a bug in the recursion logic.
203	// TODO: use relevant build/run flags to not have asserts at runtime ;)
204	assert(remaining.nonEmpty \|\| previousBucket.nonEmpty, "Can't accept empty remaining and empty previous window.")	2✔
205	assert(	2✔
206	remaining.headOption
207	.fold(false)(e => e.timestamp == timeCursor)	2✔
208	\|\| previousBucket.headOption.fold(false)(_.definedUntil == timeCursor - windowWidth)	×
209	\|\| previousBucket.lastOption.fold(false)(_.definedUntil == timeCursor),	2✔
210	"cursor MUST be sitting on the timestamp of the first remaining," +	×
211	"the end of validity of the first element in the queue, or the end of validity of the last entry in the queue"
212	)
213
214	// Cursor reached the end of the domain of definition of the timeseries.
215	if (remaining.isEmpty && previousBucket.last.definedUntil == timeCursor) {	2✔
216	// See notes in doc. Currently we don't want to slide the window further
217	// when there are no more to be added.
218	return (false, false, 0)	2✔
219	}
220
221	// Determine what kind of updates we need to do to the window content
222	val takeFromRemaining =
223	remaining.headOption.exists(_.timestamp - timeCursor == 0)	2✔
224
225	// We need to remove an entry from the elements in the sliding window once their 'definedUntil' is about to be
226	// equal to the window tail
227	val removeFromWindow =
228	previousBucket.headOption.exists(_.definedUntil - (timeCursor - windowWidth) == 0)	2✔
229
230	// Determine by how much we can advance the window:
231	// (what we must add to the cursor so it is equal to the next addition's timestamp)
232	val (spaceUntilNextAddition) = remaining match {
233	case head +: tail =>
234	if (takeFromRemaining) {
235	// If we are taking from the remaining now, the next one to be added is the first of the remaining entries.
236	// If no more entries are remaining, the cursor should only be advanced to the end of the last entry's domain.
237	// (This allows the termination condition to trigger, and ensure we don't slide the window beyond the domain
238	// of definition of the input.)
239	tail.headOption.fold(head.validity)(_.timestamp - timeCursor)	2✔
240	} else {
241	head.timestamp - timeCursor	2✔
242	}
243	case _ => Long.MaxValue // Nothing to add until infinity...	2✔
244	}
245
246	// Time until the next removal, from the present cursor
247	// (what we must add to the cursor so (cursor - windowWidth) is equal to the next removal's 'definedUntil')
248	val spaceUntilNextRemoval = if (removeFromWindow) {
249	previousBucket.tail.headOption
250	.orElse(remaining.headOption)	2✔
251	.fold(Long.MaxValue)(_.definedUntil) - (timeCursor - windowWidth)	2✔
252	} else {
253	previousBucket.headOption
254	.fold(remaining.head.definedUntil)(_.definedUntil) - (timeCursor - windowWidth)	2✔
255	}
256
257	val advance = Math.min(spaceUntilNextAddition, spaceUntilNextRemoval)	2✔
258
259	// Kinda cumbersome way to check that we don't advance the cursor beyond the last entry's validity after
260	// we added the last element to it but still need to remove previous entries while we approach the end of the series.
261	val nextAdvance = if (remaining.nonEmpty) {	2✔
262	advance	2✔
263	} else {
264	Math.min(advance, previousBucket.last.definedUntil - timeCursor)	2✔
265	}
266
267	assert(takeFromRemaining \|\| removeFromWindow, "Looks like a bug, mate...")	×
268	(takeFromRemaining, removeFromWindow, nextAdvance)	2✔
269	}
270
271	/** Slides windows over the given TSEntries that are dynamically defined by a start
272	* and a stop condition. Once the start condition is true on an entry, the window
273	* extends from this entry up to but not including the entry where the stop
274	* condition returns true. (If both conditions are true, no window is started.)
275	*
276	* The returned entries contain the value of the aggregator after the last entry
277	* of the window has been added. Their validity is given by the loose domain of
278	* the entries in the window. A new aggregator is taken for each window.
279	*
280	* @param entries over which to slide windows
281	* @param start defines the start of the windows
282	* @param stop defines the end of the windows
283	* @param aggregator a by-name aggregator to use for each window
284	* @return the value of the aggregator for each window
285	*/
286	def dynamicWindow[T, A](
287	entries: Stream[TSEntry[T]],
288	start: TSEntry[T] => Boolean,
289	stop: TSEntry[T] => Boolean,
290	aggregator: => Aggregator[T, A]
291	): Stream[TSEntry[A]] =
292	// To start a window, the starting condition needs to be true and the stopping
293	// condition needs to be false: dropWhile takes the negation of that:
294	// !(start && !stop) === !start \|\| stop
295	entries.dropWhile(e => !start(e) \|\| stop(e)) match {	2✔
296	// Nothing to do
297	case Stream() => Stream()	2✔
298
299	case started =>
300	// Window contains all entries _before_ stop is true.
301	val (window, remaining) = started.span(!stop(_))	2✔
302	// As stop is true on the first element, the window is guaranteed to have at
303	// least one entry.
304
305	// Add each entry of the window to the aggregator. Because the aggregator
306	// also takes the window as an argument, we fold that as well.
307	val (finalAgg, _) = window.foldLeft((aggregator, Queue.empty[TSEntry[T]])) {	2✔
308	case ((agg, prevWindow), entry) =>
309	agg.addEntry(entry, prevWindow)	2✔
310	(agg, prevWindow :+ entry)	2✔
311	}
312
313	val aggregatedEntry = finalAgg.currentValue.map(	2✔
314	TSEntry(window.head.timestamp, _, window.last.definedUntil - window.head.timestamp)	2✔
315	)
316
317	// lazily evaluate the rest of the stream
318	aggregatedEntry.toStream #::: dynamicWindow(remaining, start, stop, aggregator)	2✔
319	}
320
321	}

Sqooba / scala-timeseries-lib / 105

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous