Sqooba / scala-timeseries-lib / 104

pending completion

Build # 104

Build Type

push

travis-ci-com

Committed by

Shastick

Commit Message

<a href="https://github.com/Sqooba/scala-timeseries-lib/commit/b57d19bcc">Move Entities from trip to timeseries

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

Run Details

22 of 22 new or added lines in 5 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

78.26

/src/main/scala/io/sqooba/oss/timeseries/zio/ZEntryFitter.scala

package io.sqooba.oss.timeseries.zio

import io.sqooba.oss.timeseries.immutable.TSEntry
import io.sqooba.oss.timeseries.validation.TimestampValidator
import zio.{Ref, Task, UIO, ZIO}

/**
  * This encapsulates the logic of fitting TSEntries one after another. It is used to take
  * care of trimming, compressing and checking the sanity of consecutive entries.
  *
  * @param compress Whether consecutive entries of equal value should be compressed into one or not.
  */
class ZEntryFitter[T](compress: Boolean,
                      // Contains the last added entry: we need to keep it around
                      // as it may be subject to trimming or extension
                      lastAddedRef: Ref[Option[TSEntry[T]]],
                      isDomainContRef: Ref[Boolean]) {

  /**
    * Trim or extend the previous entry based on the passed entry and depending
    * on the compress flag.
    *
    * @return a finalized entry
    */
  def addAndFitLast(addMe: TSEntry[T]): Task[Option[TSEntry[T]]] = {
    for {
      previouslyAdded <- lastAddedRef.get
      previouslyContinuous <- isDomainContRef.get
      (newlyAdded, fitted, continuous) <- trimOrExpand(previouslyAdded, addMe)
      _ <- lastAddedRef.set(newlyAdded) *>
        isDomainContRef.set(previouslyContinuous && continuous)
    } yield fitted
  }

  private def trimOrExpand(last: Option[TSEntry[T]], added: TSEntry[T])
  // Return a triple of: 'last added entry', 'fitted entry', 'domain continuous
  // (Or a failed task if the added entry is invalid)
    : Task[(Option[TSEntry[T]], Option[TSEntry[T]], Boolean)] = {
    last match {
      // First Entry: save it and return no fitted entry
      case None => ZIO.succeed((Some(added), None, true))

      // A previous entry exists: attempt to append the new one
      case Some(last) =>
        if (last.timestamp >= added.timestamp) {
          ZIO.fail(
            new IllegalArgumentException(
              s"The timestamps need to be strictly increasing, was ${added.timestamp} before ${last.timestamp}."
            )
          )
        } else {

          @inline
          def isDomainCont = last.definedUntil >= added.timestamp

          ZIO.succeed(last.appendEntry(added, compress) match {
            // A compression occurred. Keep that entry around, return no fitted
            case Seq(compressed) => (Some(compressed), None, isDomainCont)

            // No further compression:
            // - return the finished entry
            // - keep the second around
            case Seq(finished, second) =>
              (Some(second), Some(finished), isDomainCont)
          })
        }
    }
  }

  private def validateTimestamps(lastTs: Long, currentTs: Long) = {
    if (lastTs < currentTs) {
      ZIO.succeed(())
    } else {
      ZIO.fail(
        new IllegalArgumentException(
          s"The timestamps need to be strictly increasing, was $lastTs before $currentTs."
        )
      )
    }
  }

  /** @return the last entry that was added to the fitter. This entry can still change
    *         if more entries are added (it might be compressed/trimmed).
    */
  def lastEntry: UIO[Option[TSEntry[T]]] = lastAddedRef.get

  /** @return whether all added entries so far were either contiguous or overlapping.
    *         I.e. there were no holes in the domain of definition of the entries seen so far.
    */
  def isDomainContinuous: UIO[Boolean] = isDomainContRef.get

}

object ZEntryFitter {

  def init[T](compress: Boolean): UIO[ZEntryFitter[T]] =
    for {
      initPrevious <- Ref.make[Option[TSEntry[T]]](None)
      continuous <- Ref.make(true)
    } yield new ZEntryFitter[T](compress, initPrevious, continuous)

}

