#!/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="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 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 # if requested, dump state to file instead 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"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 = 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]: """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)