Merge branch 'sksp2022-strategicka' of gitea.ks.matfyz.cz:KSP/ksp into sksp2022-strategicka

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

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)

klient/requirements.txt

@@ -0,0 +1 @@
+requests