Move client foler

compLib/.gitignore

@@ -0,0 +1,312 @@
+# Created by https://www.toptal.com/developers/gitignore/api/macos,python,pycharm
+# Edit at https://www.toptal.com/developers/gitignore?templates=macos,python,pycharm
+
+### macOS ###
+# General
+.DS_Store
+.AppleDouble
+.LSOverride
+
+# Icon must end with two \r
+Icon
+
+
+# Thumbnails
+._*
+
+# Files that might appear in the root of a volume
+.DocumentRevisions-V100
+.fseventsd
+.Spotlight-V100
+.TemporaryItems
+.Trashes
+.VolumeIcon.icns
+.com.apple.timemachine.donotpresent
+
+# Directories potentially created on remote AFP share
+.AppleDB
+.AppleDesktop
+Network Trash Folder
+Temporary Items
+.apdisk
+
+### macOS Patch ###
+# iCloud generated files
+*.icloud
+
+### PyCharm ###
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio, WebStorm and Rider
+# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
+
+# User-specific stuff
+.idea/**/workspace.xml
+.idea/**/tasks.xml
+.idea/**/usage.statistics.xml
+.idea/**/dictionaries
+.idea/**/shelf
+
+# AWS User-specific
+.idea/**/aws.xml
+
+# Generated files
+.idea/**/contentModel.xml
+
+# Sensitive or high-churn files
+.idea/**/dataSources/
+.idea/**/dataSources.ids
+.idea/**/dataSources.local.xml
+.idea/**/sqlDataSources.xml
+.idea/**/dynamic.xml
+.idea/**/uiDesigner.xml
+.idea/**/dbnavigator.xml
+
+# Gradle
+.idea/**/gradle.xml
+.idea/**/libraries
+
+# Gradle and Maven with auto-import
+# When using Gradle or Maven with auto-import, you should exclude module files,
+# since they will be recreated, and may cause churn.  Uncomment if using
+# auto-import.
+# .idea/artifacts
+# .idea/compiler.xml
+# .idea/jarRepositories.xml
+# .idea/modules.xml
+# .idea/*.iml
+# .idea/modules
+# *.iml
+# *.ipr
+
+# CMake
+cmake-build-*/
+
+# Mongo Explorer plugin
+.idea/**/mongoSettings.xml
+
+# File-based project format
+*.iws
+
+# IntelliJ
+out/
+
+# mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# JIRA plugin
+atlassian-ide-plugin.xml
+
+# Cursive Clojure plugin
+.idea/replstate.xml
+
+# SonarLint plugin
+.idea/sonarlint/
+
+# Crashlytics plugin (for Android Studio and IntelliJ)
+com_crashlytics_export_strings.xml
+crashlytics.properties
+crashlytics-build.properties
+fabric.properties
+
+# Editor-based Rest Client
+.idea/httpRequests
+
+# Android studio 3.1+ serialized cache file
+.idea/caches/build_file_checksums.ser
+
+### PyCharm Patch ###
+# Comment Reason: https://github.com/joeblau/gitignore.io/issues/186#issuecomment-215987721
+
+# *.iml
+# modules.xml
+# .idea/misc.xml
+# *.ipr
+
+# Sonarlint plugin
+# https://plugins.jetbrains.com/plugin/7973-sonarlint
+.idea/**/sonarlint/
+
+# SonarQube Plugin
+# https://plugins.jetbrains.com/plugin/7238-sonarqube-community-plugin
+.idea/**/sonarIssues.xml
+
+# Markdown Navigator plugin
+# https://plugins.jetbrains.com/plugin/7896-markdown-navigator-enhanced
+.idea/**/markdown-navigator.xml
+.idea/**/markdown-navigator-enh.xml
+.idea/**/markdown-navigator/
+
+# Cache file creation bug
+# See https://youtrack.jetbrains.com/issue/JBR-2257
+.idea/$CACHE_FILE$
+
+# CodeStream plugin
+# https://plugins.jetbrains.com/plugin/12206-codestream
+.idea/codestream.xml
+
+# Azure Toolkit for IntelliJ plugin
+# https://plugins.jetbrains.com/plugin/8053-azure-toolkit-for-intellij
+.idea/**/azureSettings.xml
+
+### Python ###
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+
+# End of https://www.toptal.com/developers/gitignore/api/macos,python,pycharm

compLib/Api.py

