Finalize project

3_Codeabgabe/components/event_store/redis/Dockerfile

@@ -0,0 +1,4 @@
+FROM redis
+ENV PORT 6379
+ENV HOSTNAME redis
+EXPOSE 6379

3_Codeabgabe/components/event_store/service/Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.8-slim
+COPY . /app
+WORKDIR /app
+RUN pip install -r requirements.txt
+RUN cd deps; ls; pip install dse-shared-libs-1.tar.gz;
+ENV PORT 5001
+EXPOSE 5001
+ENTRYPOINT [ "python" ]
+CMD [ "event_store_server.py" ]

3_Codeabgabe/components/event_store/service/event_logger.py

@@ -0,0 +1,106 @@
+import json
+import pickle
+import sys
+
+from dse_shared_libs import daf, traffic_light_state, traffic_light_color, target_velocity
+from dse_shared_libs.daf import DAF
+from dse_shared_libs.message_broker_wrapper import MBWrapper
+from dse_shared_libs.target_velocity import TargetVelocity
+from dse_shared_libs.traffic_light_state import TrafficLightState
+from redis import StrictRedis
+
+sys.modules['daf'] = daf
+sys.modules['traffic_light_state'] = traffic_light_state
+sys.modules['traffic_light_color'] = traffic_light_color
+sys.modules['target_velocity'] = target_velocity
+
+
+class EventLogger:
+    def __init__(self, redis: StrictRedis, log_to_redis: bool, verbose: bool):
+        """
+        :param redis:
+        :param log_to_redis:
+        :param verbose:
+        """
+        self.redis = redis
+        self.log_to_redis = log_to_redis
+        self.verbose = verbose
+
+    def setup_msq_queue(self):
+        """
+        Sets up the message queue which calls self.log on msg receive.
+        """
+        mb = MBWrapper(exchange_name='logger', callback=self.log)
+        mb.setup_receiver()
+
+    def get_keys(self):
+        """
+        :returns: the available redis keys.
+        """
+        return json.dumps([key.decode() for key in self.redis.keys()])
+
+    def get_list_of(self, key: str):
+        """
+        Returns a list of all elements inside a key. Most recent elements are stored first.
+
+        :param key: DAF:<VIN> or TV:<VIN> for either DAF events or TV events
+        :returns: list of elements
+        """
+        key = self.redis.lrange(key, 0, -1)
+        return key.decode() if key else json.dumps("")
+
+    def get_index_of(self, key: str, index: int):
+        """
+        Returns element matching index in key. Most recent elements are stored first.
+
+        :param key: DAF:<VIN> or TV:<VIN> for either DAF events or TV events
+        :param index: 0 for most recent
+        """
+        key = self.redis.lindex(key, index)
+        return key.decode() if key else json.dumps("")
+
+    def log(self, msg: bytes):
+        """
+        Logging callback. Logs the msg depending on its type to the respective redis key.
+        Types can be DAF, TrafficLightState or TargetVelocity. If type is unknown, message is still logged as key
+        UNKNOWN.
+
+        :param bytes msg: pickled msg binary to log
+        """
+        key, to_log = self._unpack_message_to_log(msg)
+
+        if self.log_to_redis:
+            self.redis.lpush(key, "{}".format(to_log))
+        if self.verbose:
+            print(to_log)
+
+    def _unpack_message_to_log(self, msg: bytes):
+        try:
+            msg = pickle.loads(msg)
+        except Exception as e:
+            print("Could not unpickle [{}]: {}", msg, e)
+            self.redis: StrictRedis
+            self.redis.append('ERRORS', str(e))
+
+        key: str
+        if type(msg) == DAF:
+            msg: DAF
+            key = 'DAF:{}'.format(msg.vehicle_identification_number)
+        elif type(msg) == TrafficLightState:
+            msg: TrafficLightState
+            key = 'TL:{}'.format(msg.tlid)
+        elif type(msg) == TargetVelocity:
+            msg: TargetVelocity
+            key = 'TV:{}'.format(msg.vin)
+        else:
+            key = 'UNKNOWN'
+
+        try:
+            to_log = json.dumps(msg.to_dict())
+        except AttributeError:
+            try:
+                to_log = json.dumps(msg)
+            except (json.decoder.JSONDecodeError, TypeError):
+                to_log = str(msg)
+
+        return key, to_log

3_Codeabgabe/components/event_store/service/event_store_server.py

