Timeline

timeline.scala

import collection.immutable.{TreeMap, SortedMap}
import collection.SortedMapLike
import org.specs2.mutable._
import org.specs2.ScalaCheck
import scalaz._
import Scalaz._

object Timeline {
  def apply[A, B](l: Traversable[(B, A)])(implicit m: Manifest[B]) = m.toString match {
    case "Int"  => new Timeline(l.map(x => x._1.asInstanceOf[Int].toLong -> x._2).toMap)
    case "Long" => new Timeline(l.map(x => x._1.asInstanceOf[Long] -> x._2).toMap)
  }
  def apply[A](m: Map[Long, A]) = new Timeline(m)
}

class Timeline[A](t: Map[Long, A]) extends SortedMap[Long, A] /* with SortedMapLike[Long, A, Timeline[A]] */ {
  private[this] lazy val iMap = TreeMap(t.toArray: _*)(Ordering.fromLessThan[Long](_ > _))

  def -(key: Long) = Timeline(iMap - key)
  def get(key: Long) = iMap.get(key)
  def rangeImpl(from: Option[Long], until: Option[Long]) = Timeline(iMap.rangeImpl(from, until))
  def iterator = iMap.iterator
  def ordering = iMap.ordering
  override def empty = Timeline[A](Map())

  def normalize[B](proj: Long => Long, agg: IndexedSeq[A] => B) =
    Timeline(iMap groupBy { _._1 |> proj } mapValues { _.values.toIndexedSeq |> agg })

  def eq[B](t: Timeline[B])    = matches(t)(identity, _.head)(_ == _)
  def relEq[B](t: Timeline[B]) = relMatches(t)(identity, _.head)(_ == _)

  def matches[B, C](t: Timeline[C])(proj: Long => Long, agg: IndexedSeq[A] => B)(f: (B, C) => Boolean) = {
    val x = this.normalize(proj, agg)
    t.forall(kvp => f(x(kvp._1), kvp._2))
  }

  def makeRelativeTo[B](proj: Long => Long, t: Timeline[B]) = {
    val timePoint = proj(t.head._1)
    Timeline(this.map(p => timePoint + p._1 -> p._2))
  }

  def relMatches[B, C](t: Timeline[C])(proj: Long => Long, agg: IndexedSeq[A] => B)(f: (B, C) => Boolean) =
    matches(t.makeRelativeTo(proj, this))(proj, agg)(f)

  def intersects[B, C](t: Timeline[C])(proj: Long => Long, agg: IndexedSeq[A] => B)(f: (B, C) => Boolean) = {
    val x = this.normalize(proj, agg)
    Timeline((proj(last._1) to proj(head._1)) map { tp => tp -> t.forall(kvp => f(x(kvp._1 + tp), kvp._2)) } toMap)
  }

  def merge[B](that: Timeline[B]): Timeline[(Option[A], Option[B])] =
    Timeline((this.keySet union that.keySet) map { k => k -> (this.get(k), that.get(k)) } toMap)

  def keepValuesFor(h: Long) = Timeline(foldRight(TreeMap[Long, A]())((p, agg) => agg ++ ((0L to h) map {_ + p._1 -> p._2}).toMap))
}

class TimelineSpec extends Specification with ScalaCheck {
  "A Timeline" should {
    val t = Timeline((10, 1) :: (20, 3) :: (1010, 3) :: Nil) // time in milliseconds

    "Values can be made valid for a predefined length of time" in {
      val tt = t.normalize(_ / 1000, v => v.sum / v.length) // 0 -> 2, 1 -> 3

      tt eq Timeline((0, 2) :: (1, 3) :: Nil) must beTrue
      tt.keepValuesFor(1) eq Timeline((0, 2) :: (1, 3) :: (2, 3) :: Nil) must beTrue
    }

    "Match absolutely" in {
      t.matches(Timeline((0, 2.0) :: (1, 3.0) :: Nil))(_ / 1000, v => v.sum / v.length)(_ == _) must beTrue
      t.matches(Timeline((0, 2.0) :: (1, 3.0) :: Nil))(_ / 1000, v => v.sum / v.length)(_ != _) must beFalse
    }

    "Match relatively" in {
      t.relMatches(Timeline((0, 3.0) :: (-1, 2.0) :: Nil))(_ / 1000, v => v.sum / v.length)(_ == _)
    }

    "Create virtual timeline of odd numbers" in {
      val newT = Timeline((0 to 100).map(v => (v, v)))
      val patT = Timeline((0, (_ : Int) % 2 == 0) :: Nil)  // This is abuse, since we basically created a pattern over functions :-)
      val resT = Timeline((0 to 100).map(v => (v, v % 2 == 0)))

      newT.matches(patT)(identity, _.head)(_ |> _) must beTrue
      newT.intersects(patT)(identity, _.head)(_ |> _) eq resT must beTrue
    }

    "Compose timelines" in {
      val tA, tB = Timeline((0 to 100).map(v => (v, v)))
      val tC = Timeline((0 to 105).map(v => (v, v)))

      val rA = Timeline((0 to 100).map(v => (v, ((Some(v), Some(v))))))
      val rB = Timeline((0 to 105).map(v => (v, ((if (v <= 100) Some(v) else None, Some(v))))))

      (tA merge tB) eq rA must beTrue
      (tA merge tC) eq rB must beTrue
    }

    "Match composed timelines" in {
      val tA = Timeline((0 to 100).map(v => (v, v)))
      val tB = Timeline((1 to 101).map(v => (v, v)))

      val patX = Timeline((   0, (     None, Some(101))) ::
                          (  -1, (Some(100), Some(100))) ::
                          (-101, (Some(  0),      None)) :: Nil)

      val patA = Timeline((-1, 100) :: (-101, 0) :: Nil)
      val patB = Timeline(( 0, 101) :: (-1, 100) :: Nil)

      (tA merge tB) relEq patX must beTrue
      (tA merge tB) relEq (patA merge patB) must beTrue
    }
    
    // Takes 1 second on my machine to normalize and match a 100k timeline
    // If we store 1 measurement per minute, that would be equivalent to
    // match with a pattern that would spawn through 2.5 months of historic data
    "Test long timeline performance" in {
      val len   = 100000
      val slice = 10

      val newT = Timeline((0 to len).map(v => (v, v)))
      val normalizedT = Timeline((0 to len/slice).zip((0 to len).sliding(slice, slice).map(vs => vs.sum / vs.length).toList))
      
      newT.matches(normalizedT)(_ / slice, v => v.sum / v.length)(_ == _)
    }
  }
}