@@ -0,0 +1,70 @@
+import json
+import logging
+import os
+from typing import Dict, Tuple, List
+
+import requests
+
+logger = logging.getLogger("complib-logger")
+
+API_URL = os.getenv("API_URL", "http://localhost:5000/") + "api/"
+CONF_URL = os.getenv("API_URL", "http://localhost:5000/") + "config/"
+
+api_override = os.getenv("API_FORCE", "")
+
+if api_override != "":
+    print(f"API_URL was set to {API_URL} but was overwritten with {api_override}")
+    API_URL = api_override
+
+API_URL_GET_HEU = API_URL + "getHeuballen"
+API_URL_GET_LOGISTIC_PLAN = API_URL + "getLogisticPlan"
+API_URL_GET_MATERIAL_DELIVERIES = API_URL + "getMaterialDeliveries"
+API_URL_GET_ROBOT_STATE = API_URL + "getRobotState"
+
+
+class Seeding:
+    """Klasse welche mit der Seeding API Kommuniziert.
+    """
+
+    @staticmethod
+    def get_heuballen() -> int:
+        """Macht den /api/getHeuballen request zur Seeding API.
+
+        :return: hueballencode als int.
+        :rtype: int
+        """
+        res = requests.get(API_URL_GET_HEU)
+        result = json.loads(res.content)
+        logger.debug(f"Seeding.get_heuballen = {result}, status code = {res.status_code}")
+        return result["heuballen"]
+
+    @staticmethod
+    def get_logistic_plan() -> List:
+        """Macht den /api/getLogisticPlan zur Seeding API.
+
+        :return: Liste an logistic-centern, welche vom roboter in genau der Reihenfolge beliefert werden sollten.
+        :rtype: List
+        """
+        res = requests.get(API_URL_GET_LOGISTIC_PLAN)
+        result = json.loads(res.content)
+        logger.debug(f"Seeding.get_logistic_plan = {result}, status code = {res.status_code}")
+        return result
+
+    @staticmethod
+    def get_material_deliveries() -> List:
+        """Macht den /api/getMaterialDeliveries zur Seeding API.
+
+        :return: Json Object and status code as returned by the api.
+        :rtype: List
+        """
+        res = requests.get(API_URL_GET_MATERIAL_DELIVERIES)
+        result = json.loads(res.content)
+        logger.debug(f"Seeding.get_material_deliveries = {result}, status code = {res.status_code}")
+        return result
+
+    @staticmethod
+    def get_robot_state() -> Tuple[Dict, int]:
+        res = requests.get(API_URL_GET_ROBOT_STATE)
+        result = json.loads(res.content)
+        logger.debug(f"Seeding.get_robot_state {result}, status code = {res.status_code}")
+        return result, res.status_code

compLib/Camera.py

