diff --git a/README.md b/README.md index d000b754c..0e177e439 100644 --- a/README.md +++ b/README.md @@ -5,13 +5,14 @@ ![Discord](https://img.shields.io/discord/1385302652014825552) [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/bcollazo/catanatron/blob/master/examples/Overview.ipynb) -Catanatron is a high-performance simulator and strong AI player for Settlers of Catan. You can run thousands of games in the order of seconds. The goal is to find the strongest Settlers of Catan bot possible. +Catanatron is a high-performance simulator and strong AI player for Settlers of Catan. You can run thousands of games in the order of seconds. The goal is to find the strongest Settlers of Catan bot possible. Get Started with the Full Documentation: https://docs.catanatron.com Join our Discord: https://discord.gg/FgFmb75TWd! ## Command Line Interface + Catanatron provides a `catanatron-play` CLI tool to run large scale simulations.

@@ -22,26 +23,30 @@ Catanatron provides a `catanatron-play` CLI tool to run large scale simulations. 1. Clone the repository: - ```bash - git clone git@github.com:bcollazo/catanatron.git - cd catanatron/ - ``` -2. Create a virtual environment (requires Python 3.11 or higher) + ```bash + git clone git@github.com:bcollazo/catanatron.git + cd catanatron/ + ``` + +2. Create a virtual environment (requires Python 3.11 or higher) + + ```bash + python -m venv venv + source ./venv/bin/activate + # ./venv/Scripts/Activate.ps1 (on windows) + ``` - ```bash - python -m venv venv - source ./venv/bin/activate - ``` 3. Install dependencies - ```bash - pip install -e . - ``` -4. (Optional) Install developer and advanced dependencies + ```bash + pip install -e . + ``` + +4. (Optional) Install developer and advanced dependencies - ```bash - pip install -e ".[web,gym,dev]" - ``` + ```bash + pip install -e ".[web,gym,dev]" + ``` ### Usage @@ -52,13 +57,13 @@ catanatron-play --players=R,R,R,W --num=100 ``` Generate datasets from the games to analyze: + ```bash catanatron-play --num 100 --output my-data-path/ --output-format json ``` See more examples at https://docs.catanatron.com. - ## Graphical User Interface We provide Docker images so that you can watch, inspect, and play games against Catanatron via a web UI! @@ -67,15 +72,15 @@ We provide Docker images so that you can watch, inspect, and play games against

- ### Installation 1. Ensure you have Docker installed (https://docs.docker.com/engine/install/) 2. Run the `docker-compose.yaml` in the root folder of the repo: - ```bash - docker compose up - ``` + ```bash + docker compose up + ``` + 3. Visit http://localhost:3000 in your browser! ## Python Library @@ -100,14 +105,17 @@ print(game.play()) # returns winning color See more at http://docs.catanatron.com ## Gymnasium Interface + For Reinforcement Learning, catanatron provides an Open AI / Gymnasium Environment. Install it with: + ```bash pip install -e .[gym] ``` and use it like: + ```python import random import gymnasium @@ -128,8 +136,8 @@ env.close() See more at: https://docs.catanatron.com - ## Documentation + Full documentation here: https://docs.catanatron.com ## Contributing @@ -144,6 +152,5 @@ coverage run --source=catanatron -m pytest tests/ && coverage report See more at: https://docs.catanatron.com ## Appendix -See the motivation of the project here: [5 Ways NOT to Build a Catan AI](https://medium.com/@bcollazo2010/5-ways-not-to-build-a-catan-ai-e01bc491af17). - +See the motivation of the project here: [5 Ways NOT to Build a Catan AI](https://medium.com/@bcollazo2010/5-ways-not-to-build-a-catan-ai-e01bc491af17). diff --git a/catanatron/catanatron/cli/play.py b/catanatron/catanatron/cli/play.py index cf098f95a..7a1d83078 100644 --- a/catanatron/catanatron/cli/play.py +++ b/catanatron/catanatron/cli/play.py @@ -7,9 +7,8 @@ from rich.console import Console from rich.table import Table from rich.progress import Progress -from rich.progress import Progress, BarColumn, TimeRemainingColumn +from rich.progress import BarColumn, TimeRemainingColumn from rich import box -from rich.console import Console from rich.theme import Theme from rich.text import Text @@ -207,7 +206,7 @@ class OutputOptions: class GameConfigOptions: discard_limit: int = 7 vps_to_win: int = 10 - catan_map: Literal["BASE", "TOURNAMENT", "MINI"] = "BASE" + map_type: Literal["BASE", "TOURNAMENT", "MINI"] = "BASE" COLOR_TO_RICH_STYLE = { @@ -238,7 +237,7 @@ def play_batch_core(num_games, players, game_config, accumulators=[]): for _ in range(num_games): for player in players: player.reset_state() - catan_map = build_map(game_config.catan_map) + catan_map = build_map(game_config.map_type) game = Game( players, discard_limit=game_config.discard_limit, @@ -275,7 +274,10 @@ def play_batch( accumulators.append( CsvDataAccumulator( - output_options.output, output_options.include_board_tensor + tuple(p.color for p in players), + game_config.map_type, + output_options.output, + output_options.include_board_tensor, ) ) elif output_options.output_format == "parquet": @@ -284,7 +286,10 @@ def play_batch( accumulators.append( ParquetDataAccumulator( - output_options.output, output_options.include_board_tensor + tuple(p.color for p in players), + game_config.map_type, + output_options.output, + output_options.include_board_tensor, ) ) elif output_options.output_format == "json": diff --git a/catanatron/catanatron/features.py b/catanatron/catanatron/features.py index d74e0f891..a09567c79 100644 --- a/catanatron/catanatron/features.py +++ b/catanatron/catanatron/features.py @@ -12,7 +12,7 @@ ) from catanatron.models.board import STATIC_GRAPH, get_edges, get_node_distances from catanatron.models.map import NUM_TILES, CatanMap, build_map, number_probability -from catanatron.models.player import Player, Color, SimplePlayer +from catanatron.models.player import Color, SimplePlayer from catanatron.models.enums import ( DEVELOPMENT_CARDS, RESOURCES, @@ -106,7 +106,7 @@ def resource_hand_features(game: Game, p0_color: Color): ] for card in DEVELOPMENT_CARDS: features[f"P0_{card}_IN_HAND"] = player_state[key + f"_{card}_IN_HAND"] - features[f"P0_HAS_PLAYED_DEVELOPMENT_CARD_IN_TURN"] = player_state[ + features["P0_HAS_PLAYED_DEVELOPMENT_CARD_IN_TURN"] = player_state[ key + "_HAS_PLAYED_DEVELOPMENT_CARD_IN_TURN" ] @@ -132,7 +132,7 @@ def map_tile_features(catan_map: CatanMap, robber_coordinate): for tile_id, tile in catan_map.tiles_by_id.items(): for resource in RESOURCES: features[f"TILE{tile_id}_IS_{resource}"] = tile.resource == resource - features[f"TILE{tile_id}_IS_DESERT"] = tile.resource == None + features[f"TILE{tile_id}_IS_DESERT"] = tile.resource is None features[f"TILE{tile_id}_PROBA"] = ( 0 if tile.resource is None else number_probability(tile.number) ) diff --git a/catanatron/catanatron/gym/accumulators.py b/catanatron/catanatron/gym/accumulators.py index fd9e63212..a7116cff1 100644 --- a/catanatron/catanatron/gym/accumulators.py +++ b/catanatron/catanatron/gym/accumulators.py @@ -1,15 +1,16 @@ import os -from collections import defaultdict import time +from collections import defaultdict +from typing import Tuple, Literal -from catanatron.utils import format_secs import numpy as np import pandas as pd +from catanatron import Action, Color, Game from catanatron.features import create_sample from catanatron.game import GameAccumulator from catanatron.gym.board_tensor_features import create_board_tensor -from catanatron.gym.envs.catanatron_env import to_action_space, to_action_type_space +from catanatron.gym.envs.action_space import to_action_space, to_action_type_space from catanatron.gym.utils import ( DISCOUNT_FACTOR, get_tournament_total_return, @@ -17,11 +18,14 @@ populate_matrices, simple_total_return, ) +from catanatron.utils import format_secs class ReinforcementLearningAccumulator(GameAccumulator): def __init__( self, + player_colors: Tuple[Color], + map_type: Literal["BASE", "TOURNAMENT", "MINI"] = "BASE", include_board_tensor=True, total_return_fns={ "RETURN": simple_total_return, @@ -29,6 +33,8 @@ def __init__( "VICTORY_POINTS_RETURN": get_victory_points_total_return, }, ): + self.player_colors = player_colors + self.map_type = map_type self.include_board_tensor = include_board_tensor # TODO: Generalize to "rewards_fn" that can yield intermediary rewards # while still rewarding big on terminal states. @@ -45,14 +51,17 @@ def before(self, game): if self.include_board_tensor: self.data["board_tensors"] = [] - def step(self, game_before_action, action): + def step(self, game_before_action: Game, action: Action): self.data["color_action_indices"][action.color].append( len(self.data["samples"]) ) self.data["acting_color"].append(action.color) self.data["samples"].append(create_sample(game_before_action, action.color)) self.data["actions"].append( - [to_action_space(action), to_action_type_space(action.action_type)] + [ + to_action_space(action, self.player_colors, self.map_type), + to_action_type_space(action.action_type), + ] ) if self.include_board_tensor: @@ -130,8 +139,14 @@ def after(self, game): class CsvDataAccumulator(ReinforcementLearningAccumulator): - def __init__(self, output, include_board_tensor=True): - super().__init__(include_board_tensor) + def __init__( + self, + player_colors: Tuple[Color], + map_type: Literal["BASE", "TOURNAMENT", "MINI"], + output, + include_board_tensor=True, + ): + super().__init__(player_colors, map_type, include_board_tensor) self.output = output def after(self, game): @@ -164,8 +179,14 @@ def after(self, game): class ParquetDataAccumulator(ReinforcementLearningAccumulator): - def __init__(self, output, include_board_tensor=True): - super().__init__(include_board_tensor) + def __init__( + self, + player_colors: Tuple[Color], + map_type: Literal["BASE", "TOURNAMENT", "MINI"], + output, + include_board_tensor=True, + ): + super().__init__(player_colors, map_type, include_board_tensor) self.output = output def after(self, game): diff --git a/catanatron/catanatron/gym/envs/action_space.py b/catanatron/catanatron/gym/envs/action_space.py new file mode 100644 index 000000000..e617a087a --- /dev/null +++ b/catanatron/catanatron/gym/envs/action_space.py @@ -0,0 +1,94 @@ +from functools import lru_cache +from typing import Tuple, Literal + +from catanatron.models.actions import Action +from catanatron.models.board import get_edges +from catanatron.models.enums import RESOURCES, ActionType +from catanatron.models.player import Color +from catanatron.models.map import build_map + + +@lru_cache(maxsize=None) +def get_action_array( + player_colors: Tuple[Color], map_type: Literal["BASE", "TOURNAMENT", "MINI"] +): + catan_map = build_map(map_type) + num_nodes = len(catan_map.land_nodes) + actions_array = [ + (ActionType.ROLL, None), + (ActionType.DISCARD, None), + *[ + (ActionType.BUILD_ROAD, tuple(sorted(edge))) + for edge in get_edges(catan_map.land_nodes) + ], + *[(ActionType.BUILD_SETTLEMENT, node_id) for node_id in range(num_nodes)], + *[(ActionType.BUILD_CITY, node_id) for node_id in range(num_nodes)], + (ActionType.BUY_DEVELOPMENT_CARD, None), + (ActionType.PLAY_KNIGHT_CARD, None), + *[ + (ActionType.PLAY_YEAR_OF_PLENTY, (first_card, RESOURCES[j])) + for i, first_card in enumerate(RESOURCES) + for j in range(i, len(RESOURCES)) + ], + *[(ActionType.PLAY_YEAR_OF_PLENTY, (first_card,)) for first_card in RESOURCES], + (ActionType.PLAY_ROAD_BUILDING, None), + *[(ActionType.PLAY_MONOPOLY, r) for r in RESOURCES], + # Move Robber actions include to every tile and from each opponent + *[ + (ActionType.MOVE_ROBBER, (coordinates, victim_color)) + for coordinates in catan_map.land_tiles.keys() + for victim_color in [None] + list(player_colors) + ], + # 4:1 with bank + *[ + (ActionType.MARITIME_TRADE, tuple(4 * [i] + [j])) + for i in RESOURCES + for j in RESOURCES + if i != j + ], + # 3:1 with port + *[ + (ActionType.MARITIME_TRADE, tuple(3 * [i] + [None, j])) # type: ignore + for i in RESOURCES + for j in RESOURCES + if i != j + ], + # 2:1 with port + *[ + (ActionType.MARITIME_TRADE, tuple(2 * [i] + [None, None, j])) # type: ignore + for i in RESOURCES + for j in RESOURCES + if i != j + ], + (ActionType.END_TURN, None), + ] + return actions_array + + +ACTION_TYPES = [i for i in ActionType] + + +def to_action_type_space(action_type: ActionType) -> int: + return ACTION_TYPES.index(action_type) + + +def to_action_space( + action: Action, + player_colors: Tuple[Color], + map_type: Literal["BASE", "TOURNAMENT", "MINI"], +): + """maps action to space_action equivalent integer""" + actions_array = get_action_array(player_colors, map_type) + return actions_array.index((action.action_type, action.value)) + + +def from_action_space( + action_int, + color: Color, + player_colors: Tuple[Color], + map_type: Literal["BASE", "TOURNAMENT", "MINI"], +): + """maps action_int to catantron.models.actions.Action""" + actions_array = get_action_array(player_colors, map_type) + (action_type, value) = actions_array[action_int] + return Action(color, action_type, value) diff --git a/catanatron/catanatron/gym/envs/catanatron_env.py b/catanatron/catanatron/gym/envs/catanatron_env.py index a1f43db93..7c6e17d08 100644 --- a/catanatron/catanatron/gym/envs/catanatron_env.py +++ b/catanatron/catanatron/gym/envs/catanatron_env.py @@ -5,13 +5,16 @@ from catanatron.game import Game, TURNS_LIMIT from catanatron.models.player import Color, Player, RandomPlayer -from catanatron.models.map import BASE_MAP_TEMPLATE, NUM_NODES, LandTile, build_map -from catanatron.models.enums import RESOURCES, Action, ActionType -from catanatron.models.board import get_edges +from catanatron.models.map import build_map from catanatron.features import ( create_sample, get_feature_ordering, ) +from catanatron.gym.envs.action_space import ( + to_action_space, + from_action_space, + get_action_array, +) from catanatron.gym.board_tensor_features import ( create_board_tensor, get_channels, @@ -19,94 +22,6 @@ ) -BASE_TOPOLOGY = BASE_MAP_TEMPLATE.topology -TILE_COORDINATES = [x for x, y in BASE_TOPOLOGY.items() if y == LandTile] -ACTIONS_ARRAY = [ - (ActionType.ROLL, None), - # TODO: One for each tile (and abuse 1v1 setting). - *[(ActionType.MOVE_ROBBER, tile) for tile in TILE_COORDINATES], - (ActionType.DISCARD, None), - *[(ActionType.BUILD_ROAD, tuple(sorted(edge))) for edge in get_edges()], - *[(ActionType.BUILD_SETTLEMENT, node_id) for node_id in range(NUM_NODES)], - *[(ActionType.BUILD_CITY, node_id) for node_id in range(NUM_NODES)], - (ActionType.BUY_DEVELOPMENT_CARD, None), - (ActionType.PLAY_KNIGHT_CARD, None), - *[ - (ActionType.PLAY_YEAR_OF_PLENTY, (first_card, RESOURCES[j])) - for i, first_card in enumerate(RESOURCES) - for j in range(i, len(RESOURCES)) - ], - *[(ActionType.PLAY_YEAR_OF_PLENTY, (first_card,)) for first_card in RESOURCES], - (ActionType.PLAY_ROAD_BUILDING, None), - *[(ActionType.PLAY_MONOPOLY, r) for r in RESOURCES], - # 4:1 with bank - *[ - (ActionType.MARITIME_TRADE, tuple(4 * [i] + [j])) - for i in RESOURCES - for j in RESOURCES - if i != j - ], - # 3:1 with port - *[ - (ActionType.MARITIME_TRADE, tuple(3 * [i] + [None, j])) # type: ignore - for i in RESOURCES - for j in RESOURCES - if i != j - ], - # 2:1 with port - *[ - (ActionType.MARITIME_TRADE, tuple(2 * [i] + [None, None, j])) # type: ignore - for i in RESOURCES - for j in RESOURCES - if i != j - ], - (ActionType.END_TURN, None), -] -ACTION_SPACE_SIZE = len(ACTIONS_ARRAY) -ACTION_TYPES = [i for i in ActionType] - - -def to_action_type_space(action_type: ActionType) -> int: - return ACTION_TYPES.index(action_type) - - -# NOTE: I think I don't need this if we separate action and action_record nicely... -def normalize_action(action): - normalized = action - if normalized.action_type == ActionType.ROLL: - return Action(action.color, action.action_type, None) - elif normalized.action_type == ActionType.MOVE_ROBBER: - return Action(action.color, action.action_type, action.value[0]) - elif normalized.action_type == ActionType.BUILD_ROAD: - return Action(action.color, action.action_type, tuple(sorted(action.value))) - elif normalized.action_type == ActionType.BUY_DEVELOPMENT_CARD: - return Action(action.color, action.action_type, None) - elif normalized.action_type == ActionType.DISCARD: - return Action(action.color, action.action_type, None) - return normalized - - -def to_action_space(action): - """maps action to space_action equivalent integer""" - normalized = normalize_action(action) - return ACTIONS_ARRAY.index((normalized.action_type, normalized.value)) - - -def from_action_space(action_int, playable_actions): - """maps action_int to catantron.models.actions.Action""" - # Get "catan_action" based on space action. - # i.e. Take first action in playable that matches ACTIONS_ARRAY blueprint - (action_type, value) = ACTIONS_ARRAY[action_int] - catan_action = None - for action in playable_actions: - normalized = normalize_action(action) - if normalized.action_type == action_type and normalized.value == value: - catan_action = action - break # return the first one - assert catan_action is not None - return catan_action - - FEATURES = get_feature_ordering(num_players=2) NUM_FEATURES = len(FEATURES) @@ -114,7 +29,7 @@ def from_action_space(action_int, playable_actions): HIGH = 19 * 5 -def simple_reward(game, p0_color): +def simple_reward(action, game, p0_color): winning_color = game.winning_color() if p0_color == winning_color: return 1 @@ -133,16 +48,20 @@ class CatanatronEnv(gym.Env): metadata = {"render_modes": []} def __init__(self, config=None): + self.dtype = np.float32 self.config = config or dict() self.invalid_action_reward = self.config.get("invalid_action_reward", -1) self.reward_function = self.config.get("reward_function", simple_reward) self.map_type = self.config.get("map_type", "BASE") self.vps_to_win = self.config.get("vps_to_win", 10) - self.enemies = self.config.get("enemies", [RandomPlayer(Color.RED)]) + self.representation = self.config.get("representation", "vector") + assert self.representation in ["mixed", "vector"] + self.enemies = self.config.get("enemies", [RandomPlayer(Color.RED)]) + self.player_colors = tuple([Color.BLUE] + [p.color for p in self.enemies]) assert all(p.color != Color.BLUE for p in self.enemies) - assert self.representation in ["mixed", "vector"] + self.p0 = Player(Color.BLUE) self.players = [self.p0] + self.enemies # type: ignore self.representation = "mixed" if self.representation == "mixed" else "vector" @@ -150,20 +69,22 @@ def __init__(self, config=None): self.invalid_actions_count = 0 self.max_invalid_actions = 10 - # TODO: Make self.action_space tighter if possible (per map_type) - self.action_space = spaces.Discrete(ACTION_SPACE_SIZE) + # Build action space depending on map type + self.action_array = get_action_array(self.player_colors, self.map_type) + self.action_space_size = len(self.action_array) + self.action_space = spaces.Discrete(self.action_space_size) if self.representation == "mixed": channels = get_channels(len(self.players)) board_tensor_space = spaces.Box( - low=0, high=1, shape=(channels, 21, 11), dtype=np.float64 + low=0, high=1, shape=(channels, 21, 11), dtype=self.dtype ) self.numeric_features = [ f for f in self.features if not is_graph_feature(f) ] # TODO: This could be tigher (e.g. _ROADS_AVAILABLE <= 15) numeric_space = spaces.Box( - low=0, high=HIGH, shape=(len(self.numeric_features),), dtype=np.float64 + low=0, high=HIGH, shape=(len(self.numeric_features),), dtype=self.dtype ) mixed = spaces.Dict( { @@ -175,7 +96,7 @@ def __init__(self, config=None): else: # TODO: This could be tigher (e.g. _ROADS_AVAILABLE <= 15) self.observation_space = spaces.Box( - low=0, high=HIGH, shape=(len(self.features),), dtype=np.float64 + low=0, high=HIGH, shape=(len(self.features),), dtype=self.dtype ) self.reset() @@ -185,20 +106,33 @@ def get_valid_actions(self): Returns: List[int]: valid actions """ - return list(map(to_action_space, self.game.playable_actions)) + return [ + to_action_space(a, self.player_colors, self.map_type) + for a in self.game.playable_actions + ] + + def action_masks(self) -> list[bool]: + """ + This method is to be compatible with SB3 SubprocVecEnv. + See https://sb3-contrib.readthedocs.io/en/master/modules/ppo_mask.html + + Returns: + List[bool]: action masks + """ + valid = set(self.get_valid_actions()) + return [action_int in valid for action_int in range(self.action_space_size)] def step(self, action): try: - catan_action = from_action_space(action, self.game.playable_actions) - except Exception as e: + catan_action = from_action_space( + action, self.p0.color, self.player_colors, self.map_type + ) + assert catan_action in self.game.playable_actions + except AssertionError: self.invalid_actions_count += 1 observation = self._get_observation() winning_color = self.game.winning_color() - done = ( - winning_color is not None - or self.invalid_actions_count > self.max_invalid_actions - ) terminated = winning_color is not None truncated = ( self.invalid_actions_count > self.max_invalid_actions @@ -216,7 +150,7 @@ def step(self, action): winning_color = self.game.winning_color() terminated = winning_color is not None truncated = self.game.state.num_turns >= TURNS_LIMIT - reward = self.reward_function(self.game, self.p0.color) + reward = self.reward_function(catan_action, self.game, self.p0.color) return observation, reward, terminated, truncated, info @@ -251,10 +185,12 @@ def _get_observation(self) -> Union[np.ndarray, MixedObservation]: board_tensor = create_board_tensor( self.game, self.p0.color, channels_first=True ) - numeric = np.array([float(sample[i]) for i in self.numeric_features]) + numeric = np.array( + [sample[i] for i in self.numeric_features], dtype=self.dtype + ) return {"board": board_tensor, "numeric": numeric} - return np.array([float(sample[i]) for i in self.features]) + return np.array([sample[i] for i in self.features], dtype=self.dtype) def _advance_until_p0_decision(self): while ( @@ -264,7 +200,7 @@ def _advance_until_p0_decision(self): self.game.play_tick() # will play bot -CatanatronEnv.__doc__ = f""" +CatanatronEnv.__doc__ = """ 1v1 environment against a random player Attributes: @@ -295,7 +231,7 @@ def _advance_until_p0_decision(self): * - Integer - Catanatron Action """ -for i, v in enumerate(ACTIONS_ARRAY): +for i, v in enumerate(get_action_array((Color.BLUE, Color.RED), "BASE")): CatanatronEnv.__doc__ += f" * - {i}\n - {v}\n" CatanatronEnv.__doc__ += """ diff --git a/catanatron/catanatron/models/enums.py b/catanatron/catanatron/models/enums.py index 49c0debb2..a129cd458 100644 --- a/catanatron/catanatron/models/enums.py +++ b/catanatron/catanatron/models/enums.py @@ -74,7 +74,9 @@ class ActionType(Enum): ROLL = "ROLL" # value is None MOVE_ROBBER = "MOVE_ROBBER" # value is (coordinate, Color|None). - DISCARD = "DISCARD" # value is None|Resource[]. TODO: Should always be Resource[]. + + # TODO: None for now to avoid complexity, but should be Resource[]. + DISCARD = "DISCARD" # value is None # Building/Buying BUILD_ROAD = "BUILD_ROAD" # value is edge_id diff --git a/catanatron/catanatron/models/map.py b/catanatron/catanatron/models/map.py index 2219bac8c..6d8c9a3fc 100644 --- a/catanatron/catanatron/models/map.py +++ b/catanatron/catanatron/models/map.py @@ -367,7 +367,7 @@ def initialize_tiles( port_autoinc += 1 elif tile_type == LandTile: resource = shuffled_tile_resources.pop() - if resource != None: + if resource is not None: number = shuffled_numbers.pop() tile = LandTile(tile_autoinc, resource, number, nodes, edges) else: diff --git a/catanatron_experimental/catanatron_experimental/machine_learning/players/reinforcement.py b/catanatron_experimental/catanatron_experimental/machine_learning/players/reinforcement.py index 705d52ac2..5b8d9fde4 100644 --- a/catanatron_experimental/catanatron_experimental/machine_learning/players/reinforcement.py +++ b/catanatron_experimental/catanatron_experimental/machine_learning/players/reinforcement.py @@ -4,19 +4,24 @@ import tensorflow as tf from tensorflow import keras -from catanatron.models.player import Player -from catanatron.models.enums import Action, ActionType +from catanatron import Color, Player from catanatron.features import ( create_sample, create_sample_vector, get_feature_ordering, ) -from catanatron.gym.envs.catanatron_env import ACTIONS_ARRAY, ACTION_SPACE_SIZE +from catanatron.gym.envs.action_space import ( + get_action_array, +) from catanatron.gym.board_tensor_features import ( NUMERIC_FEATURES, create_board_tensor, ) + +ACTIONS_ARRAY = get_action_array((Color.BLUE, Color.RED), "BASE") +ACTION_SPACE_SIZE = len(ACTIONS_ARRAY) + # from catanatron_experimental.rep_b_model import build_model # Taken from correlation analysis @@ -94,6 +99,8 @@ def p_model_path(version): def get_v_model(model_path): global V_MODEL if V_MODEL is None: + import autokeras as ak + custom_objects = None if model_path[:2] != "ak" else ak.CUSTOM_OBJECTS V_MODEL = keras.models.load_model(model_path, custom_objects=custom_objects) return V_MODEL @@ -108,29 +115,12 @@ def get_t_model(model_path): def hot_one_encode_action(action): - normalized = normalize_action(action) - index = ACTIONS_ARRAY.index((normalized.action_type, normalized.value)) + index = ACTIONS_ARRAY.index((action.action_type, action.value)) vector = np.zeros(ACTION_SPACE_SIZE, dtype=int) vector[index] = 1 return vector -def normalize_action(action): - normalized = action - if normalized.action_type == ActionType.ROLL: - return Action(action.color, action.action_type, None) - elif normalized.action_type == ActionType.MOVE_ROBBER: - return Action(action.color, action.action_type, action.value[0]) - elif normalized.action_type == ActionType.BUILD_ROAD: - return Action(action.color, action.action_type, tuple(sorted(action.value))) - elif normalized.action_type == ActionType.BUY_DEVELOPMENT_CARD: - return Action(action.color, action.action_type, None) - elif normalized.action_type == ActionType.DISCARD: - return Action(action.color, action.action_type, None) - - return normalized - - class PRLPlayer(Player): def __init__(self, color, model_path): super(PRLPlayer, self).__init__(color) @@ -147,8 +137,7 @@ def decide(self, game, playable_actions): # return playable_actions[index] # Create array like [0,0,1,0,0,0,1,...] representing possible actions - normalized_playable = [normalize_action(a) for a in playable_actions] - possibilities = [(a.action_type, a.value) for a in normalized_playable] + possibilities = [(a.action_type, a.value) for a in playable_actions] possible_indices = [ACTIONS_ARRAY.index(x) for x in possibilities] mask = np.zeros(ACTION_SPACE_SIZE, dtype=np.int) mask[possible_indices] = 1 diff --git a/examples/ppo/.gitignore b/examples/ppo/.gitignore new file mode 100644 index 000000000..469e870b1 --- /dev/null +++ b/examples/ppo/.gitignore @@ -0,0 +1,2 @@ +tensorboard_logs +checkpoints \ No newline at end of file diff --git a/examples/ppo/shaped_reward.py b/examples/ppo/shaped_reward.py new file mode 100644 index 000000000..2aea336d6 --- /dev/null +++ b/examples/ppo/shaped_reward.py @@ -0,0 +1,94 @@ +""" +Shaped reward function for Catan that provides incremental rewards. + +Instead of only rewarding at the end (+1 win, -1 loss), this gives +partial credit for progress during the game. +""" + +from catanatron.state_functions import ( + get_actual_victory_points, + get_longest_road_color, + get_largest_army, +) + + +class ShapedRewardFunction: + """ + Reward function that gives incremental rewards for game progress. + + Rewards: + - Victory point gain: +1.0 per VP + - Winning: +10.0 bonus + - Losing: -10.0 penalty + - Longest road acquired: +0.5 + - Largest army acquired: +0.5 + """ + + def __init__(self): + self.reset() + + def reset(self): + """Reset tracked state for a new game.""" + self.prev_vp = 0 + self.prev_has_longest_road = False + self.prev_has_largest_army = False + + def __call__(self, action, game, p0_color): + """ + Compute reward for the current step. + + Args: + action: The action taken + game: The game object + p0_color: Player 0's color (BLUE) + + Returns: + float: The reward for this step + """ + state = game.state + reward = 0.0 + + # Get current metrics + current_vp = get_actual_victory_points(state, p0_color) + longest_road_color = get_longest_road_color(state) + largest_army_color, _ = get_largest_army(state) + + has_longest_road = longest_road_color == p0_color + has_largest_army = largest_army_color == p0_color + + # Reward for VP gain + vp_gain = current_vp - self.prev_vp + reward += vp_gain * 1.0 + + # Reward for acquiring longest road + if has_longest_road and not self.prev_has_longest_road: + reward += 0.5 + elif not has_longest_road and self.prev_has_longest_road: + reward -= 0.5 # Lost longest road + + # Reward for acquiring largest army + if has_largest_army and not self.prev_has_largest_army: + reward += 0.5 + elif not has_largest_army and self.prev_has_largest_army: + reward -= 0.5 # Lost largest army + + # Check for game end + winning_color = game.winning_color() + if winning_color is not None: + if p0_color == winning_color: + reward += 10.0 # Big bonus for winning + else: + reward -= 10.0 # Penalty for losing + # Reset for next game + self.reset() + else: + # Update tracked state for next step + self.prev_vp = current_vp + self.prev_has_longest_road = has_longest_road + self.prev_has_largest_army = has_largest_army + + return reward + + +# Create a singleton instance to use +shaped_reward = ShapedRewardFunction() diff --git a/examples/ppo/train.py b/examples/ppo/train.py new file mode 100644 index 000000000..2e4136a27 --- /dev/null +++ b/examples/ppo/train.py @@ -0,0 +1,308 @@ +""" +Restartable Stable Baselines3 training example with TensorBoard logging. + +Features: +- Vectorized environments (parallel games for speedup) +- VecNormalize (observation and reward normalization) +- Shaped reward function (incremental rewards for progress) +- Linear learning rate schedule (decreases over time) +- GPU support (automatic detection) +- Checkpoint saving/loading for resumable training +- TensorBoard integration for monitoring +- Wandb-ready configuration object + +Usage: + # Start new training: + python train.py + + # Resume from checkpoint: + python train.py --resume + + # Train for custom timesteps: + python train.py --timesteps 500000 + + # View TensorBoard: + tensorboard --logdir ./tensorboard_logs + + # To use with wandb (add to main()): + import wandb + wandb.init(project="catan-ppo", config=config) +""" + +import random +import argparse +import os +from pathlib import Path + +import gymnasium +import numpy as np +import torch +from sb3_contrib.common.maskable.policies import MaskableActorCriticPolicy +from sb3_contrib.common.wrappers import ActionMasker +from sb3_contrib.ppo_mask import MaskablePPO +from stable_baselines3.common.callbacks import CheckpointCallback +from stable_baselines3.common.env_util import make_vec_env +from stable_baselines3.common.vec_env import SubprocVecEnv, VecNormalize + +import catanatron.gym +from catanatron import Color +from catanatron.players.value import ValueFunctionPlayer +from catanatron.gym.envs.catanatron_env import simple_reward +from shaped_reward import ShapedRewardFunction + + +# Configuration object (compatible with wandb) +config = { + # Environment parameters + "map_type": "MINI", # Map type for Catan (BASE, MINI, etc.) + "vps_to_win": 6, # Victory points needed to win + "use_shaped_reward": True, # Use shaped vs simple reward function + # PPO hyperparameters + "n_envs": 32, # Number of parallel environments + "n_steps": 256, # Number of steps to collect before update + "batch_size": 1024, # Batch size for training + "n_epochs": 5, # Number of epochs for PPO update + "gamma": 0.99, # Discount factor for future rewards + "initial_lr": 0.01, # Initial learning rate + "final_lr": 0.001, # Final learning rate + "ent_coef": 0.01, # Entropy coefficient for exploration + # Network architecture + "num_layers": 3, # Number of hidden layers + "neurons_per_layer": 256, # Neurons in each layer + # Training parameters + "seed": 42, # Random seed + "checkpoint_freq": 10_000, # Save checkpoint every N steps +} + +# Validate configuration +assert (config["n_envs"] * config["n_steps"]) % config["batch_size"] == 0, ( + "BATCH_SIZE must divide N_ENVS * N_STEPS" +) + +# Directories +CHECKPOINT_DIR = os.path.join(os.path.dirname(__file__), "checkpoints") +TENSORBOARD_DIR = os.path.join(os.path.dirname(__file__), "tensorboard_logs") + + +def main(): + # Parse arguments + parser = argparse.ArgumentParser() + parser.add_argument("--resume", action="store_true", help="Resume from checkpoint") + parser.add_argument( + "--timesteps", + type=int, + default=100_000, + help=f"Number of timesteps to train for (default: {100_000:,})", + ) + args = parser.parse_args() + + # Set random seeds for reproducibility + random.seed(config["seed"]) + np.random.seed(config["seed"]) + torch.manual_seed(config["seed"]) + torch.cuda.manual_seed_all(config["seed"]) + torch.backends.cudnn.deterministic = True + torch.backends.cudnn.benchmark = False + + # Set device (GPU if available) + device = ( + "cuda" + if torch.cuda.is_available() + else ("mps" if torch.backends.mps.is_available() else "cpu") + ) + print(f"Using device: {device}") + + # Create vectorized environments + print( + f"Using reward function: {'shaped (incremental)' if config['use_shaped_reward'] else 'simple (sparse)'}" + ) + print(f"Using map type: {config['map_type']}, VPs to win: {config['vps_to_win']}") + print(f"Creating {config['n_envs']} parallel environments...") + env = make_vec_env( + make_catan_env, + n_envs=config["n_envs"], + seed=config["seed"], + vec_env_cls=SubprocVecEnv, # Use subprocesses for CPU-heavy environments + ) + + # Wrap with VecNormalize for observation and reward normalization + print("Wrapping environments with VecNormalize (obs + reward normalization)") + env = VecNormalize( + env, + norm_obs=True, # Normalize observations + norm_reward=False, # Normalize rewards + clip_obs=10.0, # Clip observations to [-10, 10] after normalization + clip_reward=10.0, # Clip rewards to [-10, 10] after normalization + ) + + # Load or create model + # Create directories + os.makedirs(CHECKPOINT_DIR, exist_ok=True) + os.makedirs(TENSORBOARD_DIR, exist_ok=True) + if args.resume: + checkpoint = get_latest_checkpoint() + if checkpoint: + print(f"Loading checkpoint: {checkpoint}") + + # Load VecNormalize stats if available + vec_normalize_path = checkpoint.replace(".zip", "_vecnormalize.pkl") + if os.path.exists(vec_normalize_path): + print(f"Loading VecNormalize stats from: {vec_normalize_path}") + env = VecNormalize.load(vec_normalize_path, env) + + model = MaskablePPO.load( + checkpoint, + env=env, + tensorboard_log=TENSORBOARD_DIR, + device=device, + ) + else: + print("No checkpoint found, starting fresh") + args.resume = False + + if not args.resume: + # Configure network architecture + net_arch = [config["neurons_per_layer"]] * config["num_layers"] + policy_kwargs = dict(net_arch=net_arch) + + # Create learning rate schedule + lr_schedule = linear_schedule(config["initial_lr"], config["final_lr"]) + + print(f"Creating new model with architecture: {net_arch}") + print( + f"PPO config: n_steps={config['n_steps']}, batch_size={config['batch_size']}, n_epochs={config['n_epochs']}, gamma={config['gamma']}, ent_coef={config['ent_coef']}" + ) + print( + f"Learning rate schedule: {config['initial_lr']:.2e} → {config['final_lr']:.2e}" + ) + model = MaskablePPO( + MaskableActorCriticPolicy, + env, + learning_rate=lr_schedule, + n_steps=config["n_steps"], + batch_size=config["batch_size"], + gamma=config["gamma"], + n_epochs=config["n_epochs"], + ent_coef=config["ent_coef"], + verbose=1, + tensorboard_log=TENSORBOARD_DIR, + seed=config["seed"], + policy_kwargs=policy_kwargs, + device=device, + ) + + # Setup checkpoint callback (also saves VecNormalize stats) + checkpoint_callback = VecNormalizeCheckpointCallback( + save_freq=config["checkpoint_freq"], + save_path=CHECKPOINT_DIR, + name_prefix="rl_model", + ) + + # Train + print(f"\nTraining for {args.timesteps:,} timesteps") + print( + f"With {config['n_envs']} parallel environments (~{config['n_envs']}x speedup)" + ) + print(f"TensorBoard: tensorboard --logdir {TENSORBOARD_DIR}\n") + + # Create experiment name with parameters + reward_type = "shaped" if config["use_shaped_reward"] else "simple" + experiment_name = ( + f"MaskablePPO_{config['map_type']}_vp{config['vps_to_win']}_envs{config['n_envs']}_steps{config['n_steps']}_batch{config['batch_size']}_" + f"gamma{config['gamma']}_epochs{config['n_epochs']}_lr{config['initial_lr']}-{config['final_lr']}_ent{config['ent_coef']}_" + f"layers{config['num_layers']}x{config['neurons_per_layer']}_{reward_type}" + ) + + model.learn( + total_timesteps=args.timesteps, + callback=checkpoint_callback, + reset_num_timesteps=not args.resume, + tb_log_name=experiment_name, + ) + + # Save final model and VecNormalize stats + final_path = os.path.join(CHECKPOINT_DIR, "final_model.zip") + model.save(final_path) + + vec_normalize_path = os.path.join(CHECKPOINT_DIR, "final_model_vecnormalize.pkl") + env.save(vec_normalize_path) + + print(f"\nDone! Final model: {final_path}") + print(f"VecNormalize stats: {vec_normalize_path}") + + # Clean up + env.close() + + +def get_latest_checkpoint(): + """Find the most recent checkpoint.""" + checkpoint_path = Path(CHECKPOINT_DIR) + if not checkpoint_path.exists(): + return None + + checkpoints = list(checkpoint_path.glob("rl_model_*_steps.zip")) + if not checkpoints: + return None + + # Get latest by timestep number + latest = max(checkpoints, key=lambda p: int(p.stem.split("_")[2])) + return str(latest) + + +def linear_schedule(initial_value, final_value): + def schedule(progress_remaining): + # progress_remaining goes from 1.0 (start) to 0.0 (end) + return final_value + progress_remaining * (initial_value - final_value) + + return schedule + + +class VecNormalizeCheckpointCallback(CheckpointCallback): + """Custom checkpoint callback that also saves VecNormalize statistics.""" + + def _on_step(self) -> bool: + # Save model checkpoint (parent class behavior) + result = super()._on_step() + + # Also save VecNormalize stats if the model was just saved + if result and isinstance(self.model.get_env(), VecNormalize): + # Get the path of the most recently saved model + checkpoint_path = os.path.join( + self.save_path, f"{self.name_prefix}_{self.num_timesteps}_steps.zip" + ) + vec_normalize_path = checkpoint_path.replace(".zip", "_vecnormalize.pkl") + + # Save VecNormalize statistics + self.model.get_env().save(vec_normalize_path) + + return result + + +def make_catan_env(): + """Factory function to create a Catan environment for vectorization.""" + # Create fresh reward function instance for each environment + reward_fn = ShapedRewardFunction() if config["use_shaped_reward"] else simple_reward + + env = gymnasium.make( + "catanatron/Catanatron-v0", + config={ + "map_type": config["map_type"], + "vps_to_win": config["vps_to_win"], + "enemies": [ValueFunctionPlayer(Color.RED)], + "reward_function": reward_fn, + }, + ) + env = ActionMasker(env, mask_fn) + return env + + +def mask_fn(env) -> np.ndarray: + """Create action mask for valid actions.""" + valid_actions = env.unwrapped.get_valid_actions() + mask = np.zeros(env.action_space.n, dtype=np.float32) + mask[valid_actions] = 1 + return np.array([bool(i) for i in mask]) + + +if __name__ == "__main__": + main() diff --git a/tests/test_gym.py b/tests/test_gym.py index f7b5dfbe6..f2790f951 100644 --- a/tests/test_gym.py +++ b/tests/test_gym.py @@ -6,8 +6,14 @@ from catanatron.features import get_feature_ordering from catanatron.models.player import Color, RandomPlayer +from catanatron.models.enums import Action, ActionType, WHEAT, SHEEP, ORE from catanatron.players.value import ValueFunctionPlayer from catanatron.gym.envs.catanatron_env import CatanatronEnv +from catanatron.gym.envs.action_space import ( + get_action_array, + to_action_space, + from_action_space, +) features = get_feature_ordering(2) @@ -84,7 +90,7 @@ def test_invalid_action_reward(): def test_custom_reward(): - def custom_reward(game, p0_color): + def custom_reward(action, game, p0_color): return 123 env = gymnasium.make( @@ -139,3 +145,89 @@ def test_mixed_rep(): observation, info = env.reset() assert "board" in observation assert "numeric" in observation + + +def test_move_robber_action_in_base_action_array(): + """Test that a specific MOVE_ROBBER action is in the BASE action array for 2 players.""" + player_colors = (Color.BLUE, Color.RED) + action_array = get_action_array(player_colors, "BASE") + target_action = (ActionType.MOVE_ROBBER, ((-1, 0, 1), Color.BLUE)) + assert target_action in action_array, ( + f"Action {target_action} not found in BASE action array for 2 players" + ) + + target_action = (ActionType.MOVE_ROBBER, ((-1, 0, 1), None)) + assert target_action in action_array, ( + f"Action {target_action} not found in BASE action array for 2 players" + ) + + +def test_there_are_54_build_nodes_in_base(): + player_colors = (Color.BLUE, Color.RED) + action_array = get_action_array(player_colors, "BASE") + num_build_nodes = len( + [action for action in action_array if action[0] == ActionType.BUILD_SETTLEMENT] + ) + assert num_build_nodes == 54 + + +def test_there_are_less_build_nodes_in_mini(): + player_colors = (Color.BLUE, Color.RED) + action_array = get_action_array(player_colors, "MINI") + num_build_nodes = len( + [action for action in action_array if action[0] == ActionType.BUILD_SETTLEMENT] + ) + assert num_build_nodes == 24 + + +def test_outside_tiles_not_in_mini(): + player_colors = (Color.BLUE, Color.RED) + action_array = get_action_array(player_colors, "MINI") + target_action = (ActionType.MOVE_ROBBER, ((0, 2, -2), Color.BLUE)) + assert target_action not in action_array, ( + f"Action {target_action} found in MINI action array for 2 players" + ) + + +def test_action_space_conversion_roundtrip(): + """Test converting actions to action space integers and back.""" + player_colors = (Color.BLUE, Color.RED) + map_type = "BASE" + + # Create various test actions + # Note: For PLAY_YEAR_OF_PLENTY with 2 resources, they must be in RESOURCES order + # RESOURCES = ['WOOD', 'BRICK', 'SHEEP', 'WHEAT', 'ORE'] + test_actions = [ + Action(Color.BLUE, ActionType.ROLL, None), + Action(Color.BLUE, ActionType.DISCARD, None), + Action(Color.BLUE, ActionType.BUILD_SETTLEMENT, 10), + Action(Color.BLUE, ActionType.BUILD_CITY, 5), + Action(Color.BLUE, ActionType.BUILD_ROAD, (0, 1)), + Action(Color.BLUE, ActionType.BUY_DEVELOPMENT_CARD, None), + Action(Color.BLUE, ActionType.PLAY_KNIGHT_CARD, None), + Action(Color.BLUE, ActionType.PLAY_YEAR_OF_PLENTY, (SHEEP, WHEAT)), + Action(Color.BLUE, ActionType.PLAY_YEAR_OF_PLENTY, (WHEAT, ORE)), + Action(Color.BLUE, ActionType.PLAY_YEAR_OF_PLENTY, (ORE,)), + Action(Color.BLUE, ActionType.PLAY_MONOPOLY, WHEAT), + Action(Color.BLUE, ActionType.PLAY_ROAD_BUILDING, None), + Action(Color.BLUE, ActionType.MOVE_ROBBER, ((-1, 0, 1), Color.RED)), + Action(Color.BLUE, ActionType.MOVE_ROBBER, ((0, -1, 1), None)), + Action(Color.BLUE, ActionType.MARITIME_TRADE, (WHEAT, WHEAT, WHEAT, WHEAT, ORE)), + Action(Color.BLUE, ActionType.MARITIME_TRADE, (SHEEP, SHEEP, SHEEP, None, WHEAT)), + Action(Color.BLUE, ActionType.MARITIME_TRADE, (ORE, ORE, None, None, WHEAT)), + Action(Color.BLUE, ActionType.END_TURN, None), + ] + + for action in test_actions: + # Convert to action space integer + action_int = to_action_space(action, player_colors, map_type) + + # Convert back from action space + recovered_action = from_action_space( + action_int, action.color, player_colors, map_type + ) + + # Assert they are the same + assert recovered_action == action, ( + f"Action conversion failed: {action} -> {action_int} -> {recovered_action}" + )