233 lines
8.5 KiB
Python
233 lines
8.5 KiB
Python
import json
|
|
import os
|
|
import time
|
|
from typing import Dict, Tuple, List
|
|
|
|
import requests
|
|
|
|
from compLib.LogstashLogging import Logging
|
|
|
|
API_URL = os.getenv("API_URL", "http://localhost:5000/") + "api/"
|
|
|
|
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_DELIVERY = API_URL + "getDelivery"
|
|
API_URL_GET_MATERIAL = API_URL + "getMaterial"
|
|
API_URL_GET_GARBAGE = API_URL + "getGarbage"
|
|
API_URL_GET_LIST_CARGO = API_URL + "listCargo"
|
|
API_URL_GET_CARGO = API_URL + "getCargo/"
|
|
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"
|
|
API_URL_GET_METEOROID = API_URL + "getMeteoroids"
|
|
|
|
|
|
class Seeding:
|
|
"""Class used for communicating with seeding api
|
|
"""
|
|
|
|
@staticmethod
|
|
def get_delivery() -> Tuple[Dict, int]:
|
|
"""Makes the /api/getDelivery call to the api.
|
|
|
|
:return: Json Object and status code as returned by the api.
|
|
:rtype: Tuple[Dict, int]
|
|
"""
|
|
res = requests.get(API_URL_GET_DELIVERY)
|
|
result = json.loads(res.content)
|
|
Logging.get_logger().debug(f"Seeding.get_delivery = {result}, status code = {res.status_code}")
|
|
return result, res.status_code
|
|
|
|
@staticmethod
|
|
def get_material() -> Tuple[Dict, int]:
|
|
"""Makes the /api/getMaterial call to the api.
|
|
|
|
:return: Json Object and status code as returned by the api.
|
|
:rtype: Tuple[Dict, int]
|
|
"""
|
|
res = requests.get(API_URL_GET_MATERIAL)
|
|
result = json.loads(res.content)
|
|
Logging.get_logger().debug(f"Seeding.get_material = {result}, status code = {res.status_code}")
|
|
return result, res.status_code
|
|
|
|
@staticmethod
|
|
def get_garbage() -> Tuple[Dict, int]:
|
|
"""Makes the /api/getGarbage call to the api.
|
|
|
|
:return: Json Object and status code as returned by the api.
|
|
:rtype: Tuple[Dict, int]
|
|
"""
|
|
res = requests.get(API_URL_GET_GARBAGE)
|
|
result = json.loads(res.content)
|
|
Logging.get_logger().debug(f"Seeding.get_garbage {result}, status code = {res.status_code}")
|
|
return result, res.status_code
|
|
|
|
@staticmethod
|
|
def list_cargo() -> Tuple[Dict, int]:
|
|
"""Makes the /api/listCargo call to the api.
|
|
|
|
:return: Json Object and status code as returned by the api.
|
|
:rtype: Tuple[Dict, int]
|
|
"""
|
|
res = requests.get(API_URL_GET_LIST_CARGO)
|
|
result = json.loads(res.content)
|
|
Logging.get_logger().debug(f"Seeding.list_cargo {result}, status code = {res.status_code}")
|
|
return result, res.status_code
|
|
|
|
@staticmethod
|
|
def get_cargo(color: str) -> Tuple[Dict, int]:
|
|
"""Makes the /api/getCargo call to the api.
|
|
|
|
:param color: Color parameter which specifies which cargo should be taken. (A string which is either "green", "red", "yellow" or "blue") The function only picks up one package.
|
|
:return: Json Object and status code as returned by the api.
|
|
:rtype: Tuple[Dict, int]
|
|
"""
|
|
res = requests.get(API_URL_GET_CARGO + color)
|
|
result = json.loads(res.content)
|
|
Logging.get_logger().debug(f"Seeding.get_cargo {result}, status code = {res.status_code}")
|
|
return result, res.status_code
|
|
|
|
@staticmethod
|
|
def get_robot_state() -> Tuple[Dict, int]:
|
|
res = requests.get(API_URL_GET_ROBOT_STATE)
|
|
result = json.loads(res.content)
|
|
Logging.get_logger().debug(f"Seeding.get_robot_state {result}, status code = {res.status_code}")
|
|
return result, res.status_code
|
|
|
|
|
|
class Position:
|
|
"""Datastructure for holding a position
|
|
"""
|
|
|
|
def __init__(self, x, y, degrees):
|
|
self.x = x
|
|
self.y = y
|
|
self.degrees = degrees
|
|
|
|
def __repr__(self):
|
|
return f"Position(x={self.x}, y={self.y}, degrees={self.degrees})"
|
|
|
|
def __str__(self):
|
|
return f"Position(x={round(self.x, 5)}, y={round(self.y, 5)}, degrees={round(self.degrees, 5)})"
|
|
|
|
|
|
class DoubleElim:
|
|
"""Class used for communicating with double elimination api
|
|
"""
|
|
|
|
@staticmethod
|
|
def get_position() -> Tuple[Position, int]:
|
|
"""Makes the /api/getPos call to the api.
|
|
|
|
:return: A Position object with robot position
|
|
:rtype: Tuple[Position, int]
|
|
"""
|
|
res = requests.get(API_URL_GET_POS)
|
|
if res.status_code == 408:
|
|
Logging.get_logger().error(f"DoubleElim.get_position timeout!")
|
|
time.sleep(0.01)
|
|
return DoubleElim.get_position()
|
|
|
|
response = json.loads(res.content)
|
|
Logging.get_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]:
|
|
"""Makes the /api/getOp call to the api.
|
|
|
|
:return: A Position object with opponents robot position
|
|
:rtype: Tuple[Position, int]
|
|
"""
|
|
res = requests.get(API_URL_GET_OP)
|
|
if res.status_code == 408:
|
|
Logging.get_logger().error(f"DoubleElim.get_opponent timeout!")
|
|
time.sleep(0.01)
|
|
return DoubleElim.get_opponent()
|
|
|
|
response = json.loads(res.content)
|
|
Logging.get_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]:
|
|
"""Makes the /api/getGoal call to the api.
|
|
|
|
:return: A Position object with x and y coordinates of the goal, rotation is always -1
|
|
:rtype: Tuple[Position, int]
|
|
"""
|
|
res = requests.get(API_URL_GET_GOAL)
|
|
if res.status_code == 408:
|
|
Logging.get_logger().error(f"DoubleElim.get_goal timeout!")
|
|
time.sleep(0.01)
|
|
return DoubleElim.get_goal()
|
|
|
|
response = json.loads(res.content)
|
|
Logging.get_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]:
|
|
"""Makes the /api/getItems call to the api.
|
|
|
|
:return: A list will all items currently on the game field. Items are dictionaries that look like: {"id": 0, "x": 0, "y": 0}
|
|
:rtype: Tuple[List[Dict], int]
|
|
"""
|
|
res = requests.get(API_URL_GET_ITEMS)
|
|
if res.status_code == 408:
|
|
Logging.get_logger().error(f"DoubleElim.get_items timeout!")
|
|
time.sleep(0.01)
|
|
return DoubleElim.get_items()
|
|
elif res.status_code == 503:
|
|
return [], 503
|
|
|
|
response = json.loads(res.content)
|
|
Logging.get_logger().debug(f"DoubleElim.get_items = {response}, status code = {res.status_code}")
|
|
return response, res.status_code
|
|
|
|
@staticmethod
|
|
def get_scores() -> Tuple[Dict, int]:
|
|
"""Makes the /api/getScores call to the api.
|
|
|
|
:return: A dictionary with all scores included like: {"self":2,"opponent":0}
|
|
:rtype: Tuple[Dict, int]
|
|
"""
|
|
res = requests.get(API_URL_GET_SCORES)
|
|
if res.status_code == 408:
|
|
Logging.get_logger().error(f"DoubleElim.get_scores timeout!")
|
|
time.sleep(0.01)
|
|
return DoubleElim.get_scores()
|
|
elif res.status_code == 503:
|
|
return {"self": 0, "opponent": 0}, 503
|
|
|
|
response = json.loads(res.content)
|
|
Logging.get_logger().debug(f"DoubleElim.get_scores = {response}, status code = {res.status_code}")
|
|
return response, res.status_code
|
|
|
|
@staticmethod
|
|
def get_meteoroids() -> Tuple[List[Dict], int]:
|
|
"""Makes the /api/getMeteoroids call to the api.
|
|
|
|
:return: A list will all meteoroids currently on the game field. Meteoroids are dictionaries that look like: {"x": 0, "y": 0}
|
|
:rtype: Tuple[List[Dict], int]
|
|
"""
|
|
res = requests.get(API_URL_GET_METEOROID)
|
|
if res.status_code == 408:
|
|
Logging.get_logger().error(f"DoubleElim.get_meteoroids timeout!")
|
|
time.sleep(0.01)
|
|
return DoubleElim.get_meteoroids()
|
|
elif res.status_code == 503:
|
|
return [], 503
|
|
|
|
response = json.loads(res.content)
|
|
Logging.get_logger().debug(f"DoubleElim.get_meteoroids = {response}, status code = {res.status_code}")
|
|
return response, res.status_code
|
|
|