@@ -0,0 +1,221 @@
+import sys
+from typing import Any
+
+# build image is somehow different from raspberry image? opencv-python is installed to a directory which is not in the pythonpath by default....
+sys.path.append("/usr/lib/python3.9/site-packages")
+
+import logging
+import os
+import queue
+import threading
+
+import cv2
+from flask import Flask, Response
+
+logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
+
+SERVE_VIDEO = os.getenv("SERVER_SRC", "/live")
+BUILDING_DOCS = os.getenv("BUILDING_DOCS", "false")
+
+HTML = """
+<html>
+  <head>
+    <title>Opencv Output</title>
+  </head>
+  <body>
+    <h1>Opencv Output</h1>
+    <img src="{{ VIDEO_DST }}">
+  </body>
+</html>
+"""
+
+# it would be better to use jinja2 here, but I don't want to blow up the package dependencies...
+HTML = HTML.replace("{{ VIDEO_DST }}", SERVE_VIDEO)
+
+
+class Marker:
+    def __init__(self, id: int, x: float, y: float):
+        self.id: int = id
+        self.x: float = x
+        self.y: float = y
+
+    def __str__(self) -> str:
+        return f"Marker ID: {self.id}, position: {self.x} x, {self.y} y"
+
+    def __repr__(self) -> str:
+        return str({"id": self.id, "x": self.x, "y": self.y})
+
+
+class Camera:
+    class __Webserver:
+        def __init__(self, camera):
+            self.app = Flask(__name__)
+            self.__camera = camera
+            self.__thread = threading.Thread(target=self.__start_flask, daemon=True)
+            self.__thread.start()
+
+            @self.app.route("/live")
+            def __video_feed():
+                """
+                Define route for serving jpeg stream.
+                :return: Return the response generated along with the specific media.
+                """
+                return Response(self.__camera._newest_frame_generator(),
+                                mimetype="multipart/x-mixed-replace; boundary=frame")
+
+            @self.app.route("/")
+            def __index():
+                """
+                Define route for serving a static http site to view the stream.
+                :return: Static html page
+                """
+                return HTML
+
+        def __start_flask(self):
+            """
+            Function for running flask server in a thread.
+            :return:
+            """
+            logging.getLogger("complib-logger").info("starting flask server")
+            self.app.run(host="0.0.0.0", port=9898, debug=True, threaded=True, use_reloader=False)
+
+    class __NoBufferVideoCapture:
+        def __init__(self, cam):
+            self.cap = cv2.VideoCapture(cam)
+            self.cap.set(3, 640)
+            self.cap.set(4, 480)
+            self.q = queue.Queue(maxsize=3)
+            self.stopped = False
+            self.t = threading.Thread(target=self._reader, daemon=True)
+            self.t.start()
+
+        def _reader(self):
+            while not self.stopped:
+                ret, frame = self.cap.read()
+                if not ret:
+                    continue
+                if self.q.full():
+                    try:
+                        self.q.get_nowait()
+                    except queue.Empty:
+                        pass
+                self.q.put(frame)
+
+        def read(self):
+            return self.q.get()
+
+        def stop(self):
+            self.stopped = True
+            self.t.join()
+
+    def __init__(self):
+        self.__logger = logging.getLogger("complib-logger")
+        self.__logger.info("capturing rtmp stream is disabled in this version")
+        self.__camera_stream = self.__NoBufferVideoCapture(-1)
+        self.__newest_frame = None
+        self.__lock = threading.Lock()
+        self.__webserver = self.__Webserver(self)
+
+        self.aruco_dict = cv2.aruco.Dictionary_get(cv2.aruco.DICT_6X6_50)
+        self.aruco_params = cv2.aruco.DetectorParameters_create()
+
+        self.__logger.info("Initialized vision")
+
+    def get_frame(self):
+        """
+        Die Funktion das neuste Bild, welches die Kamera aufgenommen hat zurück.
+        :return: Ein "opencv image frame"
+        """
+        img16 = self.__camera_stream.read()
+        return img16
+
+    def detect_markers(self, image):
+        """
+        Funktion um die ArUco Marker in einem Bild zu erkennen.
+        :param image: Bild, welches die Kamera aufgenommen hat.
+        :return: Gibt drei Variablen zurueck. Erstens eine Liste an Postionen der "Ecken" der erkannten Markern. Zweitens eine Liste an IDs der erkannten Markern und dritten noch Debug Informationen (diese können ignoriert werden).
+        """
+        return cv2.aruco.detectMarkers(image, self.aruco_dict, parameters=self.aruco_params)
+
+    def detect_markers_midpoint(self, image) -> tuple[list[Marker], Any]:
+        """
+        Funktion um die ArUco Marker in einem Bild zu erkennen, einzuzeichnen und den Mittelpunkt der Marker auszurechnen.
+        :param image: Bild, welches die Kamera aufgenommen hat.
+        :return: Gibt zwei Variablen zurueck. Erstens eine Liste an "Markern" und zweitens das Bild mit den eigezeichneten Marken.
+        """
+        (corners, ids, rejected) = self.detect_markers(image)
+        self.draw_markers(image, corners, ids)
+
+        res = []
+        for i in range(0, len(corners)):
+            x = sum([point[0] for point in corners[i][0]]) / 4
+            y = sum([point[1] for point in corners[i][0]]) / 4
+            res.append(Marker(ids[i][0], x, y))
+
+        return res, image
+
+    def draw_markers(self, image, corners, ids):
+        """
+        Zeichnet die erkannten Markern mit ihren IDs in das Bild.
+        :param image: Original Bild, in dem die Marker erkannt wurden.
+        :param corners: List der Positionen der Ecken der erkannten Marker.
+        :param ids: IDs der erkannten Markern.
+        :return: Neues Bild mit den eigezeichneten Markern.
+        """
+        return cv2.aruco.drawDetectedMarkers(image, corners, ids)
+
+    def publish_frame(self, image):
+        """
+        Sendet das Bild, welches der Funktion übergeben wird, an den Webserver, damit es der Nutzer in seinem Browser ansehen kann.
+        :param image: Opencv Bild, welches dem Nutzer angezeigt werden soll.
+        :return: None
+        """
+        with self.__lock:
+            if image is not None:
+                self.__newest_frame = image.copy()
+
+    def _newest_frame_generator(self):
+        """
+        Private generator which is called directly from flask server.
+        :return: Yields image/jpeg encoded frames published from publish_frame function.
+        """
+        while True:
+            # use a buffer frame to copy the newest frame with lock and then freeing it immediately
+            buffer_frame = None
+            with self.__lock:
+                if self.__newest_frame is None:
+                    continue
+
+                buffer_frame = self.__newest_frame.copy()
+
+            # encode frame for jpeg stream
+            (flag, encoded_image) = cv2.imencode(".jpg", buffer_frame)
+
+            # if there was an error try again with the next frame
+            if not flag:
+                continue
+
+            # else yield encoded frame with mimetype image/jpeg
+            yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' +
+                   bytearray(encoded_image) + b'\r\n')
+
+
+# for debugging and testing start processing frames and detecting a 6 by 9 calibration chessboard
+if __name__ == '__main__' and BUILDING_DOCS == "false":
+    camera = Camera()
+    while True:
+        image = camera.get_frame()
+
+        criteria = (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_MAX_ITER, 30, 0.001)
+
+        # processing
+        # gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        # find the chessboard corners
+        # ret, corners = cv2.findChessboardCorners(gray, (6, 9), None)
+        # cv2.drawChessboardCorners(frame, (6, 9), corners, ret)
+
+        markers, image = camera.detect_markers_midpoint(image)
+        print(markers)
+        print("-----------------")
+
+        camera.publish_frame(image)

compLib/CompLib.proto

