You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

83 lines
2.5 KiB

import logging
import json
import requests
from typing import Optional, Tuple
TIME_BEFORE_RETRY = 2.0
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("client")
def set_log_level(log_level: str):
logger.setLevel(log_level)
def get_state(min_round: int, args) -> Tuple[Optional[dict], float]:
"""Returns None and wait time if there was an error."""
try:
r = requests.get(f"{args.server}/api/state", params={
"game": args.game,
"token": args.token,
"min_round": min_round
})
# retry later if there was an error
except requests.exceptions.RequestException as e:
logger.warning(f"Request error: {e}")
return None, TIME_BEFORE_RETRY
if not r.ok:
logger.warning(f"Server error: {r.status_code} {r.reason}")
return None, TIME_BEFORE_RETRY
state = r.json()
# also retry if the server is not willing to give us the state yet
if state["status"] == "waiting":
logger.info("Server is busy.")
return None, state["wait"]
if state["status"] == "too_early":
logger.info("Round didn't start yet.")
return None, state["wait"]
logger.info("Received new state.")
return state, 0
def send_turn(turn: dict, round: int, args) -> bool:
"""Returns True if the server received the request."""
try:
r = requests.post(
f"{args.server}/api/action",
params={
"game": args.game,
"token": args.token,
"round": round
},
json=json.dumps(turn)
)
except requests.exceptions.RequestException as e:
logger.warning("Request error: {e}")
return False
if not r.ok:
logger.warning(f"Server error: {r.status_code} {r.reason}")
return False
# print errors
# because such errors are caused by the submitted turn,
# retrying will not help, so return True
response = r.json()
if response["status"] == "ok":
logger.info("Turn accepted.")
elif response["status"] == "too_late":
logger.error("Turn submitted too late.")
elif response["status"] == "error":
logger.error(f"Turn error: {response['description']}")
elif response["status"] == "warning":
member_errors = [
f" {member}: {error}"
for member, error in response["errors"].items()
]
logger.error("Member errors:\n" + "\n".join(member_errors))
return True