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("--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("--command", type=str, nargs="+", default=["python3", "%"], help="Command to execute as strategy code. Use '%' for program.")
parser.add_argument("--program", type=str, default=None, help="Program to substitute to command.")
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="info", choices=["error", "warning", "info"])
parser.add_argument("--save-state", type=str, default=None, help="Save state to this file.")


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())
    min_round = 0
    program = None
    if args.save_state is None:
        program = get_command(args)

    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 get_command(args) -> List[str]:
    if args.program is None:
        logger.error("Program not specified.")
        sys.exit(1)

    program = args.program
    if args.copy:
        program = tempfile.mktemp()
        shutil.copyfile(args.program, program)

    command: List[str] = []
    for arg in args.command:
        command.append(arg.replace("%", program))
    return command


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

    proc = Popen(command, 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:
            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
        res_json = res.json()
        if res_json["status"] == "ok":
            logger.info("Turn accepted.")
        elif res_json["status"] == "too_late":
            logger.error("Turn submitted too late.")
        elif res_json["status"] == "error":
            logger.error(f"Turn error: {res_json['description']}")
        elif res_json["status"] == "warning":
            member_warns = "\n".join([
                f"    {member['id']}: {member['description']}"
                for member in res_json["members"]
            ])
            logger.info("Turn accepted with warnings.")
            logger.warning(f"Member warnings:\n{member_warns}")
        return True

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


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


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