@@ -0,0 +1,126 @@
+syntax = "proto3";
+
+package CompLib;
+
+message Header {
+  string message_type = 1;
+}
+
+message Status {
+  bool successful = 1;
+  string error_message = 2;
+}
+
+message GenericRequest {
+  Header header = 1;
+}
+
+message GenericResponse {
+  Header header = 1;
+  Status status = 2;
+}
+
+message EncoderReadPositionsRequest {
+  Header header = 1;
+}
+
+message EncoderReadPositionsResponse {
+  Header header = 1;
+  Status status = 2;
+  repeated int32 positions = 3 [packed = true];
+}
+
+message EncoderReadVelocitiesRequest {
+  Header header = 1;
+}
+
+message EncoderReadVelocitiesResponse {
+  Header header = 1;
+  Status status = 2;
+  repeated double velocities = 3 [packed = true];
+}
+
+message IRSensorsEnableRequest {
+  Header header = 1;
+}
+
+message IRSensorsDisableRequest {
+  Header header = 1;
+}
+
+message IRSensorsReadAllRequest {
+  Header header = 1;
+}
+
+message IRSensorsReadAllResponse {
+  Header header = 1;
+  Status status = 2;
+  repeated uint32 data = 3 [packed = true];
+}
+
+message MotorSetPowerRequest {
+  Header header = 1;
+  uint32 port = 2;
+  double power = 3;
+}
+
+message MotorsSetPowerRequest {
+  Header header = 1;
+  repeated MotorSetPowerRequest requests = 2;
+}
+
+message MotorSetSpeedRequest {
+  Header header = 1;
+  uint32 port = 2;
+  double speed = 3;
+}
+
+message MotorsSetSpeedRequest {
+  Header header = 1;
+  repeated MotorSetSpeedRequest requests = 2;
+}
+
+message MotorSetPulseWidthRequest {
+  Header header = 1;
+  uint32 port = 2;
+  double percent = 3;
+}
+
+message MotorsSetPulseWidthRequest {
+  Header header = 1;
+  repeated MotorSetPulseWidthRequest requests = 2;
+}
+
+message OdometryReadRequest {
+  Header header = 1;
+}
+
+message OdometryReadResponse {
+  Header header = 1;
+  Status status = 2;
+  double x_position = 3;
+  double y_position = 4;
+  double orientation = 5;
+}
+
+message DriveDistanceRequest {
+  Header header = 1;
+  double distance_m = 2;
+  double velocity_m_s = 3;
+}
+
+message TurnDegreesRequest {
+  Header header = 1;
+  double angle_degrees = 2;
+  double velocity_rad_s = 3;
+}
+
+message DriveRequest {
+  Header header = 1;
+  double linear_velocity_m_s = 2;
+  double angular_velocity_rad_s = 3;
+}
+
+message HealthUpdateRequest {
+  Header header = 1;
+}

compLib/CompLibClient.py

@@ -0,0 +1,70 @@
+import socket
+from threading import Lock
+
+import compLib.CompLib_pb2 as CompLib_pb2
+
+
+class CompLibClient(object):
+    UNIX_SOCKET_PATH = "/tmp/compLib"
+    TCP_SOCKET_HOST = ""
+    TCP_SOCKET_PORT = 9090
+    SOCKET = None
+    LOCK = Lock()
+
+    @staticmethod
+    def use_unix_socket(socket_path="/tmp/compLib"):
+        CompLibClient.UNIX_SOCKET_PATH = socket_path
+
+        CompLibClient.SOCKET = socket.socket(
+            socket.AF_UNIX, socket.SOCK_STREAM)
+        CompLibClient.SOCKET.connect(CompLibClient.UNIX_SOCKET_PATH)
+
+        from compLib.HealthCheck import HealthUpdater
+        HealthUpdater.start()
+
+    @staticmethod
+    def use_tcp_socket(socket_host, socket_port=TCP_SOCKET_PORT):
+        CompLibClient.TCP_SOCKET_HOST = socket_host
+        CompLibClient.TCP_SOCKET_PORT = socket_port
+
+        CompLibClient.SOCKET = socket.socket(
+            socket.AF_INET, socket.SOCK_STREAM)
+        CompLibClient.SOCKET.connect(
+            (CompLibClient.TCP_SOCKET_HOST, CompLibClient.TCP_SOCKET_PORT))
+
+        from compLib.HealthCheck import HealthUpdater
+        HealthUpdater.start()
+
+    @staticmethod
+    def send(data: bytes, size: int) -> bytes:
+        with CompLibClient.LOCK:
+            if CompLibClient.SOCKET is None:
+                CompLibClient.use_unix_socket()
+
+            CompLibClient.SOCKET.sendall(size.to_bytes(1, byteorder='big'))
+            CompLibClient.SOCKET.sendall(data)
+
+            response_size_bytes = CompLibClient.SOCKET.recv(1)
+            response_size = int.from_bytes(
+                response_size_bytes, byteorder="big")
+            # print(response_size)
+
+            response_bytes = CompLibClient.SOCKET.recv(response_size)
+            # print(response_bytes.hex())
+            # print(len(response_bytes))
+
+            CompLibClient.check_response(response_bytes)
+
+            return response_bytes
+
+    @staticmethod
+    def check_response(response_bytes: bytes) -> bool:
+        # print(f"{response_bytes}")
+        res = CompLib_pb2.GenericResponse()
+        res.ParseFromString(response_bytes)
+
+        if res.status.successful:
+            return True
+
+        # TODO: Log error message if unsuccessful
+        return False

compLib/DoubleElimination.py

