Browse Source

Strategická: Klient

master
David Klement 2 years ago
parent
commit
40309ef291
  1. 78
      klient/client.py
  2. 74
      klient/play.py
  3. 1
      klient/requirements.txt

78
klient/client.py

@ -0,0 +1,78 @@
import logging
import json
import requests
from typing import Optional, Tuple
TIME_BEFORE_RETRY = 2.0
logger = logging.Logger("client")
def get_state(min_round: int, args) -> Tuple[Optional[dict], float]:
"""Returns None and wait time if there was an error."""
try:
r = requests.get(f"{args.server}/api/state", params={
"game": args.game,
"token": args.token,
"min_round": min_round
})
# retry later if there was an error
except requests.exceptions.RequestException as e:
logger.warning("Request error: {e}")
return None, TIME_BEFORE_RETRY
if not r.ok:
logger.warning(f"Server error: {r.status_code} {r.reason}")
return None, TIME_BEFORE_RETRY
# also retry if the server is not willing to give us the state yet
if state["status"] == "waiting":
logger.info("Server is busy.")
return None, state["wait"]
if state["status"] == "too_early":
logger.info("Round didn't start yet.")
return None, state["wait"]
state = json.loads(r.json())
logger.info("Received new state.")
return state, 0
def send_turn(turn: dict, round: int, args) -> bool:
"""Returns True if the server received the request."""
try:
r = requests.post(
f"{args.server}/api/state",
params={
"game": args.game,
"token": args.token,
"round": round
},
json=json.dumps(turn)
)
except requests.exceptions.RequestException as e:
logger.warning("Request error: {e}")
return False
if not r.ok:
logger.warning(f"Server error: {r.status_code} {r.reason}")
return False
# print errors
# because such errors are caused by the submitted turn,
# retrying will not help, so return True
response = json.loads(r.json())
if response["status"] == "ok":
logger.info("Turn accepted.")
elif response["status"] == "too_late":
logger.error("Turn submitted too late.")
elif response["status"] == "error":
logger.error(f"Turn error: {response['description']}")
elif response["status"] == "warning":
member_errors = [
f" {member}: {error}"
for member, error in response["errors"].items()
]
logger.error(f"Member errors:\n{'\n'.join(member_errors)}")
return True

74
klient/play.py

@ -0,0 +1,74 @@
#!/usr/bin/env python3
import argparse
import time
from client import get_state, send_turn, logger, 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
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"
}]
}
```
"""
return {}
def main(args: argparse.Namespace):
min_round = 0
logger.setLevel(args.log_level)
while True:
state, wait_time = get_state(min_round)
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):
# 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)

1
klient/requirements.txt

@ -0,0 +1 @@
requests
Loading…
Cancel
Save