From 7cb7908da7f1cbcbd45d1544412732f0a1e4f9f3 Mon Sep 17 00:00:00 2001 From: kulisak12 Date: Sat, 17 Sep 2022 17:11:17 +0200 Subject: [PATCH] =?UTF-8?q?Strategick=C3=A1:=20Podproces?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- klient/client.py | 94 ++++++++++++++++++++++++++++++++++++-- klient/play.py | 116 ++++++++++++----------------------------------- 2 files changed, 119 insertions(+), 91 deletions(-) diff --git a/klient/client.py b/klient/client.py index 5b125ba..820db5f 100644 --- a/klient/client.py +++ b/klient/client.py @@ -1,6 +1,24 @@ +#!/usr/bin/env python3 +import argparse +import json import logging import requests -from typing import Optional, Tuple +import shutil +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 @@ -8,12 +26,71 @@ logging.basicConfig(level=logging.INFO) logger = logging.getLogger("client") -def set_log_level(log_level: str): - logger.setLevel(log_level) +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, wait_time = get_state(min_round, args) + if state is None: + # retry later + time.sleep(wait_time) + continue + + 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.") + continue + + while not send_turn(turn, round, args): + # if there was a connection error, retry later + time.sleep(TIME_BEFORE_RETRY) + + +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.") + return None + return stdout def get_state(min_round: int, args) -> Tuple[Optional[dict], float]: - """Returns None and wait time if there was an error.""" + """Get game data from the server. + + Returns None and wait time if there was an error.""" try: r = requests.get(f"{args.server}/api/state", params={ @@ -43,7 +120,9 @@ def get_state(min_round: int, args) -> Tuple[Optional[dict], float]: def send_turn(turn: dict, round: int, args) -> bool: - """Returns True if the server received the request.""" + """Send turn to the server. + + Returns True if the server received the request.""" try: r = requests.post( @@ -80,3 +159,8 @@ def send_turn(turn: dict, round: int, args) -> bool: logger.warn("Member warnings:\n" + "\n".join(member_warns)) return True + + +if __name__ == "__main__": + args = parser.parse_args() + main(args) diff --git a/klient/play.py b/klient/play.py index 7a3fc11..18674ae 100755 --- a/klient/play.py +++ b/klient/play.py @@ -1,19 +1,17 @@ #!/usr/bin/env python3 -import argparse -import time +import json +import sys from typing import Callable, Iterable, List, Optional, Tuple -from client import get_state, send_turn, set_log_level, TIME_BEFORE_RETRY -parser = argparse.ArgumentParser() -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("--token", type=str, default=None, help="API token.") -parser.add_argument("--log_level", type=str, default="WARNING", choices=["ERROR", "WARNING", "INFO"]) -# you can add your own arguments, the strategy function will get them +Coords = Tuple[int, int] -Coords = Tuple[int, int] +class State: + def __init__(self, state: dict) -> None: + self.world = parse_world(state["state"]["map"]) + self.my_team_id: int = state["team_id"] + self.round_number: int = state["round"] class Member: @@ -34,77 +32,44 @@ class Field: self.members = members -def strategy(team_id: int, state: dict, args: argparse.Namespace) -> dict: - """Finds the best move for the given state. - - This function is called every round. - It should return a dict with actions for each of team's soldiers. - - State format: - ```yaml - { - map: [[{ - home_for_team: Optional[int], - occupied_by_team: Optional[int], - hill: bool, - members: [{ - type: "soldier", - team: int, - id: int, - remaining_rounds: int - }] - }]] - } - ``` - - Turn format: - ```yaml - { - members: [{ - id: int, - action: Optional[str] - # None, "left", "right", "up", "down" - }] - } - ``` - """ +state: State = None - world = parse_world(state["map"]) - home = find_fields(world, lambda f: f.home_for_team == team_id)[0][1] - members = find_members(world, lambda m: m.team == team_id) - # TODO: implement your strategy here - for member, coords in members: +def main() -> None: + global state + state = State(json.loads(sys.stdin.read())) + my_members = find_team_members(state.my_team_id) + + # TODO: set actions for all members + # example: all members go up + for member in my_members: member.action = "up" - return build_turn(map(lambda x: x[0], members)) + + print(json.dumps(build_turn(my_members))) def find_fields( - world: List[List[Field]], predicate: Callable[[Field], bool] ) -> List[Tuple[Field, Coords]]: - """Finds all fields that satisfy the given predicate.""" + """Find all fields that satisfy the given predicate.""" result = [] - for y, row in enumerate(world): - for x, field in enumerate(row): + for row in state.world: + for field in row: if predicate(field): - result.append((field, (x, y))) + result.append(field) return result -def find_members( - world: List[List[Field]], - predicate: Callable[[Member], bool] -) -> List[Tuple[Member, Coords]]: - """Finds all members that satisfy the given predicate.""" +def find_team_members(team_id: int) -> List[Tuple[Member]]: + """Find all members that belong to the given team.""" result = [] - for y, row in enumerate(world): - for x, field in enumerate(row): + for row in state.world: + for field in row: for member in field.members: - if predicate(member): - result.append((member, (x, y))) + if member.team == team_id: + result.append((member)) return result @@ -139,26 +104,5 @@ def parse_world(world: dict) -> List[List[Field]]: return fields -def main(args: argparse.Namespace): - min_round = 0 - set_log_level(args.log_level) - - while True: - state, wait_time = get_state(min_round, args) - if state is None: - # retry later - time.sleep(wait_time) - continue - - turn = strategy(state["team_id"], state["state"], args) - - round = state["round"] - while not send_turn(turn, round, args): - # if there was a connection error, retry later - time.sleep(TIME_BEFORE_RETRY) - min_round = round + 1 - - if __name__ == '__main__': - args = parser.parse_args() - main(args) + main()