70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
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:
|
|
"""Class used for communicating with seeding api
|
|
"""
|
|
|
|
@staticmethod
|
|
def get_heuballen() -> int:
|
|
"""Makes the /api/getHeuballen call to the api.
|
|
|
|
:return: hueballencode as 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:
|
|
"""Makes the /api/getLogisticPlan call to the api.
|
|
|
|
:return: Json Object and status code as returned by the api.
|
|
: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:
|
|
"""Makes the /api/getMaterialDeliveries call to the 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
|