@@ -0,0 +1,161 @@
+import json
+import os
+import time
+from typing import Tuple, List, Dict
+
+import requests
+import logging
+
+logger = logging.getLogger("complib-logger")
+
+RETRY_TIMEOUT = 0.05
+
+# TODO: rethink how the api url is read
+API_URL = os.getenv("API_URL", "http://localhost:5000/") + "api/"
+
+api_override = os.getenv("API_FORCE", "")
+
+if api_override != "":
+    logger.warning(f"API_URL was set to {API_URL} but was overwritten with {api_override}")
+    API_URL = api_override
+
+API_URL_GET_ROBOT_STATE = API_URL + "getRobotState"
+
+API_URL_GET_POS = API_URL + "getPos"
+API_URL_GET_OP = API_URL + "getOp"
+API_URL_GET_GOAL = API_URL + "getGoal"
+API_URL_GET_ITEMS = API_URL + "getItems"
+API_URL_GET_SCORES = API_URL + "getScores"
+
+
+class Position:
+    """
+    Datenstruktur, welche eine Position representiert.
+
+    :ivar x: X Position in Centimeter
+    :ivar y: Y Position in Centimeter
+    :ivar degrees: Rotation in Grad von -180 bis 180
+    """
+
+    def __init__(self, x, y, degrees):
+        self.x = x
+        self.y = y
+        self.degrees = degrees
+
+    def __repr__(self):
+        return "{x=%s, y=%s, degrees=%s}" % (self.x, self.y, self.degrees)
+
+    def __str__(self):
+        return f"Position(x={round(self.x, 5)}, y={round(self.y, 5)}, degrees={round(self.degrees, 5)})"
+
+    def __eq__(self, o: object) -> bool:
+        if isinstance(o, Position):
+            return self.x == o.x and self.y == o.y and self.degrees == o.degrees
+        return False
+
+    def __ne__(self, o: object) -> bool:
+        return not self.__eq__(o)
+
+    @staticmethod
+    def position_from_json(json_str: Dict):
+        return Position(json_str["x"], json_str["y"], json_str["degrees"])
+
+
+class DoubleElim:
+    """Klasse für die Kommunikation mit Double Elimination Api
+    """
+
+    @staticmethod
+    def get_pos() -> Tuple[Position, int]:
+        """Führt den /api/getPos Aufruf an die API aus.
+
+        :return: Ein Objekt der Klasse :class:`.Position` mit der Position des Roboters und der Status Code
+        :rtype: Tuple[Position, int]
+        """
+        res = requests.get(API_URL_GET_POS)
+        if res.status_code == 408:
+            logger.error(f"DoubleElim.get_position timeout. API={API_URL_GET_POS}")
+            time.sleep(RETRY_TIMEOUT)
+            return DoubleElim.get_pos()
+        elif res.status_code == 503:
+            return Position(0, 0, -1), 503
+
+        response = json.loads(res.content)
+        logger.debug(f"DoubleElim.get_position = {response}, status code = {res.status_code}")
+        return Position(response["x"], response["y"], response["degrees"]), res.status_code
+
+    @staticmethod
+    def get_opponent() -> Tuple[Position, int]:
+        """Führt den /api/getOp Aufruf an die API aus.
+
+        :return: Ein Objekt der Klasse :class:`.Position` mit der Position des gegnerischen Roboters relativ zum eigenen Roboter und der Status Code
+        :rtype: Tuple[Position, int]
+        """
+        res = requests.get(API_URL_GET_OP)
+        if res.status_code == 408:
+            logger.error(f"DoubleElim.get_opponent timeout. API={API_URL_GET_OP}")
+            time.sleep(RETRY_TIMEOUT)
+            return DoubleElim.get_opponent()
+        elif res.status_code == 503:
+            return Position(0, 0, -1), 503
+
+        response = json.loads(res.content)
+        logger.debug(f"DoubleElim.get_opponent = x:{response}, status code = {res.status_code}")
+        return Position(response["x"], response["y"], response["degrees"]), res.status_code
+
+    @staticmethod
+    def get_goal() -> Tuple[Position, int]:
+        """Führt den /api/getGoal Aufruf an die API aus.
+
+        :return: Ein Objekt der Klasse :class:`.Position` mit der Position des Ziels relativ zum eigenen Roboter und der Status Code
+        :rtype: Tuple[Position, int]
+        """
+        res = requests.get(API_URL_GET_GOAL)
+        if res.status_code == 408:
+            logger.error(f"DoubleElim.get_goal timeout. API={API_URL_GET_GOAL}")
+            time.sleep(RETRY_TIMEOUT)
+            return DoubleElim.get_goal()
+        elif res.status_code == 503:
+            return Position(0, 0, -1), 503
+
+        response = json.loads(res.content)
+        logger.debug(f"DoubleElim.get_goal = {response}, status code = {res.status_code}")
+        return Position(response["x"], response["y"], -1), res.status_code
+
+    @staticmethod
+    def get_items() -> Tuple[List[Dict], int]:
+        """Führt den /api/getItems Aufruf an die API aus.
+
+        :return: Eine Liste aller Items, die sich derzeit auf dem Spielfeld befinden. Items sind "dictionaries", die wie folgt aussehen: {"id": 0, "x": 0, "y": 0}
+        :rtype: Tuple[List[Dict], int]
+        """
+        res = requests.get(API_URL_GET_ITEMS)
+        if res.status_code == 408:
+            logger.error(f"DoubleElim.get_items timeout. API={API_URL_GET_ITEMS}")
+            time.sleep(RETRY_TIMEOUT)
+            return DoubleElim.get_items()
+        elif res.status_code == 503:
+            return [], 503
+
+        response = json.loads(res.content)
+        logger.debug(f"DoubleElim.get_items = {response}, status code = {res.status_code}")
+        return response, res.status_code
+
+    @staticmethod
+    def get_scores() -> Tuple[Dict, int]:
+        """Führt den /api/getScores Aufruf an die API aus.
+
+        :return: Ein "dictionary" mit dem eignen Score und dem des Gegners: {"self":2,"opponent":0}
+        :rtype: Tuple[Dict, int]
+        """
+        res = requests.get(API_URL_GET_SCORES)
+        if res.status_code == 408:
+            logger.error(f"DoubleElim.get_scores timeout. API={API_URL_GET_SCORES}")
+            time.sleep(RETRY_TIMEOUT)
+            return DoubleElim.get_scores()
+        elif res.status_code == 503:
+            return {"self": 0, "opponent": 0}, 503
+
+        response = json.loads(res.content)
+        logger.debug(f"DoubleElim.get_scores = {response}, status code = {res.status_code}")
+        return response, res.status_code

