Browse Source

Strategická: Podproces

master
David Klement 2 years ago
parent
commit
7cb7908da7
  1. 94
      klient/client.py
  2. 114
      klient/play.py

94
klient/client.py

@ -1,6 +1,24 @@
#!/usr/bin/env python3
import argparse
import json
import logging import logging
import requests 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 TIME_BEFORE_RETRY = 2.0
@ -8,12 +26,71 @@ logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("client") logger = logging.getLogger("client")
def set_log_level(log_level: str): def main(args: argparse.Namespace) -> None:
logger.setLevel(log_level) 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]: 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: try:
r = requests.get(f"{args.server}/api/state", params={ 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: 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: try:
r = requests.post( 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)) logger.warn("Member warnings:\n" + "\n".join(member_warns))
return True return True
if __name__ == "__main__":
args = parser.parse_args()
main(args)

114
klient/play.py

@ -1,19 +1,17 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import argparse import json
import time import sys
from typing import Callable, Iterable, List, Optional, Tuple from typing import Callable, Iterable, List, Optional, Tuple
from client import get_state, send_turn, set_log_level, TIME_BEFORE_RETRY
parser = argparse.ArgumentParser() Coords = Tuple[int, int]
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] 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: class Member:
@ -34,77 +32,44 @@ class Field:
self.members = members self.members = members
def strategy(team_id: int, state: dict, args: argparse.Namespace) -> dict: state: State = None
"""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: def main() -> None:
```yaml global state
{ state = State(json.loads(sys.stdin.read()))
map: [[{ my_members = find_team_members(state.my_team_id)
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"
}]
}
```
"""
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 # TODO: set actions for all members
for member, coords in members: # example: all members go up
for member in my_members:
member.action = "up" member.action = "up"
return build_turn(map(lambda x: x[0], members))
print(json.dumps(build_turn(my_members)))
def find_fields( def find_fields(
world: List[List[Field]],
predicate: Callable[[Field], bool] predicate: Callable[[Field], bool]
) -> List[Tuple[Field, Coords]]: ) -> List[Tuple[Field, Coords]]:
"""Finds all fields that satisfy the given predicate.""" """Find all fields that satisfy the given predicate."""
result = [] result = []
for y, row in enumerate(world): for row in state.world:
for x, field in enumerate(row): for field in row:
if predicate(field): if predicate(field):
result.append((field, (x, y))) result.append(field)
return result return result
def find_members( def find_team_members(team_id: int) -> List[Tuple[Member]]:
world: List[List[Field]], """Find all members that belong to the given team."""
predicate: Callable[[Member], bool]
) -> List[Tuple[Member, Coords]]:
"""Finds all members that satisfy the given predicate."""
result = [] result = []
for y, row in enumerate(world): for row in state.world:
for x, field in enumerate(row): for field in row:
for member in field.members: for member in field.members:
if predicate(member): if member.team == team_id:
result.append((member, (x, y))) result.append((member))
return result return result
@ -139,26 +104,5 @@ def parse_world(world: dict) -> List[List[Field]]:
return fields 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__': if __name__ == '__main__':
args = parser.parse_args() main()
main(args)

Loading…
Cancel
Save