Change stream input methods to use events. Fix bugs with the push to pull stream conversion method. Enhance CSV handling.

Author: patr1ck-m

src/lib/anomaly_detection.effekt

@@ -14,6 +14,12 @@ record AnomalyEvaluation (
   anomalyScore: Double
 )
 
+def show(a: AnomalyEvaluation): String = {
+  "event=" ++ a.event.show() ++
+  ", isAnomaly=" ++ a.isAnomaly.show() ++
+  ", anomalyScore=" ++ a.anomalyScore.show()
+}
+
 /// Check equality of two AnomalyEvaluation records
 ///
 /// Performs a field-by-field equality check (timestamp, value, anomaly flag, score)
@@ -122,17 +128,16 @@ namespace examples {
       } else {
         val first = lst.head()
         lst = lst.deleteAt(0)
-        println("Providing value: " ++ first.value.show())
+        println("Providing value: " ++ first.show())
         resume { unbox first }
       }
     } with AnomalyDetection { 
       def anomaly(ev: AnomalyEvaluation) = {
-        println("Anomaly detected: " ++ ev.event.value.show() ++ " at t=" ++ ev.event.timestamp.show() ++
-                " (score: " ++ ev.anomalyScore.show() ++ ")")
+        println("Anomaly detected: " ++ ev.show())
         resume(())
       }
       def noAnomaly(ev: AnomalyEvaluation) = {
-        println("No anomaly: " ++ ev.event.value.show() ++ " at t=" ++ ev.event.timestamp.show())
+        println("No anomaly: " ++ ev.show())
         resume(())
       }
     }
@@ -158,16 +163,16 @@ namespace examples {
       } else {
         val first = lst.head()
         lst = lst.deleteAt(0)
-        println("Providing value: " ++ first.value.show())
+        println("Providing value: " ++ first.show())
         resume { unbox first }
       }
     } with AnomalyDetection { 
       def anomaly(ev: AnomalyEvaluation) = {
-        println("Anomaly detected: " ++ ev.event.value.show() ++ " at t=" ++ ev.event.timestamp.show() ++ " (score: " ++ ev.anomalyScore.show() ++ ")")
+        println("Anomaly detected: " ++ ev.show())
         resume(())
       }
       def noAnomaly(ev: AnomalyEvaluation) = {
-        println("No anomaly: " ++ ev.event.value.show() ++ " at t=" ++ ev.event.timestamp.show() ++ " (score: " ++ ev.anomalyScore.show() ++ ")")
+        println("No anomaly: " ++ ev.show())
         resume(())
       }
     }

src/lib/csv.effekt

@@ -15,6 +15,7 @@ import scanner
 import stream
 import string
 import partial
+import map
 
 
 /*
@@ -48,6 +49,16 @@ def collectString{ body: => Unit / { emit[Char] } }: String = {
   str
 }
 
+def contains[T](lst: List[T], v: T) {comp: (T, T) => Bool}: Bool = {
+  var found = false
+  lst.foreach { x =>
+    if (comp(x, v)) {
+      found = true
+    }
+  }
+  return found
+}
+
 /*
 ---------------------------------------------------
 --- The following part is written by @jiribenes ---
@@ -238,6 +249,43 @@ def getColumnOf[R](columnName: String) { csv: => R / CsvBuilder }: R / {emit[Str
   }
 }
 
+/// Not in PR, added by myself
+/// 
+/// Get multiple columns by their names
+def getColumnsOf[R](columnNames: List[String]) { csv: => R / CsvBuilder }: R / {emit[Map[String, String]], Exception[WrongFormat]} = {
+  with on[MissingValue].panic() // should never happen
+  val compFuncString = box {(x: String, y: String) => if (x < y) Less() else if (x > y) Greater() else Equal()}
+  val compFuncInt = box {(x: Int, y: Int) => if (x < y) Less() else if (x > y) Greater() else Equal()}
+  var columnIdsMap = empty[Int, String](compFuncInt)
+  try csv() with CsvBuilder {
+    def rows[R]() = resume { {r} =>
+      var currentColumn = 0
+      var resultMap: Map[String, String] = empty[String, String](compFuncString)
+      val ret = try r() with RowBuilder {
+        def cell(s) = {
+          currentColumn = currentColumn + 1
+          if (not(columnIdsMap.contains(currentColumn)) and columnNames.contains(s.trim()) {(x, y) => x == y}) {
+            columnIdsMap = columnIdsMap.put(currentColumn, s.trim())
+          } else if (columnIdsMap.contains(currentColumn)) {
+            val colName = columnIdsMap.get(currentColumn).value()
+            resultMap = map::put(resultMap, colName, s.trim())
+          }
+          resume(())
+        }
+      }
+      if (not(resultMap.isEmpty())) {
+        do emit(resultMap)
+      }
+
+      // it's the first iteration and we haven't found a column yet!
+      if (columnIdsMap.size() < columnNames.size()) {
+        wrongFormat("Header doesn't have the given column name: " ++ columnNames.show())
+      }
+      ret
+    }
+  }
+}
+
 
 def main() = {
   with on[WrongFormat].report

src/lib/event.effekt

@@ -1,6 +1,41 @@
 module src/lib/event
 
+import src/lib/timestamp
+import src/lib/typeconversion
+
 record Event (
   timestamp: Int,
   value: Double
-)
+)
+
+def toEvent(value: Double): Event = {
+  Event(currentTimestampMillis(), value)
+}
+
+def toEvent(value: Double, timestamp: Int): Event = {
+  Event(timestamp, value)
+}
+
+def toEvent(value: Int): Event = {
+  Event(currentTimestampMillis(), value.toDouble)
+}
+
+def toEvent(value: Int, timestamp: Int): Event = {
+  Event(timestamp, value.toDouble)
+}
+
+def toEvent(value: String): Event / Exception[WrongFormat] = {
+  Event(currentTimestampMillis(), value.toDouble)
+}
+
+def show(e: Event): String = {
+  "Event(value=" ++ e.value.show() ++ ", timestamp=" ++ timeMillisToString(e.timestamp) ++ ")"
+}
+
+def equals(a: Event, b: Event): Bool = {
+  a.timestamp == b.timestamp && a.value == b.value
+}
+
+def equalsIgnoreTimestamp(a: Event, b: Event): Bool = {
+  a.value == b.value
+}