Merge pull request patr1ck-m#8 from patr1ck-m/feature/data_input

Add stream input. Update codebase for effekt version 0.57.0

Author: patr1ck-m

flake.lock

@@ -67,7 +67,7 @@ type CsvValue = List[CsvRow]
 def build[R] { body: => R / CsvBuilder }: (R, CsvValue) = {
   var csvrows: List[CsvRow] = []
   val r: R = try body() with CsvBuilder {
-    def rows() = resume { {row} =>
+    def rows[R]() = resume { {row} =>
       val (ret, csvrow) = buildRow {row}
       csvrows = Cons(csvrow, csvrows)
       ret
@@ -122,7 +122,7 @@ def readQuotedString(): Unit / { Scan[Char], emit[Char], Exception[WrongFormat]
   }
 }
 
-def decodeRow(): Unit / {Scan[Char], RowBuilder, Exception[WrongFormat]} = exhaustively {
+def decodeRow(): Unit / {Scan[Char], RowBuilder, Exception[WrongFormat]} = loop {
   val cell = readUntil { c => c == ',' || c == '\n' || c == '"' }
   do cell(cell)
 
@@ -138,7 +138,7 @@ def decodeRow(): Unit / {Scan[Char], RowBuilder, Exception[WrongFormat]} = exhau
   }
 }
 
-def decodeCsv(): Unit / {Scan[Char], CsvBuilder, Exception[WrongFormat]} = exhaustively {
+def decodeCsv(): Unit / {Scan[Char], CsvBuilder, Exception[WrongFormat]} = loop {
   do peek[Char]() match {
     case '\n' => 
       // do skip[Char]()
@@ -158,7 +158,7 @@ def encodeCsv[R] { csv: => R / CsvBuilder }: R / emit[String] = {
   }
 
   try csv() with CsvBuilder {
-    def rows() = resume { {row} => 
+    def rows[R]() = resume { {row} => 
       sep(); encodeRow {row}
     }
   }
@@ -182,7 +182,7 @@ def validate[R] { csv: => R / CsvBuilder }: R / Exception[WrongFormat] = {
   var rowNumber = 0
 
   try csv() with CsvBuilder {
-    def rows() = resume { {r} =>
+    def rows[R]() = resume { {r} =>
       rowNumber = rowNumber + 1
 
       var cells = 0
@@ -214,7 +214,7 @@ def validate[R] { csv: => R / CsvBuilder }: R / Exception[WrongFormat] = {
 def getColumnOf[R](columnName: String) { csv: => R / CsvBuilder }: R / {emit[String], Exception[WrongFormat]} = {
   var columnId = None()
   try csv() with CsvBuilder {
-    def rows() = resume { {r} =>
+    def rows[R]() = resume { {r} =>
       var currentColumn = 0
       val ret = try r() with RowBuilder {
         def cell(s) = {

src/lib/csv.effekt

@@ -0,0 +1,117 @@
+module src/lib/stream_input
+
+import stream
+import src/lib/csv
+import queue
+import scanner
+import io/error
+import io/filesystem
+import io/time
+
+def pushToPullStream[T] {pushStr: () => Unit / { emit[T], stop } } {pullStr: () => Unit / { read[T] } } = {
+  def q = emptyQueue[T]()
+  var running = true
+  try pushStr() with emit[T] {
+    def emit(v: T) = {
+      q.pushBack(v)
+      resume(())
+    }
+  } with stop {
+    running = false
+  }
+  try pullStr() with read[T] {
+    while (running) {
+      val v = q.popFront()
+      v match {
+        case Some(value) => {resume { unbox value }}
+        case None() => {}
+      }
+    }
+    resume { do stop() }
+  }
+}
+
+/// Parse string to double (not in standard library???)
+extern def toDouble(s: String) at {}: Double =
+  js "parseFloat(${s})"
+
+def csvFeed(path: String, columnName: String, delayMs: Int): Unit / { emit[Double], Exception[IOError], Exception[WrongFormat] } = {
+  readFileUTF8(path) {
+    returning::scanner[Char, Unit] {
+      try {
+        getColumnOf(columnName) {
+          decodeCsv()
+        }
+      } with emit[String] { s =>
+        do emit(s.toDouble)
+        time::sleep(delayMs)
+        resume(())
+      }
+    }
+  }
+}
+
+def csvFeedStr(csvStr: String, columnName: String, delayMs: Int): Unit / { emit[Double], Exception[WrongFormat] } = {
+  feed(csvStr) {
+    returning::scanner[Char, Unit] {
+      try {
+        getColumnOf(columnName) {
+          decodeCsv()
+        }
+      } with emit[String] { s =>
+        do emit(s.toDouble)
+        time::sleep(delayMs)
+        resume(())
+      }
+    }
+  }
+}
+
+def csvFeedPull(path: String, columnName: String, delayMs: Int) {body: () => Unit / { read[Double], Exception[IOError], Exception[WrongFormat] } }: Unit / { Exception[IOError], Exception[WrongFormat] } = {
+  pushToPullStream[Double] { csvFeed(path, columnName, delayMs) } { body() }
+}
+
+def csvFeedStrPull(csvStr: String, columnName: String, delayMs: Int) {body: () => Unit / { read[Double], Exception[WrongFormat] } }: Unit / { Exception[WrongFormat] } = {
+  pushToPullStream[Double] { csvFeedStr(csvStr, columnName, delayMs) } { body() }
+}
+
+
+namespace examples {
+  def example1(): Unit = {
+    with on[WrongFormat].panic()
+
+    val csv = "name,age\nJohn,42\nJoe,10\nJolene,27"
+
+    try csvFeedStr(csv, "age", 500) with emit[Double] { v =>
+      println("Received value: " ++ v.show())
+      resume(())
+    }
+  }
+
+  def example2(): Unit = {
+    with on[IOError].panic()
+    with on[WrongFormat].panic()
+
+    try csvFeed("src/test/testdata.csv", "value", 500) with emit[Double] { v =>
+      println("Received value: " ++ v.show())
+      resume(())
+    }
+  }
+
+  def example3(): Unit = {
+    with on[IOError].panic()
+    with on[WrongFormat].panic()
+    with boundary
+    csvFeedPull("src/test/testdata.csv", "value", 500) { 
+      while (true) {
+        println("Waiting to read value...")
+        val v = do read[Double]()
+        println("Received value: " ++ v.show())
+      }
+    }
+  }
+
+  def main(): Unit = {
+    example2()
+  }
+}

src/lib/stream_input.effekt

@@ -4,13 +4,16 @@ import process
 import test
 import src/test/aggregation_test
 import src/test/anomaly_detection_test
+import src/test/stream_input_test
 
 def main() = {
   val testResults = [
     aggregation_test::testSuite(),
-    anomaly_detection_test::testSuite()
+    anomaly_detection_test::testSuite(),
+    // stream_input_test::testSuite()
   ]
   val allPassed = testResults.all { res => res }
   val exitCode = if (allPassed) 0 else 1
   process::exit(exitCode)
+  ()
 }

src/test.effekt

@@ -4,6 +4,7 @@ import test
 import src/lib/aggregation
 import src/lib/event
 import src/test/lib
+import process
 
 def eventsFromList(values: List[Double]): List[Event] = {
   def loop(vals: List[Double], ts: Int): List[Event] = {
@@ -99,4 +100,9 @@ def testSuite() = suite("aggregation") {
       }
     }
   }
+}
+
+def main() = {
+  process::exit(if (testSuite()) 0 else 1)
+  ()
 }

src/test/aggregation_test.effekt

@@ -4,6 +4,7 @@ import test
 import src/lib/anomaly_detection
 import src/lib/event
 import src/test/lib
+import process
 
 def eventsFromList(values: List[Double]): List[Event] = {
 	def loop(vals: List[Double], ts: Int): List[Event] = {
@@ -114,4 +115,9 @@ def testSuite() = suite("anomaly_detection") {
 			anomaly_detection::zScoreAnomalyDetector(1.0)
 		}
 	}
+}
+
+def main() = {
+  process::exit(if (testSuite()) 0 else 1)
+  ()
 }

src/test/anomaly_detection_test.effekt

@@ -15,7 +15,9 @@ def assertEmits[T](expected: List[T]) { equals: (T, T) => Bool } { program: () =
         assertTrue(equals(head, v), "Emitted value did not match expected value")
         lst = tail
     }
+    resume(())
   }
+  assertTrue(lst.isEmpty(), "Did not emit all expected values")
 }
 
 def assertNoEmits[T] { program: () => Unit / emit[T] } : Unit / { Assertion } = {
@@ -31,17 +33,40 @@ def assertReads[T](inputs: List[T]) { equals: (T, T) => Bool } { program: () =>
   var lst = inputs
   assertFalse(lst.isEmpty(), "assertReads requires a non empty list of inputs")
   val someValue: T = lst.head() // dummy value to return when we run out of inputs (not used because of the failing assert but necessary for type checking)
+  var isStop: Bool = false
   try {
     program()
   } with read[T] { () =>
-    val r: T = lst match {
+    lst match {
       case Nil() =>
-        assertTrue(false, "Read more values than expected")
-        someValue // to satisfy type checker
+        if (isStop) {
+          assertTrue(false, "Read more values than expected (stop already signaled)")
+        } else {
+          isStop = true
+          resume { do stop() }
+        }
       case Cons(head, tail) =>
         lst = tail
-        unbox head
+        resume { unbox head }
+    }
+  }
+}
+
+def assertDoRaise[E] {block: () => Unit / Exception[E]} = {
+  try {
+    block()
+    assertTrue(false, "Expected exception was not raised")
+  } with Exception[E] {
+    def raise(e: E, msg: String) = {
+    }
+  }
+}
+def assertDoNotRaise[E] {block: () => Unit / Exception[E]} = {
+  try {
+    block()
+  } with Exception[E] {
+    def raise(e: E, msg: String) = {
+      assertTrue(false, "Expected exception not to raise but was raised")
     }
-    resume { unbox r }
   }
 }

src/test/lib.effekt

@@ -0,0 +1,45 @@
+module src/test/stream_input_test
+
+import test
+import src/lib/stream_input
+import src/lib/event
+import src/test/lib
+import io/error
+import list
+import process
+import io/filesystem
+
+/// The content of the test_events.csv as a list of doubles for comparison
+def csvContentAsList(): List[Double] = {
+  [10.0, 27.0, 5.5, 42.0, 14.5, 29.0, 20.0]
+}
+
+def testSuite() = suite("stream_input_csv") {
+  test("reads csv events push") {
+    with assertDoNotRaise[WrongFormat]
+    with assertDoNotRaise[IOError]
+    assertEmits[Double](csvContentAsList()) { (a: Double, b: Double) => a == b } {
+      stream_input::csvFeed("src/test/testdata.csv", "value", 0)
+    }
+  }
+
+  // test("raises IOError on missing file") {
+  //   with assertDoRaise[IOError]
+  //   with assertDoNotRaise[WrongFormat]
+  //   assertEmits[Double]([]) { (a: Double, b: Double) => a == b } {
+  //     stream_input::csvFeed("src/test/data/missing_file.csv", "value", 0)
+  //   }
+  // }
+  
+  // test("read non existing file") {
+  //   assertDoRaise[IOError] {
+  //     readFile("non_existing_file") // should raise IOError but backend error occurs
+  //     ()
+  //   }
+  // }
+}
+
+def main() = {
+  process::exit(if (testSuite()) 0 else 1)
+  ()
+}

src/test/stream_input_test.effekt

@@ -0,0 +1,8 @@
+value
+10.0
+27.0
+5.5
+42.0
+14.5
+29.0
+20.0