compLib/Encoder.py

@@ -0,0 +1,37 @@
+import compLib.CompLib_pb2 as CompLib_pb2
+from compLib.CompLibClient import CompLibClient
+
+
+class Encoder(object):
+    """Klasse zum Zugriff auf die Encoder der einzelnen Motoren
+    """
+
+    @staticmethod
+    def read_all_positions():
+        """Lesen aller absoluten Positionen der einzelnen Encoder
+
+        :return: Tupel mit allen aktuellen Encoderpositionen
+        """
+        request = CompLib_pb2.EncoderReadPositionsRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+
+        response = CompLib_pb2.EncoderReadPositionsResponse()
+        response.ParseFromString(CompLibClient.send(
+            request.SerializeToString(), request.ByteSize()))
+
+        return tuple(i for i in response.positions)
+
+    @staticmethod
+    def read_all_velocities():
+        """Lesen der Geschwindigkeit aller angeschlossenen Motoren.
+
+        :return: Tupel aller aktuellen Motorgeschwindigkeiten in Radianten pro Sekunde
+        """
+        request = CompLib_pb2.EncoderReadVelocitiesRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+
+        response = CompLib_pb2.EncoderReadVelocitiesResponse()
+        response.ParseFromString(CompLibClient.send(
+            request.SerializeToString(), request.ByteSize()))
+
+        return tuple(i for i in response.velocities)

compLib/HealthCheck.py

@@ -0,0 +1,23 @@
+import threading
+import time
+
+import compLib.CompLib_pb2 as CompLib_pb2
+from compLib.CompLibClient import CompLibClient
+
+
+class HealthUpdater(object):
+    started = False
+
+    @staticmethod
+    def start():
+        if not HealthUpdater.started:
+            threading.Thread(target=HealthUpdater.loop, daemon=True).start()
+            HealthUpdater.started = True
+
+    @staticmethod
+    def loop():
+        while True:
+            request = CompLib_pb2.HealthUpdateRequest()
+            request.header.message_type = request.DESCRIPTOR.full_name
+            CompLibClient.send(request.SerializeToString(), request.ByteSize())
+            time.sleep(0.25)

compLib/IRSensor.py

@@ -0,0 +1,45 @@
+import time
+
+import compLib.CompLib_pb2 as CompLib_pb2
+
+from compLib.CompLibClient import CompLibClient
+
+
+class IRSensor(object):
+    """Ermöglicht den Zugriff auf die einzelnen IRSensoren des Roboters
+    """
+
+    @staticmethod
+    def read_all():
+        """Auslesen aller Sensoren gleichzeitig
+
+        :return: Array aller Sensorwerte
+        """
+        request = CompLib_pb2.IRSensorsReadAllRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+
+        response = CompLib_pb2.IRSensorsReadAllResponse()
+        response.ParseFromString(CompLibClient.send(
+            request.SerializeToString(), request.ByteSize()))
+
+        return [i for i in response.data]
+
+    @staticmethod
+    def enable():
+        """Aktivieren Infrarot-Sender. Muss bei jedem Programmstart ausgeführt werden.
+        """
+        request = CompLib_pb2.IRSensorsEnableRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+
+        CompLibClient.send(request.SerializeToString(), request.ByteSize())
+        time.sleep(0.1)  # IR sensor reading is async -> Wait a bit
+
+    @staticmethod
+    def disable():
+        """Deaktivieren der Infrarot-Sender
+        """
+        request = CompLib_pb2.IRSensorsDisableRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+
+        CompLibClient.send(request.SerializeToString(), request.ByteSize())
+        time.sleep(0.1)  # IR sensor reading is async -> Wait a bit

compLib/Motor.py

