SKSP_2022_strategicka_hra/klient/client.py

#!/usr/bin/env python3
import argparse
import json
import logging
import os
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="Adresa severu.")
parser.add_argument("--game", type=str, default="main", help="'main' nebo 'test_#'.")
parser.add_argument("--command", type=str, nargs="+", default=["python3", "%"], help="Příkaz pro spuštění strategie. '%' bude nahrazeno programem.")
parser.add_argument("--program", type=str, default="play.py", help="Program, který má být použít v příkazu.")
parser.add_argument("--copy", default=False, action="store_true", help="Vyrob kopii souboru, aby se vždy použil stejný kód.")
parser.add_argument("--log-level", type=str, default="info", choices=["error", "warning", "info"])
parser.add_argument("--save-state", type=str, default=None, help="Pouze ulož herní info do souboru.")


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

    while True:
        state = get_state(min_round, args)
        round = state["round"]
        min_round = round + 1
        # ulož herní stav a ukonči program
        if args.save_state is not None:
            if args.save_state == "-":
                print(json.dumps(state))
            else:
                with open(args.save_state, "w") as f:
                    json.dump(state, f)
            return

        turn_json = run_subprocess(
            command, 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"Strategie vypsala nevalidní JSON.")
            sys.exit(1)

        send_turn(turn, round, args)


def get_command(args) -> List[str]:
    if args.program is None:
        logger.error("Specifikuj program.")
        sys.exit(1)

    program = os.path.realpath(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]:
    """Spusť strategii a vypiš její výstup."""

    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("Strategie se počítala příliš dlouho.")
        return None
    if stderr:
        logger.error(f"Chybový výstup strategie:\n{stderr}")
    if proc.returncode != 0:
        logger.error("Strategie skončila s chybovým návratovým kódem.")
        sys.exit(1)
    return stdout


def get_state(min_round: int, args) -> dict:
    """Opakovaně se pokus získat herní data ze serveru."""

    def get_state_once() -> Tuple[Optional[dict], float]:
        """Vrací stav případě úspěchu, jinak None a dobu čekání."""

        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 mohl vypadnout, zkus to znovu později
            logger.warning(f"Chyba při dotazu na stav: {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("Nový stav obdržen.")
            return state, 0

        # musíme chvíli počkat
        if state["status"] == "waiting":
            logger.info("Server počítá.")
        if state["status"] == "too_early":
            logger.info("Kolo ještě nezačalo.")
        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:
    """Opakovaně se pokus poslat tah na server."""

    def send_turn_once() -> bool:
        """Vrací True pokud server odpověděl."""

        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 mohl vypadnout, zkus to znovu později
            logger.warning(f"Chyba při posílání tahu: {e}")
            return False
        if not res.ok:
            log_server_error(res)
            sys.exit(1)

        # vypiš chyby
        # tyto chyby jsou způsobeny špatným tahem,
        # opakování požadavku nepomůže, takže vracíme True
        res_json = res.json()
        if res_json["status"] == "ok":
            logger.info("Tah úspěšně přijat.")
        elif res_json["status"] == "too_late":
            logger.error("Tah poslán příliš pozdě.")
        elif res_json["status"] == "error":
            logger.error(f"Chyba v tahu: {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("Tah přijat s varováními.")
            logger.warning(f"Varování pro vojáky:\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"Chyba serveru: {res.status_code} {res.reason}: "
        f"{res_json['description']}"
    )


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