use installable shared lib for shared code

This commit is contained in:
Marco Zeisler
2021-05-27 19:53:41 +02:00
parent 37bffd5ce8
commit 73174858d6
21 changed files with 50 additions and 15 deletions
+16
View File
@@ -0,0 +1,16 @@
from dataclasses import dataclass
from datetime import datetime
import geopy.distance
@dataclass
class DAF:
"""
"Datenaufzeichnung für automatisiertes Fahren" (DAF), see 2021_DSE_Exercise_V1.0_Final.pdf page 7
"""
vehicle_identification_number: str
gps_location: geopy.Point
velocity: float
near_crash_event: bool
timestamp: datetime
@@ -0,0 +1,92 @@
from threading import Thread
from typing import Union
import pika
class MBWrapper:
host: str
exchange_type: str
exchange_name: str
callback: callable
_logger: Union['MBWrapper', None]
_type: str = None
_connection: pika.BlockingConnection
_channel: pika.BlockingConnection.channel
def __init__(self,
host: str = 'localhost',
exchange_type: str = 'fanout',
exchange_name: str = None,
callback: callable = None,
verbose: bool = False):
assert exchange_name, 'Please define an exchange name'
if exchange_name != 'logger':
self._logger = MBWrapper(exchange_name='logger')
self._logger.setup_sender()
else:
self._logger = None
self.host = host
self.exchange_type = exchange_type
self.exchange_name = exchange_name
self.callback = callback
self.verbose = verbose
def print(self, *msg):
if self.verbose:
print(*msg)
def setup_sender(self):
assert self._type != 'receiver', 'MBWrapper is already a receiver. Use another MBWrapper.'
self._type = 'sender'
self._setup_channel()
def setup_receiver(self):
assert self._type != 'sender', 'MBWrapper is already a sender. Use another MBWrapper.'
assert self.callback, \
'Please setup MBWrapper with "on response" self.callback which can handle a byte string as input.'
def consumer():
self._setup_channel()
result = self._channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
self._channel.queue_bind(exchange=self.exchange_name, queue=queue_name)
self.print('[{}] Waiting for messages.'.format(self.exchange_name))
self._channel.basic_consume(
queue=queue_name, on_message_callback=self._receive, auto_ack=True)
self._channel.start_consuming()
Thread(target=consumer).start()
def send(self, message: bytes):
if type(message) is not bytes:
message = str(message).encode()
self._channel.basic_publish(exchange=self.exchange_name, routing_key='', body=message)
self.print("[{}] Sent {}".format(self.exchange_name, message))
if self.__getattribute__('_logger'):
# print('logging')
self._logger.send(message)
def close(self):
self._connection.close()
def _setup_channel(self):
self._connection = pika.BlockingConnection(
pika.ConnectionParameters(host=self.host))
self._channel = self._connection.channel()
self._channel.exchange_declare(exchange=self.exchange_name, exchange_type=self.exchange_type)
def _receive(self, ch, method, properties, body):
"""
Reduce complexity, only forward the message body to the callback method
"""
self.callback(body)
@@ -0,0 +1,7 @@
from dataclasses import dataclass
@dataclass
class TargetVelocity:
vin: str
target_velocity: float
@@ -0,0 +1,6 @@
from enum import Enum
class TrafficLightColor(Enum):
GREEN = 0
RED = 1
@@ -0,0 +1,11 @@
from dataclasses import dataclass
from datetime import datetime
from dse_shared_libs.traffic_light_color import TrafficLightColor
@dataclass
class TrafficLightState:
tlid: str
color: TrafficLightColor
last_switch: datetime