AndreasKostler

package au.com.cba.omnia.eventually.ops

import au.com.cba.omnia.eventually.schema.{Entity, EntityAware}
import SelectEntitySyntax._

import com.twitter.scalding.typed.{ TypedPipe, Grouped }

import scala.reflect.ClassTag

trait GroupByEntity[C] {

AndreasKostler

/* The lag function is an analytic function that lets
   * you query more than one event at a time.
   *
   * Finds events associated with previous events of the same type.
   * For each item in the `events` pipe, find all `events` which occurred `offset` time steps
   * in the past
   *
   * @param events Input data
   * @param offset Time lag
   * @return TypedPipe of Tupels (E[t-offset],E[t])

AndreasKostler

 def lag[A: Ordering : Timestamped : EntityAware]
    (events: TypedPipe[A],
     offset: Int): TypedPipe[(A, A)] = {

    events
      .groupByEntity
      .sortBy(_.timestamp)
      .mapValueStream {
        events => {
          events.sliding(2,offset)

AndreasKostler

// These are my two event classes...
case class Event1(id: Int, eventStartTime: DateTime, entityRoles: EntityRoles)
object Event1 {
  override def toString = "ev1"
  implicit val isTimeStamped: Timestamped[Event1] =
    Timestamped.from(_.eventStartTime)
  implicit val isEntityRoleAware: EntityRoleAware[Event1] =
    EntityRoleAware.from(_.entityRoles.roles)
  implicit val hasOrdering: Ordering[Event1] =
    Ordering.by(e => (e.eventStartTime, e.entityRoles))

AndreasKostler

trait LeftFolder[C <: Coproduct, In, F] extends Serializable {
  def apply(c: C, in: In): In
}

object LeftFolder {
  def apply[C <: Coproduct, In, F](implicit folder: LeftFolder[C, In, F]) = folder

  implicit def hdLeftFolder[H, T <: Coproduct, In, F]
  (implicit f: Case2.Aux[F, In, H, In]): LeftFolder[H :+: T, In, F] = new LeftFolder[H :+: T, In, F] {
    def apply(c: H :+: T, in: In): In = c match {

AndreasKostler

import shapeless._
import syntax._
import poly._

object LeftFolderOps {

  trait LeftFolder[C <: Coproduct, In, F] extends Serializable {
    def apply(c: C, in: In): In
  }

AndreasKostler

import shapeless._

trait Test {
  def ex[InL <: HList, H, T <: HList](in : InL)(implicit isHCons: IsHCons1.Aux[InL, List[H], T])
  val ii: List[Int]::List[Int]::HNil
  ex(ii)
}

trait IsHCons1[L] { type H; type T <: HList}
object IsHCons1 {

AndreasKostler

object Abc {
  import shapeless.{ =:!=, Nat, Succ }
  import shapeless.nat.{ _0 => D0, _1 => D1, _2 => D2, _3 => D3 }
  import shapeless.ops.nat.{ Diff, GT, LTEq, ToInt }
  import shapeless.ops.hlist._

  trait Value {
    def toShortString = ???
  }
  object Value {

AndreasKostler

object Def {
  import shapeless.{ =:!=, HNil, HList, Nat, Poly1, Succ, Witness }
  import shapeless.nat.{ _0 => D0, _1 => D1, _2 => D2, _3 => D3 }
  import shapeless.ops.nat.{ Diff, GT, LTEq, ToInt }
  import shapeless.ops.hlist.{ Mapper, ToTraversable, ZipConst }

  trait Value {
    def toShortString = ???
  }
  object Value {

AndreasKostler

object Def {
  import shapeless.{ =:!=, HNil, HList, Nat, Poly1, Succ, Witness }
  import shapeless.nat.{ _0 => D0, _1 => D1, _2 => D2, _3 => D3 }
  import shapeless.ops.nat.{ Diff, GT, LTEq, ToInt }
  import shapeless.ops.hlist.{ Mapper, ToTraversable, ZipConst }

  trait Value {
    def toShortString = ???
  }
  object Value {