SKSP_2022_strategicka_hra/klient/client.py

import logging
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=turn
        )
    except requests.exceptions.RequestException as e:
        logger.warning(f"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