1	package io.sqooba.oss.timeseries.zio
2
3	import io.sqooba.oss.timeseries.immutable.TSEntry
4	import io.sqooba.oss.timeseries.validation.TimestampValidator
5	import zio.{Ref, Task, UIO, ZIO}
6
7	/**
8	* This encapsulates the logic of fitting TSEntries one after another. It is used to take
9	* care of trimming, compressing and checking the sanity of consecutive entries.
10	*
11	* @param compress Whether consecutive entries of equal value should be compressed into one or not.
12	*/
13	class ZEntryFitter[T](compress: Boolean,
14	// Contains the last added entry: we need to keep it around
15	// as it may be subject to trimming or extension
16	lastAddedRef: Ref[Option[TSEntry[T]]],
17	isDomainContRef: Ref[Boolean]) {
18
19	/**
20	* Trim or extend the previous entry based on the passed entry and depending
21	* on the compress flag.
22	*
23	* @return a finalized entry
24	*/
25	def addAndFitLast(addMe: TSEntry[T]): Task[Option[TSEntry[T]]] = {
26	for {
27	previouslyAdded <- lastAddedRef.get	2✔
28	previouslyContinuous <- isDomainContRef.get	2✔
29	(newlyAdded, fitted, continuous) <- trimOrExpand(previouslyAdded, addMe)	2✔
30	_ <- lastAddedRef.set(newlyAdded) *>	2✔
31	isDomainContRef.set(previouslyContinuous && continuous)	2✔
32	} yield fitted
33	}
34
35	private def trimOrExpand(last: Option[TSEntry[T]], added: TSEntry[T])
36	// Return a triple of: 'last added entry', 'fitted entry', 'domain continuous
37	// (Or a failed task if the added entry is invalid)
38	: Task[(Option[TSEntry[T]], Option[TSEntry[T]], Boolean)] = {
39	last match {
40	// First Entry: save it and return no fitted entry
41	case None => ZIO.succeed((Some(added), None, true))	2✔
42
43	// A previous entry exists: attempt to append the new one
44	case Some(last) =>
45	if (last.timestamp >= added.timestamp) {	2✔
46	ZIO.fail(	2✔
47	new IllegalArgumentException(	2✔
48	s"The timestamps need to be strictly increasing, was ${added.timestamp} before ${last.timestamp}."
49	)
50	)
51	} else {	2✔
52
53	@inline
54	def isDomainCont = last.definedUntil >= added.timestamp	2✔
55
56	ZIO.succeed(last.appendEntry(added, compress) match {	2✔
57	// A compression occurred. Keep that entry around, return no fitted
58	case Seq(compressed) => (Some(compressed), None, isDomainCont)	2✔
59
60	// No further compression:
61	// - return the finished entry
62	// - keep the second around
63	case Seq(finished, second) =>
64	(Some(second), Some(finished), isDomainCont)	2✔
65	})
66	}
67	}
68	}
69
70	private def validateTimestamps(lastTs: Long, currentTs: Long) = {
71	if (lastTs < currentTs) {	×
72	ZIO.succeed(())	×
73	} else {
74	ZIO.fail(	×
75	new IllegalArgumentException(	×
76	s"The timestamps need to be strictly increasing, was $lastTs before $currentTs."
77	)
78	)
79	}
80	}
81
82	/** @return the last entry that was added to the fitter. This entry can still change
83	* if more entries are added (it might be compressed/trimmed).
84	*/
85	def lastEntry: UIO[Option[TSEntry[T]]] = lastAddedRef.get	2✔
86
87	/** @return whether all added entries so far were either contiguous or overlapping.
88	* I.e. there were no holes in the domain of definition of the entries seen so far.
89	*/
90	def isDomainContinuous: UIO[Boolean] = isDomainContRef.get	×
91
92	}
93
94	object ZEntryFitter {
95
96	def init[T](compress: Boolean): UIO[ZEntryFitter[T]] =
97	for {
98	initPrevious <- Ref.make[Option[TSEntry[T]]](None)	2✔
99	continuous <- Ref.make(true)	2✔
100	} yield new ZEntryFitter[T](compress, initPrevious, continuous)	2✔
101
102	}

Sqooba / scala-timeseries-lib / 104

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