@@ -0,0 +1,68 @@
+import threading
+
+from flask import Flask
+from flask_redis import Redis
+from redis import StrictRedis
+
+from event_logger import EventLogger
+
+# TOGGLE if every received message shall be redirected to redis
+LOG_TO_REDIS = True
+# TOGGLE if every received message shall be printed to console
+VERBOSE = True
+
+# make sure redis (container) is running and accessible
+
+# see https://flask-and-redis.readthedocs.io/en/latest/
+# see https://github.com/andymccurdy/redis-py
+
+app = Flask(__name__)
+app.config["REDIS_HOST"] = "redis"
+app.config["REDIS_PORT"] = 6379
+redis: StrictRedis = Redis(app)
+
+
+@app.route("/")
+def get_keys():
+    keys = redis.keys()
+    keys.sort()
+    return 'Existing keys:<br><br>{}'.format(
+        ['<a href="/api/keys/{0}/">{0}</a>'.format(key.decode()) for key in keys]) + \
+           'DAF: DatenaufzeichnungAutonomesFahren, TL: TrafficLight, TV: TargetVelocity' \
+           '<br><br>' \
+           '<br><br>Access Api via <br><br>' \
+           'GET /api/keys<br>' \
+           'GET /api/keys/&lt;key&gt;<br>'
+
+
+@app.route("/api/keys")
+def api_get_keys():
+    """
+    :return: list of available redis keys
+    """
+    return el.get_keys()
+
+
+@app.route("/api/keys/<key>/")
+def api_get_of_key(key):
+    """
+    :param key: key to get data of
+    :return: list of data to the respective key
+    """
+    return el.get_list_of(key)
+
+
+@app.route("/api/keys/<key>/<index>/")
+def api_get_index_of_key(key, index):
+    """
+    :param key: key to get data element of
+    :param index: index of keys data list
+    :return: element matching index of key
+    """
+    return el.get_index_of(key, index)
+
+
+if __name__ == '__main__':
+    el = EventLogger(redis, LOG_TO_REDIS, VERBOSE)
+    threading.Thread(target=el.setup_msq_queue).start()
+    threading.Thread(target=app.run, args=('0.0.0.0', 5001)).start()

3_Codeabgabe/components/event_store/service/requirements.txt

@@ -0,0 +1,2 @@
+flask
+Flask-And-Redis

3_Codeabgabe/components/event_store/service/test_event_logger.py

@@ -0,0 +1,68 @@
+import datetime
+import json
+import pickle
+import unittest
+
+import geopy
+from dse_shared_libs.daf import DAF
+from dse_shared_libs.target_velocity import TargetVelocity
+from dse_shared_libs.traffic_light_color import TrafficLightColor
+from dse_shared_libs.traffic_light_state import TrafficLightState
+from redis import StrictRedis
+
+from event_logger import EventLogger
+
+
+class TestEventLogger(unittest.TestCase):
+    def setUp(self) -> None:
+        self.el = EventLogger(StrictRedis(), False, False)
+        self.timestamp = datetime.datetime.utcnow()
+
+    def test_unpack_daf(self):
+        daf = DAF(vehicle_identification_number='my_vin',
+                  gps_location=geopy.Point(0, 0, 0),
+                  velocity=130.0,
+                  near_crash_event=False,
+                  timestamp=self.timestamp,
+                  )
+
+        key, message = self.el._unpack_message_to_log(pickle.dumps(daf))
+
+        self.assertEqual(key, 'DAF:my_vin')
+        self.assertEqual(message, json.dumps(daf.to_dict()))
+
+    def test_unpack_target_velocity(self):
+        tv = TargetVelocity(vin='my_other_vin',
+                            target_velocity=120.0,
+                            timestamp=self.timestamp)
+
+        key, message = self.el._unpack_message_to_log(pickle.dumps(tv))
+
+        self.assertEqual(key, 'TV:my_other_vin')
+        self.assertEqual(message, json.dumps(tv.to_dict()))
+
+    def test_unpack_traffic_light_state(self):
+        tls = TrafficLightState(tlid='my_traffic_light',
+                                color=TrafficLightColor.GREEN,
+                                last_switch=self.timestamp)
+
+        key, message = self.el._unpack_message_to_log(pickle.dumps(tls))
+
+        self.assertEqual(key, 'TL:my_traffic_light')
+        self.assertEqual(message, json.dumps(tls.to_dict()))
+
+    def test_unpack_unknown_dict(self):
+        unknown = dict(foo='foo', bar='bar')
+
+        key, message = self.el._unpack_message_to_log(pickle.dumps(unknown))
+
+        self.assertEqual(key, 'UNKNOWN')
+        self.assertEqual(message, json.dumps(unknown))
+
+    def test_unpack_unknown_object(self):
+        obj = datetime.datetime.utcnow()
+
+        key, message = self.el._unpack_message_to_log(pickle.dumps(obj))
+
+        self.assertEqual(key, 'UNKNOWN')
+        self.assertEqual(message, str(obj))