diff --git a/examples/client_example.py b/examples/client_example.py index 6cc357c..d23228a 100644 --- a/examples/client_example.py +++ b/examples/client_example.py @@ -1,11 +1,11 @@ -from generals.agents.random_agent import RandomAgent +from generals.agents import ExpanderAgent from generals.remote import GeneralsIOClient if __name__ == "__main__": - agent = RandomAgent() + agent = ExpanderAgent() with GeneralsIOClient(agent, "user_id9l") as client: # register call will fail when given username is already registered - client.register_agent("[Bot]MyEpicUsername") - client.join_private_lobby("queueID") + # client.register_agent("[Bot]MyEpicUsername") + client.join_private_lobby("6ngz") client.join_game() diff --git a/generals/core/game.py b/generals/core/game.py index 0330945..6aa707f 100644 --- a/generals/core/game.py +++ b/generals/core/game.py @@ -1,4 +1,3 @@ -import warnings from typing import Any import gymnasium as gym diff --git a/generals/core/observation.py b/generals/core/observation.py index 02b95f9..a410a4f 100644 --- a/generals/core/observation.py +++ b/generals/core/observation.py @@ -1,8 +1,6 @@ import numpy as np -from scipy.ndimage import maximum_filter # type: ignore from generals.core.game import DIRECTIONS, Game -from generals.remote.generalsio_client import GeneralsIOState def observation_from_simulator(game: Game, agent_id: str) -> "Observation": @@ -48,56 +46,13 @@ def observation_from_simulator(game: Game, agent_id: str) -> "Observation": ) -def observation_from_generalsio_state(state: GeneralsIOState) -> "Observation": - width, height = state.map[0], state.map[1] - size = height * width - - armies = np.array(state.map[2 : 2 + size]).reshape((height, width)) - terrain = np.array(state.map[2 + size : 2 + 2 * size]).reshape((height, width)) - cities = np.zeros((height, width)) - for city in state.cities: - cities[city // width, city % width] = 1 - - generals = np.zeros((height, width)) - for general in state.generals: - if general != -1: - generals[general // width, general % width] = 1 - - army = armies - owned_cells = np.where(terrain == state.player_index, 1, 0) - opponent_cells = np.where(terrain == state.opponent_index, 1, 0) - neutral_cells = np.where(terrain == -1, 1, 0) - visible_cells = maximum_filter(np.where(terrain == state.player_index, 1, 0), size=3) - structures_in_fog = np.where(terrain == -4, 1, 0) - owned_land_count = state.scores[state.player_index]["tiles"] - owned_army_count = state.scores[state.player_index]["total"] - opponent_land_count = state.scores[state.opponent_index]["tiles"] - opponent_army_count = state.scores[state.opponent_index]["total"] - timestep = state.turn - - return Observation( - army=army, - generals=generals, - city=cities, - owned_cells=owned_cells, - opponent_cells=opponent_cells, - neutral_cells=neutral_cells, - visible_cells=visible_cells, - structures_in_fog=structures_in_fog, - owned_land_count=owned_land_count, - owned_army_count=owned_army_count, - opponent_land_count=opponent_land_count, - opponent_army_count=opponent_army_count, - timestep=timestep, - ) - - class Observation: def __init__( self, - army: np.ndarray, + armies: np.ndarray, generals: np.ndarray, - city: np.ndarray, + cities: np.ndarray, + mountains: np.ndarray, owned_cells: np.ndarray, opponent_cells: np.ndarray, neutral_cells: np.ndarray, @@ -109,9 +64,10 @@ def __init__( opponent_army_count: int, timestep: int, ): - self.army = army + self.armies = armies self.generals = generals - self.city = city + self.cities = cities + self.mountains = mountains self.owned_cells = owned_cells self.opponent_cells = opponent_cells self.neutral_cells = neutral_cells @@ -138,7 +94,7 @@ def action_mask(self) -> np.ndarray: height, width = self.owned_cells.shape ownership_channel = self.owned_cells - more_than_1_army = (self.army > 1) * ownership_channel + more_than_1_army = (self.armies > 1) * ownership_channel owned_cells_indices = np.argwhere(more_than_1_army) valid_action_mask = np.zeros((height, width, 4), dtype=bool) @@ -155,9 +111,9 @@ def action_mask(self) -> np.ndarray: destinations = destinations[in_first_boundary & in_height_boundary & in_width_boundary] # check if destination is road - passable_cells = self.neutral_cells + self.owned_cells + self.opponent_cells + self.city + passable_cells = 1 - self.mountains # assert that every value is either 0 or 1 in passable cells - assert np.all(np.isin(passable_cells, [0, 1])) + assert np.all(np.isin(passable_cells, [0, 1])), f"{passable_cells}" passable_cell_indices = passable_cells[destinations[:, 0], destinations[:, 1]] == 1 action_destinations = destinations[passable_cell_indices] @@ -169,9 +125,10 @@ def action_mask(self) -> np.ndarray: def as_dict(self, with_mask=True): _obs = { - "armies": self.army, + "armies": self.armies, "generals": self.generals, - "cities": self.city, + "cities": self.cities, + "mountains": self.mountains, "owned_cells": self.owned_cells, "opponent_cells": self.opponent_cells, "neutral_cells": self.neutral_cells, diff --git a/generals/remote/generalsio_client.py b/generals/remote/generalsio_client.py index 5b620f9..00bc7ac 100644 --- a/generals/remote/generalsio_client.py +++ b/generals/remote/generalsio_client.py @@ -3,7 +3,10 @@ from socketio import SimpleClient # type: ignore from generals.agents.agent import Agent -from generals.core.config import Observation +from generals.core.game import Direction +from generals.core.observation import Observation + +DIRECTIONS = [Direction.UP, Direction.DOWN, Direction.LEFT, Direction.RIGHT] class GeneralsBotError(Exception): @@ -56,7 +59,7 @@ def apply_diff(old: list[int], diff: list[int]) -> list[int]: print("All tests passed") -class GeneralsIOState: +class GeneralsIOself: def __init__(self, data: dict): self.replay_id = data["replay_id"] self.usernames = data["usernames"] @@ -77,7 +80,7 @@ def update(self, data: dict) -> None: if "stars" in data: self.stars = data["stars"] - def agent_observation(self) -> Observation: + def get_observation(self) -> "Observation": width, height = self.map[0], self.map[1] size = height * width @@ -91,27 +94,36 @@ def agent_observation(self) -> Observation: for general in self.generals: if general != -1: generals[general // width, general % width] = 1 - _observation = { - "army": armies, - "general": generals, - "city": cities, - "owned_cells": np.where(terrain == self.player_index, 1, 0), - "opponent_cells": np.where(terrain == self.opponent_index, 1, 0), - "neutral_cells": np.where(terrain == -1, 1, 0), - "visible_cells": maximum_filter(np.where(terrain == self.player_index, 1, 0), size=3), - "structures_in_fog": np.where(terrain == -4, 1, 0), - "owned_land_count": self.scores[self.player_index]["tiles"], - "owned_army_count": self.scores[self.player_index]["total"], - "opponent_land_count": self.scores[self.opponent_index]["tiles"], - "opponent_army_count": self.scores[self.opponent_index]["total"], - "is_winner": False, - "timestep": self.turn, - } - - observation = { - "observation": _observation, - } - return observation + + army = armies + owned_cells = np.where(terrain == self.player_index, 1, 0) + opponent_cells = np.where(terrain == self.opponent_index, 1, 0) + neutral_cells = np.where(terrain == -1, 1, 0) + mountain_cells = np.where(terrain == -2, 1, 0) + visible_cells = maximum_filter(np.where(terrain == self.player_index, 1, 0), size=3) + structures_in_fog = np.where(terrain == -4, 1, 0) + owned_land_count = self.scores[self.player_index]["tiles"] + owned_army_count = self.scores[self.player_index]["total"] + opponent_land_count = self.scores[self.opponent_index]["tiles"] + opponent_army_count = self.scores[self.opponent_index]["total"] + timestep = self.turn + + return Observation( + armies=army, + generals=generals, + cities=cities, + mountains=mountain_cells, + owned_cells=owned_cells, + opponent_cells=opponent_cells, + neutral_cells=neutral_cells, + visible_cells=visible_cells, + structures_in_fog=structures_in_fog, + owned_land_count=owned_land_count, + owned_army_count=owned_army_count, + opponent_land_count=opponent_land_count, + opponent_army_count=opponent_army_count, + timestep=timestep, + ).as_dict() class GeneralsIOClient(SimpleClient): @@ -124,6 +136,7 @@ def __init__(self, agent: Agent, user_id: str): super().__init__() self.connect("https://botws.generals.io") self.user_id = user_id + self.agent = agent self._queue_id = "" @property @@ -178,7 +191,7 @@ def _initialize_game(self, data: dict) -> None: Triggered after server starts the game. :param data: dictionary of information received in the beginning """ - self.game_state = GeneralsIOState(data[0]) + self.game_state = GeneralsIOself(data[0]) def _play_game(self) -> None: """ @@ -193,7 +206,19 @@ def _play_game(self) -> None: match event: case "game_update": self.game_state.update(data) - self.game_state.agent_observation() + obs = self.game_state.get_observation() + # This code here should be made way prettier, its just POC + action = self.agent.act(obs) + if not action["pass"]: + source = action["cell"] + direction = DIRECTIONS[action["direction"]].value + split = action["split"] + destination = source + direction + # convert to index + source_index = source[0] * self.game_state.map[0] + source[1] + destination_index = destination[0] * self.game_state.map[0] + destination[1] + self.emit("attack", (int(source_index), int(destination_index), int(split))) + case "game_lost" | "game_won": # server sends game_lost or game_won before game_over winner = event == "game_won"