feat: First game between human and bot made by this repo!

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()

generals/core/game.py

@@ -1,4 +1,3 @@
-import warnings
 from typing import Any
 
 import gymnasium as gym

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,

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"