@@ -0,0 +1,142 @@
+import compLib.CompLib_pb2 as CompLib_pb2
+from compLib.CompLibClient import CompLibClient
+
+MOTOR_COUNT = 4
+
+
+class Motor(object):
+    """Klasse zum Ansteuern der Motoren
+    """
+
+    @staticmethod
+    def power(port: int, percent: float):
+        """Motor auf eine prozentuale Leistung der Höchstgeschwindigkeit einstellen
+
+        :param port: Port, an welchen der Motor angesteckt ist. 0-3
+        :param percent: Prozentsatz der Höchstgeschwindigkeit. zwischen -100 und 100
+        :raises: IndexError
+        """
+
+        if port < 0 or port >= MOTOR_COUNT:
+            raise IndexError("Invalid Motor port specified!")
+
+        if percent < -100 or percent > 100:
+            raise IndexError(
+                "Invalid Motor speed specified! Speed is between -100 and 100 percent!")
+
+        request = CompLib_pb2.MotorSetPowerRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+        request.port = port
+        request.power = percent
+
+        CompLibClient.send(request.SerializeToString(), request.ByteSize())
+
+    @staticmethod
+    def multiple_power(*arguments: tuple[int, float]):
+        """Mehrere Motoren auf eine prozentuale Leistung der Höchstgeschwindigkeit einstellen
+
+        :param arguments: tuple von port, percentage
+        :raises: IndexError
+        """
+        request = CompLib_pb2.MotorsSetPowerRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+
+        for port, percent in arguments:
+            if port < 0 or port >= MOTOR_COUNT:
+                raise IndexError("Invalid Motor port specified!")
+
+            if percent < -100 or percent > 100:
+                raise IndexError(
+                    "Invalid Motor speed specified! Speed is between -100 and 100 percent!")
+
+            inner_request = CompLib_pb2.MotorSetPowerRequest()
+            inner_request.port = port
+            inner_request.power = percent
+
+            request.requests.append(inner_request)
+
+        CompLibClient.send(request.SerializeToString(), request.ByteSize())
+
+    @staticmethod
+    def speed(port: int, speed: float):
+        """Geschwindigkeit des Motors einstellen
+
+        :param port: Port, an welchen der Motor angesteckt ist. 0-3
+        :param speed: Drehzahl, mit der sich ein Motor dreht, in Centimeter pro Sekunde (cm/s)
+        :raises: IndexError
+        """
+
+        if port < 0 or port >= MOTOR_COUNT:
+            raise IndexError("Invalid Motor port specified!")
+
+        request = CompLib_pb2.MotorSetSpeedRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+        request.port = port
+        request.speed = speed
+
+        CompLibClient.send(request.SerializeToString(), request.ByteSize())
+
+    @staticmethod
+    def multiple_speed(*arguments: tuple[int, float]):
+        """Geschwindigkeit mehrerer Motoren einstellen
+
+        :param arguments: tuple von port, Geschwindigkeit in Radianten pro Sekunde (rad/s)
+        :raises: IndexError
+        """
+
+        request = CompLib_pb2.MotorsSetSpeedRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+
+        for port, speed in arguments:
+            if port < 0 or port >= MOTOR_COUNT:
+                raise IndexError("Invalid Motor port specified!")
+
+            inner_request = CompLib_pb2.MotorSetSpeedRequest()
+            inner_request.port = port
+            inner_request.speed = speed
+
+            request.requests.append(inner_request)
+
+        CompLibClient.send(request.SerializeToString(), request.ByteSize())
+
+    @staticmethod
+    def pulse_width(port: int, percent: float):
+        """Setzen den Pulsbreite eines Motors in Prozent der Periode
+
+        :param port: Port, an welchen der Motor angesteckt ist. 0-3
+        :param percent: Prozent der Periode zwischen -100 und 100
+        :raises: IndexError
+        """
+
+        if port < 0 or port >= MOTOR_COUNT:
+            raise IndexError("Invalid Motor port specified!")
+
+        request = CompLib_pb2.MotorSetPulseWidthRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+        request.port = port
+        request.percent = percent
+
+        CompLibClient.send(request.SerializeToString(), request.ByteSize())
+
+    @staticmethod
+    def multiple_pulse_width(*arguments: tuple[int, float]):
+        """Setzen den Pulsbreite mehrerer Motoren in Prozent der Periode
+
+        :param arguments: tuple von port, prozent
+        :raises: IndexError
+        """
+
+        request = CompLib_pb2.MotorsSetPulseWidthRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+
+        for port, percent in arguments:
+            if port < 0 or port >= MOTOR_COUNT:
+                raise IndexError("Invalid Motor port specified!")
+
+            inner_request = CompLib_pb2.MotorSetPulseWidthRequest()
+            inner_request.port = port
+            inner_request.percent = percent
+
+            request.requests.append(inner_request)
+
+        CompLibClient.send(request.SerializeToString(), request.ByteSize())

compLib/Movement.py

@@ -0,0 +1,57 @@
+import compLib.CompLib_pb2 as CompLib_pb2
+from compLib.CompLibClient import CompLibClient
+
+
+class Movement(object):
+    """High level class to control movement of the robot
+    """
+
+    @staticmethod
+    def drive_distance(distance: float, speed: float):
+        """
+        Drive a given distance with a certain speed.
+        Positive distance and speed with result in forward motion. Everything else will move backwards.
+        :param distance: Distance in meters
+        :param speed: Speed in meters per second
+        :return: None
+        """
+        request = CompLib_pb2.DriveDistanceRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+        request.distance_m = distance
+        request.velocity_m_s = speed
+
+        response = CompLib_pb2.GenericResponse()
+        response.ParseFromString(CompLibClient.send(request.SerializeToString(), request.ByteSize()))
+
+    @staticmethod
+    def turn_degrees(degrees: float, speed: float):
+        """
+        Turn specified degrees with a given speed.
+        Positive degrees and speed with result in counter-clockwise motion. Everything else will be clockwise
+        :param degrees: Degrees between -180 and 180
+        :param speed: Speed in radians per second
+        :return: None
+        """
+        request = CompLib_pb2.TurnDegreesRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+        request.angle_degrees = degrees
+        request.velocity_rad_s = speed
+
+        response = CompLib_pb2.GenericResponse()
+        response.ParseFromString(CompLibClient.send(request.SerializeToString(), request.ByteSize()))
+
+    @staticmethod
+    def drive(linear: float, angular: float):
+        """
+        Non-blocking way to perform a linear and angular motion at the same time.
+        :param linear: Linear speed in meters per second
+        :param angular: Angular speed in radians per second
+        :return: None
+        """
+        request = CompLib_pb2.DriveDistanceRequest()
+        request.header.message_type = request.DESCRIPTOR.full_name
+        request.linear_velocity_m_s = linear
+        request.angular_velocity_rad_s = angular
+
+        response = CompLib_pb2.GenericResponse()
+        response.ParseFromString(CompLibClient.send(request.SerializeToString(), request.ByteSize()))

