SKSP_2022_strategicka_hra/klient/client.py

#!/usr/bin/env python3
import argparse
import json
import logging
import requests
import shutil
import sys
import tempfile
import time
from subprocess import Popen, PIPE, TimeoutExpired
from typing import List, Optional, Tuple


parser = argparse.ArgumentParser()
parser.add_argument("--token", type=str, required=True, help="API token.")
parser.add_argument("--program", type=str, required=True, help="Program to run.")
parser.add_argument("--server", type=str, default="http://localhost:5000", help="Server address.")
parser.add_argument("--game", type=str, default="main", help="'main' or 'test_#'.")
parser.add_argument("--copy", default=False, action="store_true", help="Save a copy of the program to ensure the same code is used all the time.")
parser.add_argument("--log-level", type=str, default="warning", choices=["error", "warning", "info"])
parser.add_argument("--save-state", type=str, default=None, help="Save state to this file.")
# parser.add_argument("program", nargs=argparse.REMAINDER, help="Program to run.")


TIME_BEFORE_RETRY = 2.0
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("client")


def main(args: argparse.Namespace) -> None:
    logger.setLevel(args.log_level.upper())
    program = args.program
    if args.copy:
        program = tempfile.mktemp()
        shutil.copyfile(args.program, program)
    program = "./" + program
    min_round = 0

    while True:
        state = get_state(min_round, args)
        round = state["round"]
        min_round = round + 1
        # if requested, dump state to file instead
        if args.save_state is not None:
            with open(args.save_state, "w") as f:
                json.dump(state, f)
                return

        turn_json = run_subprocess(
            program, json.dumps(state), state["time_to_response"]
        )
        if turn_json is None:
            continue
        try:
            turn = json.loads(turn_json)
        except json.JSONDecodeError:
            logger.error(f"Invalid JSON returned by strategy code.")
            sys.exit(1)

        send_turn(turn, round, args)


def run_subprocess(
    program: str, state_json: str, timeout: Optional[float]
) -> Optional[str]:
    """Run user program and return its output."""

    proc = Popen([program], encoding="utf-8",
                 stdin=PIPE, stdout=PIPE, stderr=PIPE)
    try:
        stdout, stderr = proc.communicate(input=state_json, timeout=timeout)
    except TimeoutExpired:
        proc.kill()
        logger.error("Strategy code took too long.")
        return None
    if stderr:
        logger.error(f"Strategy code stderr:\n{stderr}")
    if proc.returncode != 0:
        logger.error("Strategy code exited with non-zero exit code.")
        sys.exit(1)
    return stdout


def get_state(min_round: int, args) -> dict:
    """Iteratively try to get game data from the server."""

    def get_state_once() -> Tuple[Optional[dict], float]:
        try:
            res = requests.get(f"{args.server}/api/state", params={
                "game": args.game,
                "token": args.token,
                "min_round": min_round
            })
        except requests.exceptions.RequestException as e:
            # server might be down, retry later
            logger.warning(f"Request error: {e}")
            return None, TIME_BEFORE_RETRY
        if not res.ok:
            logger.error(f"Server error: {res.status_code} {res.reason}")
            log_server_error(res)
            sys.exit(1)

        state = res.json()
        if state["status"] == "ok":
            logger.info("Received new state.")
            return state, 0

        # retry after some time
        if state["status"] == "waiting":
            logger.info("Server is busy.")
        if state["status"] == "too_early":
            logger.info("Round didn't start yet.")
        return None, state["wait"]

    state, wait_time = get_state_once()
    while state is None:
        time.sleep(wait_time)
        state, wait_time = get_state_once()
    return state


def send_turn(turn: dict, round: int, args) -> None:
    """Iteratively try to send turn to the server."""

    def send_turn_once() -> bool:
        """Returns True if server answered."""

        try:
            res = requests.post(
                f"{args.server}/api/action",
                params={
                    "game": args.game,
                    "token": args.token,
                    "round": round
                },
                json=turn
            )
        except requests.exceptions.RequestException as e:
            # server might be down, retry later
            logger.warning(f"Request error: {e}")
            return False
        if not res.ok:
            log_server_error(res)
            sys.exit(1)

        # print errors
        # because such errors are caused by the submitted turn,
        # retrying will not help, so return True
        response = res.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_warns = [
                f"    {member['id']}: {member['description']}"
                for member in response["members"]
            ]
            logger.info("Turn accepted with warnings.")
            logger.warning(
                "Member warnings:\n"
                "\n".join(member_warns)
            )
        return True

    while not send_turn_once():
        time.sleep(TIME_BEFORE_RETRY)


def log_server_error(res: requests.Response) -> None:
    logger.error(
        f"Server error: {res.status_code} {res.reason}\n"
        f"{res['description']}"
    )


if __name__ == "__main__":
    args = parser.parse_args()
    main(args)