Merge pull request patr1ck-m#27 from patr1ck-m/feature/examples

Add examples for CPU usage and exponential distribution anomaly detection

Author: patr1ck-m

src/examples.effekt

@@ -0,0 +1,38 @@
+module src/examples
+
+import src/lib/cpu_utilization_input
+import src/lib/anomaly_detection
+import src/lib/anomaly_logger
+import src/lib/stream_input
+import src/lib/event_logger
+import src/lib/aggregation
+import src/lib/event
+import io/error
+import io/time
+
+def cpuUsageAnomalyDetectionExample(): Unit = {
+  with on[IOError].panic()
+  with boundary()
+  with anomaly_logger::logToConsole();
+  with anomaly_logger::logToFileCsvRethrow("cpu_anomalies.csv");
+  with minMaxAnomalyDetector(0.0, 0.25);
+  with event_logger::logToFileCsvRethrow("cpu_usage_aggregated.csv"); // Just for convenience, information is already in cpu_anomalies.csv
+  
+  with cpuUsagePullStream(100);
+  with event_logger::logToFileCsvPull("cpu_usage_raw.csv");
+  with pullStreamDelayer(100);
+  aggregateMeanWindow(10);
+}
+
+/// Example using exponential distribution based anomaly detection on an exponential distributed data stream
+def exponentialDistributionWithZScoreAnomalyDetectionExample(): Unit = {
+  with on[IOError].panic()
+  with boundary()
+  with anomaly_logger::logAnomaliesToConsole();
+  with zScoreAnomalyDetector(0.90);
+  
+  with stream_input::exponentialDistributedPull(1.0);
+  with pullStreamDelayer(50);
+  with pullStreamLimiter(100);
+  aggregateMean();
+}

src/lib/anomaly_logger.effekt

@@ -258,17 +258,9 @@ namespace examples {
       Event(6, -1.0),
       Event(7, 3.0)
     ]
-    try {
-      minMaxAnomalyDetector(0.0, 5.0)
-    } with read[Event] { () =>
-      if (lst.isEmpty()) {
-        println("No more values to read, stopping...")
-        resume { do stop() }
-      } else {
-        val first = lst.head()
-        lst = lst.deleteAt(0)
-        println("Providing value: " ++ first.value.show())
-        resume { unbox first }
+    minMaxAnomalyDetector(0.0, 5.0) {
+      lst.foreach { e =>
+        do emit(e)
       }
     }
   }
@@ -285,17 +277,9 @@ namespace examples {
       Event(6, -1.0),
       Event(7, 3.0)
     ]
-    try {
-      zScoreAnomalyDetector(1.0)
-    } with read[Event] { () =>
-      if (lst.isEmpty()) {
-        println("No more values to read, stopping...")
-        resume { do stop() }
-      } else {
-        val first = lst.head()
-        lst = lst.deleteAt(0)
-        println("Providing value: " ++ first.value.show())
-        resume { unbox first }
+    zScoreAnomalyDetector(1.0) {
+      lst.foreach { e =>
+        do emit(e)
       }
     }
   }
@@ -315,25 +299,17 @@ namespace examples {
       Event(6, -1.0),
       Event(7, 3.0)
     ]
-    try {
-      zScoreAnomalyDetector(1.0)
-    } with read[Event] { () =>
-      if (lst.isEmpty()) {
-        println("No more values to read, stopping...")
-        resume { do stop() }
-      } else {
-        val first = lst.head()
-        lst = lst.deleteAt(0)
-        println("Providing value: " ++ first.value.show())
-        resume { unbox first }
+    zScoreAnomalyDetector(1.0) {
+      lst.foreach { e =>
+        do emit(e)
       }
     }
   }
-}
 
-def main(): Unit = {
-  examples::example3()
-  ()
+  def main(): Unit = {
+    examples::example3()
+    ()
+  }
 }
 
 namespace internal {

src/lib/event_logger.effekt

@@ -27,6 +27,20 @@ def logToConsoleRethrow() {body: () => Unit / emit[Event]}: Unit / emit[Event] =
   }
 }
 
+/// An AnomalyDetection handler that logs all events to the console for pull-based streams
+def logToConsolePull() {body: () => Unit / read[Event]}: Unit / read[Event] = {
+  with boundary
+  try {
+    body()
+  } with read[Event] {
+    val e = do read[Event]()
+    resume {
+      println("Event: " ++ e.value.show() ++ " at t=" ++ e.timestamp.show())
+      e
+    }
+  }
+}
+
 /// An AnomalyDetection handler that logs all events to a CSV file
 def logToFileCsv(filePath: String) {body: () => Unit / emit[Event]}: Unit / Exception[IOError] = {
   var buffer: List[CsvRow] = [["timestamp", "value"]]
@@ -82,6 +96,39 @@ def logToFileCsvRethrow(filePath: String) {body: () => Unit / emit[Event]}: Unit
   }
 }
 
+def logToFileCsvPull(filePath: String) {body: () => Unit / read[Event]}: Unit / { read[Event], Exception[IOError] } = {
+    var buffer: List[CsvRow] = [["timestamp", "value"]]
+    var first = true
+    val maxBufferSize = 1000
+  try { // Double try is required because the read in the with block throws stop when the stream ends
+    try {
+      body()
+    } with read[Event] {
+      val e = do read[Event]()
+      resume {
+        val row: CsvRow = [
+          e.timestamp.show(),
+          e.value.show()
+        ]
+        buffer = Cons(row, buffer)
+        if (buffer.size >= maxBufferSize) {
+          // Flush buffer to file
+          internal::saveCsvRows(buffer.reverse, filePath, first)
+          buffer = []
+          first = false
+        }
+        e
+      }
+    }
+  } with stop {
+    ()
+  }
+  // Flush remaining buffer to file
+  if (not(buffer.isEmpty())) {
+    internal::saveCsvRows(buffer.reverse, filePath, first)
+  }
+}
+
 
 namespace examples {
   def example1(): Unit = {

src/lib/stream_input.effekt

@@ -16,6 +16,31 @@ def pushToPullStream[T]() {pushStr: () => Unit / { emit[T] } } {pullStr: () => U
   stream::source[T] { pushStr() } { pullStr() }
 }
 
+def pullStreamDelayer(delayMs: Int) {body: () => Unit / { read[Event] } }: Unit / { read[Event] } = {
+  try body() with read[Event] {
+    resume {
+      val e = do read[Event]()
+      time::sleep(delayMs)
+      e
+    }
+  }
+}
+
+def pullStreamLimiter(maxEvents: Int) {body: () => Unit / { read[Event] } }: Unit / { read[Event] } = {
+  var count = 0
+  try body() with read[Event] {
+    if (count >= maxEvents) {
+      resume { do stop() }
+    } else {
+      resume {
+        count = count + 1
+        val e = do read[Event]()
+        e
+      }
+    }
+  }
+}
+
 def csvFeed(path: String, columnNameValue: String, delayMs: Int): Unit / { emit[Event], Exception[IOError], Exception[WrongFormat] } = {
   readFileUTF8(path) {
     returning::scanner[Char, Unit] {

src/main.effekt

@@ -1,8 +1,8 @@
 module main // must be named same as the file!
 
-import src/lib/aggregation
+import src/examples
 
 
 def main(): Unit = {
-  examples::example2()
+  examples::exponentialDistributionWithZScoreAnomalyDetectionExample()
 }