compLib/Seeding.py

@@ -0,0 +1,124 @@
+import logging
+import os
+
+import numpy as np
+
+# TODO: if set to competition mode, get the seed from the api
+FORCE_SEED = int(os.getenv("FORCE_SEED", "-1"))
+
+logger = logging.getLogger("complib-logger")
+
+
+class Gamestate:
+    @staticmethod
+    def __set_random_seed(seed: int):
+        logger.debug(f"Seeding seed to: {seed}")
+        np.random.seed(seed)
+
+    @staticmethod
+    def __get_random_number(min: int, max: int):
+        return np.random.randint(256 ** 4, dtype='<u4', size=1)[0] % (max - min + 1) + min
+
+    def __str__(self) -> str:
+        return f"""Seed: {self.seed}
+Heu Color: {self.heu_color}
+Material Pairs: {self.material_pairs}
+Material Zones: {self.materials}
+Logistic Plan: {self.logistic_plan}
+Logistic Centers: {self.logistic_center}"""
+
+    def __init__(self, seed: int):
+        """
+        Erstellt den Seeding "Gamestate" für den angegebenen Seed.
+
+        :param seed: Seed welcher zum Erstellen des Gamestates benutzt werden soll.
+        """
+
+        if FORCE_SEED == -1:
+            self.seed = seed
+        else:
+            print(f"Wettkampfmodus, zufälliger Seed wird verwendet: Seed={FORCE_SEED}")
+            self.seed = FORCE_SEED
+
+        logger.debug(f"Creating gamestate with seed: {self.seed}")
+        self.__set_random_seed(self.seed)
+
+        self.heu_color = self.__get_random_number(1, 2)
+
+        self.materials = [0, 0, 0, 0]
+        self.material_pairs = []
+        for i in range(0, 4):
+            num1 = self.__get_random_number(0, 3)
+            self.material_pairs.append([num1, num1])
+            while self.material_pairs[i][1] == num1:
+                self.material_pairs[i][1] = self.__get_random_number(0, 3)
+
+        flat = [item for sublist in self.material_pairs for item in sublist]
+        for i in range(0, 4):
+            self.materials[i] = flat.count(i)
+
+        self.logistic_plan = [0 for i in range(0, 21)]
+        self.logistic_center = [[0, 0, 0, 0] for i in range(0, 4)]
+        visited = [5, 5, 5, 5]
+
+        def __logistic_plan_generator(i: int):
+            drive_to = self.__get_random_number(0, 3)
+            for j in range(0, 4):
+                drive_to = (drive_to + j) % 4
+                if visited[drive_to] <= 0 or drive_to == self.logistic_plan[i - 1]:
+                    continue
+                self.logistic_plan[i] = drive_to
+
+                visited[drive_to] -= 1
+                finished = True
+                for k in visited:
+                    if k != 0:
+                        finished = False
+
+                if finished and drive_to == 2:
+                    visited[drive_to] += 1
+                    continue
+
+                if finished:
+                    return True
+
+                if i < len(self.logistic_plan):
+                    if __logistic_plan_generator(i + 1):
+                        return True
+                visited[drive_to] += 1
+            return False
+
+        self.logistic_plan[0] = 2
+        visited[2] -= 1
+        _ = __logistic_plan_generator(1)
+
+        self.logistic_plan[-1] = 2
+        for i in range(0, len(self.logistic_plan) - 1):
+            self.logistic_center[self.logistic_plan[i]][self.logistic_plan[i + 1]] += 1
+
+        self.logistic_plan = [x + 10 for x in self.logistic_plan]
+        logger.debug(f"Created gamesate: {str(self)}")
+
+    def get_heuballen(self) -> int:
+        """
+        Die Funktion gibt entweder die Zahl "1" oder "2" zurück. Wenn die Funktion "1" zurückgibt, dann liegen die Heuballen auf den gelben Linien. Wenn die Funktion "2" zurückgibt, dann liegen sie auf den blauen Flächen.
+
+        :return: Gibt entweder die Zahl 1 oder 2 zurück.
+        """
+        return self.heu_color
+
+    def get_logistic_plan(self) -> []:
+        """
+        Die Funktion gibt den "Logistik Plan" zurück. Also die Reihenfolge, in welcher der Roboter die Logistik Zonen Abfahren muss, um die Pakete welche dort liegen zu sortieren.
+
+        :return: Eine Liste an Zahlen zwischen 10 und 13.
+        """
+        return self.logistic_plan
+
+    def get_material_deliveries(self) -> [[]]:
+        """
+        Die Funktion gibt die einzelnen "Material Lieferungen" zurück. Da der Roboter immer zwei Paare an Materialien anliefern muss, gibt die Funktion eine Liste an Material Paaren zurück. Die Materialien werden dabei durch ihre Zonen-ID representiert. Also Holz ist z.B. "0" und die Ziegelsteine sind "3".
+
+        :return: Eine Liste and Material Paaren.
+        """
+        return self.material_pairs

compLib/__init__.py

@@ -0,0 +1,7 @@
+import logging
+import os
+
+if os.getenv("DEBUG", "0") != "0":
+    logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.DEBUG)
+else:
+    logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)