10 files changed, 127 insertions, 0 deletions
diff --git a/python_scripts/src/python_scripts/__init__.py b/python_scripts/src/python_scripts/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/python_scripts/src/python_scripts/__init__.py
diff --git a/python_scripts/src/python_scripts/__main__.py b/python_scripts/src/python_scripts/__main__.py
new file mode 100644
index 00000000..ecddd38f
--- /dev/null
+++ b/python_scripts/src/python_scripts/__main__.py
@@ -0,0 +1,83 @@
+import redis
+import pandas as pd
+import module2
+import requests
+from redis.exceptions import (
+    ConnectionError,
+    DataError,
+    NoScriptError,
+    RedisError,
+    ResponseError,
+)
+
+api = "http://localhost:1234"
+redis_host = "localhost"
+stream_key = "postgres_topic"
+group = "python_consumer"
+r = redis.Redis(redis_host)
+PATH = "/home/matt/src/sunfish/python_scripts/src/python_scripts/resources"
+threshold = 0
+dictionary = {}
+
+
+def caluclate_threshold(frame):
+    global threshold
+    df = frame[(frame["downtime"] > 0)]
+    threshold = df["downtime"].mean()
+    return threshold
+
+
+def create_dataframes():
+    hosts: str = "%s/host.parquet" % PATH
+    tasks: str = "%s/task.parquet" % PATH
+
+    try:
+        df_hosts = pd.read_parquet(hosts)
+        df_tasks = pd.read_parquet(tasks)
+        return (df_hosts, df_tasks)
+
+    except Exception as e:
+        print(f"exception caught: {e}")
+        exit(1)
+
+
+"""
+As suggested in https://redis.readthedocs.io/en/stable/examples/redis-stream-example.html
+"""
+toack = lambda k, g, e: r.xack(k, g, e)
+
+
+def print_xreadgroup_reply(reply, f):
+    global threshold
+    for d_stream in reply:
+        for element in d_stream[1]:
+            threshold = module2.check_metric(element[1], f, threshold, dictionary)
+            toack(d_stream[0], group, element[0])
+
+
+def read_entries(r, f):
+    while True:
+        d = r.xreadgroup(
+            groupname=group,
+            consumername="c",
+            block=5,
+            count=10,
+            streams={stream_key: ">"},
+        )
+        print_xreadgroup_reply(d, f)
+
+
+def main():
+    try:
+        frames = create_dataframes()
+        caluclate_threshold(frames[0])
+        print(threshold)
+        read_entries(r, frames[0])
+
+    except Exception as e:
+        print(dictionary)
+        print(f"exception caught: {e}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/python_scripts/src/python_scripts/__pycache__/__main__.cpython-314.pyc b/python_scripts/src/python_scripts/__pycache__/__main__.cpython-314.pyc
new file mode 100644
index 00000000..82758955
--- /dev/null
+++ b/python_scripts/src/python_scripts/__pycache__/__main__.cpython-314.pyc
diff --git a/python_scripts/src/python_scripts/__pycache__/module2.cpython-314.pyc b/python_scripts/src/python_scripts/__pycache__/module2.cpython-314.pyc
new file mode 100644
index 00000000..95473381
--- /dev/null
+++ b/python_scripts/src/python_scripts/__pycache__/module2.cpython-314.pyc
diff --git a/python_scripts/src/python_scripts/module2.py b/python_scripts/src/python_scripts/module2.py
new file mode 100644
index 00000000..8c09d965
--- /dev/null
+++ b/python_scripts/src/python_scripts/module2.py
@@ -0,0 +1,44 @@
+import pandas as pd
+import requests
+import numpy as np
+
+"""
+The threshold should be set based on a OpenDC failure model with statistical distribution.
+"""
+confidence_red = 0.95
+confidence_yellow = 0.80
+sigma = 0
+
+
+def check_metric(stream, frame, threshold, dictionary):
+    timestamp = int(stream[b"timestamp"].decode())
+    host = str(stream[b"host_id"].decode())
+    new_threshold = threshold
+    filtered_df = frame[
+        (frame["timestamp"] == timestamp) & (frame["host_name"] == host)
+    ]
+    y_t = float(stream[b"downtime"].decode())
+    y_t_hat = float(filtered_df["downtime"].iloc[0])
+    e_t = abs(y_t - y_t_hat)
+
+    if e_t > threshold * confidence_red:
+        print(f"Red,{timestamp}")
+        dictionary[str(timestamp)] = dictionary.get(str(timestamp), 0) + 1
+
+        alert = {"Name": "Alert", "Type": "Red", "Timestamp": str(timestamp)}
+        alert_dt(alert)
+        return new_threshold
+
+    if e_t > threshold * confidence_yellow:
+        print(new_threshold)
+        dictionary[str(timestamp)] = dictionary.get(str(timestamp), 0) + 1
+        print(f"Yellow,{timestamp}")
+        alert = {"Name": "Alert", "Type": "Yellow", "Timestamp": str(timestamp)}
+        alert_dt(alert)
+        return new_threshold
+
+    return new_threshold
+
+
+def alert_dt(alert):
+    requests.post("http://localhost:1234/insight", json=alert)
diff --git a/python_scripts/src/python_scripts/resources/battery.parquet b/python_scripts/src/python_scripts/resources/battery.parquet
new file mode 100644
index 00000000..93b3f21c
--- /dev/null
+++ b/python_scripts/src/python_scripts/resources/battery.parquet
diff --git a/python_scripts/src/python_scripts/resources/host.parquet b/python_scripts/src/python_scripts/resources/host.parquet
new file mode 100644
index 00000000..a99f3967
--- /dev/null
+++ b/python_scripts/src/python_scripts/resources/host.parquet
diff --git a/python_scripts/src/python_scripts/resources/powerSource.parquet b/python_scripts/src/python_scripts/resources/powerSource.parquet
new file mode 100644
index 00000000..aa5e56c0
--- /dev/null
+++ b/python_scripts/src/python_scripts/resources/powerSource.parquet
diff --git a/python_scripts/src/python_scripts/resources/service.parquet b/python_scripts/src/python_scripts/resources/service.parquet
new file mode 100644
index 00000000..694f1271
--- /dev/null
+++ b/python_scripts/src/python_scripts/resources/service.parquet
diff --git a/python_scripts/src/python_scripts/resources/task.parquet b/python_scripts/src/python_scripts/resources/task.parquet
new file mode 100644
index 00000000..fda77c58
--- /dev/null
+++ b/python_scripts/src/python_scripts/resources/task.parquet