commit 4eb34bfeeb6265aa2a15fbdd356ee3a2890def36 Author: mattlamb227@gmail.com Date: Wed Aug 5 16:42:57 2026 -0400 initial commit diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 0000000..692b197 --- /dev/null +++ b/PLAN.md @@ -0,0 +1,457 @@ +# Blokus Gym Harness — Design & Implementation Plan + +## 1. Overview + +A Gymnasium-compatible RL environment for the board game **Blokus**, supporting both **single-agent** (agent vs. configurable bots) and **multi-agent** (PettingZoo-compatible) modes. The harness uses **numpy** for maximum training speed and broad framework compatibility, with **matplotlib** as an optional rendering backend. + +## 2. File Structure + +``` +blokus/ +├── pyproject.toml # Package config, dependencies, entry points +├── README.md +├── LICENSE +├── PLAN.md # This file +├── src/blokus_gym/ +│ ├── __init__.py # Package exports +│ ├── envs/ +│ │ ├── __init__.py # Env exports +│ │ ├── blokus_env.py # Main gymnasium.Env implementation +│ │ ├── multiagent.py # PettingZoo-compatible multi-agent wrapper +│ │ └── registration.py # gymnasium.register() calls for presets +│ ├── core/ +│ │ ├── __init__.py +│ │ ├── pieces.py # Piece definitions, orientation generation, custom piece sets +│ │ ├── board.py # Board state (numpy array), placement validation, corner tracking +│ │ ├── game.py # Game rules: turn order, move validation, scoring, game-over detection +│ │ └── bots.py # Bot implementations: Random, Greedy, Minimax +│ ├── wrappers/ +│ │ ├── __init__.py +│ │ └── action_mask.py # Optional wrapper to expose action masks as part of observation +│ └── utils/ +│ ├── __init__.py +│ └── render.py # Text/ANSI and matplotlib rendering +├── tests/ +│ ├── __init__.py +│ ├── conftest.py # Shared fixtures +│ ├── test_pieces.py +│ ├── test_board.py +│ ├── test_game.py +│ ├── test_env.py +│ ├── test_multiagent.py +│ └── test_registration.py +└── examples/ + ├── basic_usage.py + ├── train_sb3.py # Stable-Baselines3 training example + ├── train_rllib.py # Ray RLlib training example + └── multiagent_example.py +``` + +## 3. Core Design Decisions + +### 3.1 Action Space: Discrete + Action Mask + +**Design**: Pre-compute all unique `(piece, orientation, position)` combinations into a flat `Discrete(N)` action space. At each step, provide an action mask in `info["action_mask"]` indicating which of the N actions are currently valid. + +**Why**: +- Works with any RL algorithm (masking-aware libs like SB3, RLlib use the mask; others can ignore it) +- Simple, standard API +- Action index is stable across episodes (important for policy networks) + +**Action space size** (standard 20×20 board, 21 pieces): ~25,000 discrete actions. Smaller boards scale down proportionally. + +**Action generation algorithm** (at `__init__` time): +1. For each piece in the piece set, generate all unique orientations (rotations × flips, deduplicated) +2. For each orientation, enumerate all positions where the piece fits on the board +3. Assign each `(piece, orientation, x, y)` a unique integer index +4. Store as a list of `Move` objects (piece_id, orientation_id, x, y) for O(1) lookup + +**Action mask computation** (at each `step`): +- Start with all-False mask +- For each action index, check if the move is valid (piece available, in bounds, no overlap, corner rule satisfied) +- Set valid indices to True + +### 3.2 Observation Space: Dict with Board, Pieces, and Corners + +```python +observation_space = spaces.Dict({ + "board": spaces.Box(0, num_players, (board_size, board_size), dtype=np.int8), + "pieces": spaces.MultiBinary(num_pieces), # 1 = still available + "my_turn": spaces.Discrete(1), # Always 1 (agent's perspective) + "corners": spaces.Box(0, board_size, (board_size, board_size), dtype=bool), # Valid corner cells +}) +``` + +**Info dict**: +```python +info = { + "action_mask": np.array([bool] * num_actions), # Valid actions + "current_player": int, + "players_with_moves": [int, ...], # Who still has valid moves + "step_count": int, +} +``` + +### 3.3 Reward Design + +**Single-agent mode** (agent is player 0): +- **Step reward**: `0` by default (sparse). Optional shaping: `+0.01 * piece_size` for each placed piece +- **Terminal reward**: Based on final score comparison + - `+1.0` if agent wins (higher score than all opponents) + - `0.0` if tie + - `-1.0` if agent loses + - Alternative: normalized score difference `agent_score / max_possible_score` + +**Multi-agent mode** (PettingZoo): +- Each agent gets `reward = placed_squares / total_squares` at game end +- Penalty for unplaced squares: `-unplaced_squares / total_squares` +- `+15/total_squares` bonus for emptying all pieces +- `+5/total_squares` bonus for last piece being monomino + +### 3.4 Dependencies + +| Dependency | Purpose | Required? | +|---|---|---| +| `numpy` | Board state, coordinate math, action masks | Yes | +| `gymnasium` | RL environment API | Yes | +| `pettingzoo` | Multi-agent wrapper | Yes (for multi-agent mode) | +| `matplotlib` | Visual rendering | No (optional) | + +**Rationale**: numpy is the universal standard for array operations in Python RL. Every framework (SB3, RLlib, Tianshou, etc.) consumes numpy arrays natively. No torch/tensorflow dependency means faster installation and broader compatibility. + +## 4. Component Designs + +### 4.1 Pieces (`core/pieces.py`) + +**Piece definition**: A piece is a set of `(x, y)` integer coordinates relative to a reference point. + +```python +@dataclass +class Piece: + name: str + squares: frozenset[tuple[int, int]] # Relative coordinates + size: int # Number of squares + +@dataclass +class OrientedPiece: + piece_id: int + orientation_id: int + squares: list[tuple[int, int]] # Absolute coordinates on board + corners: list[tuple[int, int]] # Corner cells for placement rule +``` + +**Standard piece set** (`STANDARD_PIECES`): All 21 free polyominoes (1 monomino, 1 domino, 2 trominoes, 5 tetrominoes, 12 pentominoes). + +**Orientation generation**: +1. Start with original coordinates +2. Apply 4 rotations (0°, 90°, 180°, 270°) using rotation matrix `(x, y) → (y, -x)` +3. For each rotation, apply 2 flips: original and horizontal-flip `(x, y) → (-x, y)` +4. Normalize each result (shift so min x = 0, min y = 0) +5. Deduplicate by comparing sorted coordinate sets + +**Custom piece sets**: +```python +# Option 1: Define custom pieces +custom_pieces = [ + Piece("A", frozenset([(0,0), (1,0), (0,1)])), # L-tromino + Piece("B", frozenset([(0,0), (1,0), (2,0)])), # I-tromino +] + +# Option 2: Use a preset +from blokus_gym.core.pieces import STANDARD_PIECES, DUO_PIECES, JUNIOR_PIECES + +# Option 3: Load from JSON +pieces = PieceSet.from_json("my_pieces.json") + +# Option 4: Filter standard set +pentominoes_only = [p for p in STANDARD_PIECES if p.size == 5] +``` + +### 4.2 Board (`core/board.py`) + +```python +class Board: + def __init__(self, size: int): + self.size = size + self.grid = np.zeros((size, size), dtype=np.int8) # 0 = empty, 1-4 = player + + def place(self, player_idx: int, squares: list[tuple[int, int]]) -> None + def is_empty(self, x: int, y: int) -> bool + def in_bounds(self, x: int, y: int) -> bool + def has_overlap(self, squares: list[tuple[int, int]]) -> bool + def get_player_corners(self, player_idx: int) -> set[tuple[int, int]] +``` + +**Key methods**: +- `place()`: Sets grid cells to player index +- `has_overlap()`: Checks if any square is already occupied +- `get_player_corners()`: Returns all corner cells adjacent to player's pieces (for placement rule) + +### 4.3 Game Logic (`core/game.py`) + +```python +class BlokusGame: + def __init__(self, board: Board, pieces: list[Piece], num_players: int = 4): + self.board = board + self.pieces = pieces + self.num_players = num_players + self.current_player = 0 + self.player_pieces: list[set[str]] # Available piece names per player + self.player_corners: list[set[tuple[int, int]]] + self.rounds = 0 + self.game_over = False + + def valid_move(self, player_idx: int, move: Move) -> bool + def apply_move(self, player_idx: int, move: Move) -> None + def get_valid_moves(self, player_idx: int) -> list[int] # Returns action indices + def get_action_mask(self, player_idx: int) -> np.ndarray + def next_player(self) -> int # Cycles to next player with moves + def is_game_over(self) -> bool + def get_scores(self) -> list[int] # Negative = unplaced squares + def get_winners(self) -> list[int] +``` + +**Move validation rules**: +1. Piece must be in player's available pieces +2. All squares must be in bounds +3. No overlap with existing pieces +4. **Corner rule**: At least one square of the placed piece must touch a same-color piece at a corner (diagonal adjacency) +5. **No edge adjacency**: No square of the placed piece may share an edge with a same-color piece +6. **First move**: Must be placed in a corner (if `corner_rule=True`) + +### 4.4 Bots (`core/bots.py`) + +```python +class Bot(ABC): + @abstractmethod + def select_action(self, game: BlokusGame, player_idx: int, valid_actions: np.ndarray) -> int + +class RandomBot(Bot): + """Randomly selects from valid actions.""" + +class GreedyBot(Bot): + """Selects the move that places the most squares (largest piece).""" + +class GreedyCornersBot(Bot): + """Greedy + prefers moves that open up more corners for future play.""" + +class MinimaxBot(Bot): + """Minimax search with configurable depth. Primarily for 2-player games.""" +``` + +### 4.5 Gymnasium Environment (`envs/blokus_env.py`) + +```python +class BlokusEnv(gymnasium.Env): + metadata = {"render_modes": ["human", "ansi", "rgb_array"]} + + def __init__( + self, + num_players: int = 4, + board_size: int = 20, + pieces: list[Piece] | None = None, # Default: STANDARD_PIECES + render_mode: str | None = None, + bot_type: type[Bot] = RandomBot, + bot_strength: int = 1, + reward_shaping: bool = False, + corner_rule: bool = True, + max_steps: int | None = None, + seed: int | None = None, + ): + super().__init__() + # Initialize game, spaces, action lookup table + + def reset(self, seed=None, options=None) -> tuple[dict, dict] + def step(self, action: int) -> tuple[dict, float, bool, bool, dict] + def render(self) -> str | np.ndarray | None + def close(self) +``` + +**Key implementation details**: +- `__init__` generates the full action lookup table (piece, orientation, position → index) +- `reset()` initializes a new game, places bots, optionally lets bots take first turns +- `step()` applies the agent's action, then lets bots play until it's the agent's turn again +- Returns `(observation, reward, terminated, truncated, info)` where `info` contains the action mask + +### 4.6 Multi-Agent Wrapper (`envs/multiagent.py`) + +PettingZoo-compatible wrapper that exposes all players as separate agents: + +```python +class BlokusMultiAgentEnv(AECEnv): + def __init__(self, **kwargs): + # Same config as BlokusEnv, but wraps it + + def reset(self, seed=None, options=None) -> tuple[dict, dict] + def step(self, action: int) -> None + # PettingZoo API: observe(), close(), render(), etc. + + @property + def agents(self) -> list[str] # ["player_0", "player_1", ...] + @property + def agent_selection(self) -> str +``` + +**Behavior**: +- Cycles through players in order (0, 1, 2, 3) +- Each agent gets its own observation (board + own pieces + own corners + action mask) +- Rewards are per-agent: negative for unplaced squares, bonus for emptying pieces +- Agents with no valid moves are marked as done (truncated) but still receive final rewards + +### 4.7 Registration (`envs/registration.py`) + +Predefined environment configurations registered with `gymnasium.register()`: + +```python +# Standard 4-player Blokus +"Blokus-v0": num_players=4, board_size=20, pieces=STANDARD_PIECES + +# Blokus Duo (2 players, 14x14) +"BlokusDuo-v0": num_players=2, board_size=14, pieces=STANDARD_PIECES + +# Blokus Junior (2 players, 14x14, simplified pieces) +"BlokusJunior-v0": num_players=2, board_size=14, pieces=JUNIOR_PIECES + +# Simple test env (2 players, 7x7, only pieces < 5 squares) +"BlokusSimple-v0": num_players=2, board_size=7, pieces=[p for p in STANDARD_PIECES if p.size < 5] + +# Greedy bot variants +"BlokusGreedy-v0": bot_type=GreedyBot +"BlokusDuoGreedy-v0": bot_type=GreedyBot, num_players=2, board_size=14 +``` + +### 4.8 Rendering (`utils/render.py`) + +**Text/ANSI mode** (`render_mode="human"`): +``` + 0 1 2 3 4 5 6 7 8 9 ... +0 . . . . . . . . . . +1 . . . . . . . . . . +2 . . . . . . . . . . +3 . . . . . . . . . . +4 . . . . . . . . . . +... +Player 0 (Red): I1 I2 I3 L4 O4 I4 S4 T4 I5 L5 N P T U V W X Y Z +Player 1 (Blue): I1 I2 I3 L4 O4 I4 S4 T4 I5 L5 N P T U V W X Y Z +Current player: 0 +Valid moves: 152 +``` + +**RGB array mode** (`render_mode="rgb_array"`): +- Matplotlib figure with colored squares (red, blue, yellow, green, grey) +- Grid lines +- Legend showing remaining pieces per player +- Valid move indicators (semi-transparent overlays) + +### 4.9 Action Mask Wrapper (`wrappers/action_mask.py`) + +Optional wrapper that moves the action mask from `info` into the observation space: + +```python +class ActionMaskWrapper(Wrapper): + def __init__(self, env): + super().__init__(env) + self.observation_space = spaces.Dict({ + "observation": env.observation_space, + "action_mask": spaces.Box(0, 1, (env.action_space.n,), dtype=bool), + }) + + def step(self, action): + obs, reward, terminated, truncated, info = self.env.step(action) + return { + "observation": obs, + "action_mask": info["action_mask"], + }, reward, terminated, truncated, info +``` + +## 5. Customization Options + +All options are passed as keyword arguments to `BlokusEnv()` or via `gymnasium.make()`: + +| Option | Type | Default | Description | +|---|---|---|---| +| `num_players` | int | 4 | Number of players (2-4) | +| `board_size` | int | 20 | Board dimension (standard: 20, Duo: 14) | +| `pieces` | list[Piece] | STANDARD_PIECES | Custom piece set | +| `bot_type` | type[Bot] | RandomBot | Opponent bot type | +| `bot_strength` | int | 1 | Bot search depth (for Minimax) | +| `reward_shaping` | bool | False | Enable intermediate rewards | +| `corner_rule` | bool | True | Require first move in corner | +| `max_steps` | int | None | Max steps before truncation | +| `render_mode` | str | None | "human", "ansi", "rgb_array" | +| `seed` | int | None | Random seed | +| `starting_corners` | list | [(0,0), (0,19), (19,0), (19,19)] | Custom starting corners | +| `scoring_mode` | str | "standard" | "standard", "coverage", "winloss" | +| `first_player` | int | 0 | Which player goes first (agent is always 0) | + +## 6. Implementation Phases + +### Phase 1: Core Game Logic (Days 1-3) +- [ ] `core/pieces.py` — Piece definitions, orientation generation, custom piece support +- [ ] `core/board.py` — Board state, placement validation, corner tracking +- [ ] `core/game.py` — Game rules, move validation, scoring, game-over detection +- [ ] Unit tests for all three modules + +### Phase 2: Bots (Days 4-5) +- [ ] `core/bots.py` — Random, Greedy, GreedyCorners, Minimax bots +- [ ] Tests for bot behavior + +### Phase 3: Gymnasium Environment (Days 6-8) +- [ ] `envs/blokus_env.py` — Main environment class +- [ ] Action space generation and masking +- [ ] Observation and reward design +- [ ] `envs/registration.py` — Preset registrations +- [ ] `wrappers/action_mask.py` — Optional action mask wrapper +- [ ] `gymnasium.utils.env_checker.check_env()` passes + +### Phase 4: Multi-Agent (Days 9-10) +- [ ] `envs/multiagent.py` — PettingZoo AECEnv wrapper +- [ ] Per-agent observations and rewards +- [ ] Tests for multi-agent mode + +### Phase 5: Rendering (Days 11-12) +- [ ] `utils/render.py` — Text/ANSI and matplotlib rendering +- [ ] Integration with environment + +### Phase 6: Testing & Examples (Days 13-14) +- [ ] Comprehensive test suite (unit + integration) +- [ ] `examples/basic_usage.py` — Basic usage +- [ ] `examples/train_sb3.py` — SB3 training +- [ ] `examples/train_rllib.py` — RLlib training +- [ ] `examples/multiagent_example.py` — Multi-agent +- [ ] `pyproject.toml` with all dependencies +- [ ] `README.md` with full documentation + +## 7. Testing Strategy + +### Unit Tests +- **Pieces**: Orientation generation correctness, deduplication, custom piece loading +- **Board**: Placement, overlap detection, bounds checking, corner computation +- **Game**: Move validation (all rules), scoring, game-over detection, turn cycling +- **Bots**: Random bot returns valid moves, greedy bot prefers larger pieces + +### Integration Tests +- **Environment**: `check_env()` passes, reset/step return correct types +- **Action mask**: All masked actions are valid, all valid actions are masked +- **Game flow**: Full game from start to end, correct winner detection +- **Multi-agent**: All agents receive observations, rewards, and termination signals + +### Edge Cases +- Player with no valid moves (should be skipped) +- Game ends with all players blocked +- Last piece is monomino (bonus scoring) +- Custom piece sets with unusual shapes +- Small boards where pieces don't fit + +## 8. Key Design Tradeoffs + +| Decision | Choice | Rationale | +|---|---|---| +| Gym library | Gymnasium | Maintained fork, latest API | +| Agent mode | Single + Multi-agent | Covers both use cases | +| Action space | Discrete + mask | Broadest RL framework compatibility | +| Dependencies | numpy only | Fastest, broadest compatibility | +| Rendering | Text + matplotlib | No Pygame dependency, optional | +| Piece definition | Coordinate sets | Simple, flexible for custom pieces | +| Action pre-computation | At init time | Stable action indices, fast step() | +| Multi-agent | PettingZoo AECEnv | Standard API, compatible with RLlib/PettingZoo algorithms | diff --git a/examples/basic_game.py b/examples/basic_game.py new file mode 100644 index 0000000..0f7d0e3 --- /dev/null +++ b/examples/basic_game.py @@ -0,0 +1,50 @@ +"""Basic example: play a random game of Blokus.""" +import numpy as np + +from blokus_gym import BlokusEnv + + +def main(): + # Create environment with 4 players + env = BlokusEnv(num_players=4, board_size=20, render_mode="ansi") + + obs, info = env.reset(seed=42) + print("=== Initial State ===") + print(env.render()) + + total_reward = 0 + steps = 0 + terminated = False + truncated = False + + while not (terminated or truncated): + # Select a random valid action + valid_actions = np.where(info["action_mask"])[0] + if len(valid_actions) == 0: + print("No valid moves!") + break + + action = np.random.choice(valid_actions) + obs, reward, terminated, truncated, info = env.step(action) + total_reward += reward + steps += 1 + + if steps % 10 == 0: + print(f"\n=== Step {steps} ===") + print(env.render()) + print(f"Reward: {reward}, Total: {total_reward}") + + print("\n=== Game Over ===") + print(f"Total steps: {steps}") + print(f"Total reward: {total_reward}") + print(f"Terminated: {terminated}, Truncated: {truncated}") + + # Final state + print("\n=== Final Board ===") + print(env.render()) + + env.close() + + +if __name__ == "__main__": + main() diff --git a/examples/multiagent.py b/examples/multiagent.py new file mode 100644 index 0000000..d3a4d5f --- /dev/null +++ b/examples/multiagent.py @@ -0,0 +1,46 @@ +"""Example: multi-agent self-play with BlokusMultiAgentEnv.""" +import numpy as np + +from blokus_gym import BlokusMultiAgentEnv, GreedyBot, RandomBot + + +def main(): + env = BlokusMultiAgentEnv(num_players=2, board_size=7) + + # Create bots for each player + bots = { + "player_0": RandomBot(player_idx=0, seed=42), + "player_1": GreedyBot(player_idx=1), + } + + obs, info = env.reset(seed=42) + print("=== Multi-Agent Game ===") + print(f"Agents: {env.agents}") + + steps = 0 + while True: + for agent in env.agents: + obs, info = env.last() if hasattr(env, "last") else (obs.get(agent, {}), {}) + mask = env.get_action_mask(agent) + valid_actions = np.where(mask)[0] + + if len(valid_actions) > 0: + bot = bots[agent] + action = bot.select_action(env.game, mask) + if action is not None: + obs, rewards, terminations, truncations, infos = env.step(action) + steps += 1 + + if terminations[agent]: + print(f"Game over after {steps} steps") + scores = env.game.get_scores() + for i, score in enumerate(scores): + print(f" Player {i}: {score}") + env.close() + return + + env.close() + + +if __name__ == "__main__": + main() diff --git a/examples/train_sb3.py b/examples/train_sb3.py new file mode 100644 index 0000000..5cdf4bb --- /dev/null +++ b/examples/train_sb3.py @@ -0,0 +1,54 @@ +"""Example: train an agent with Stable-Baselines3 using action masking. + +Requires: pip install blokus-gym[train] +""" +from stable_baselines3 import PPO + +from blokus_gym import ActionMaskWrapper, BlokusEnv + + +def main(): + # Create environment with ActionMaskWrapper + env = BlokusEnv(num_players=2, board_size=7, bot_type=None) + env = ActionMaskWrapper(env) + + # Train PPO agent + model = PPO( + "MultiInputPolicy", + env, + verbose=1, + learning_rate=3e-4, + n_steps=2048, + batch_size=64, + n_epochs=10, + gamma=0.99, + seed=42, + ) + + print("Training for 10,000 steps...") + model.learn(total_timesteps=10000) + + # Save the model + model.save("blokus_ppo_agent") + + # Evaluate + obs, info = env.reset(seed=42) + total_reward = 0 + steps = 0 + terminated = False + truncated = False + + while not (terminated or truncated): + action, _ = model.predict(obs, deterministic=True) + obs, reward, terminated, truncated, info = env.step(action) + total_reward += reward + steps += 1 + + print(f"Evaluation: {steps} steps, reward: {total_reward}") + + # Cleanup + env.close() + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d711891 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,65 @@ +[build-system] +requires = ["setuptools>=64", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "blokus-gym" +version = "0.1.0" +description = "A Gymnasium-compatible RL environment for the board game Blokus" +readme = "README.md" +requires-python = ">=3.10" +license = {text = "MIT"} +authors = [{name = "Blokus Gym Contributors"}] +keywords = ["reinforcement-learning", "gymnasium", "blokus", "board-game", "pettingzoo"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Science/Research", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Scientific/Engineering :: Artificial Intelligence", +] +dependencies = [ + "numpy>=1.24", + "gymnasium>=0.29", + "pettingzoo>=1.26", +] + +[project.optional-dependencies] +render = ["matplotlib>=3.7"] +dev = [ + "pytest>=7.4", + "pytest-cov>=4.1", + "black>=23.7", + "ruff>=0.1.0", + "mypy>=1.5", +] +train = [ + "stable-baselines3>=2.2", + "torch>=2.0", +] + +[project.urls] +Homepage = "https://github.com/blokus-gym/blokus-gym" +Documentation = "https://github.com/blokus-gym/blokus-gym#readme" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["test_*.py"] +addopts = "--tb=short -q" + +[tool.black] +line-length = 100 +target-version = ["py310"] + +[tool.ruff] +line-length = 100 +target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "I", "W", "UP", "B", "C4"] diff --git a/src/blokus_gym.egg-info/PKG-INFO b/src/blokus_gym.egg-info/PKG-INFO new file mode 100644 index 0000000..505c211 --- /dev/null +++ b/src/blokus_gym.egg-info/PKG-INFO @@ -0,0 +1,33 @@ +Metadata-Version: 2.4 +Name: blokus-gym +Version: 0.1.0 +Summary: A Gymnasium-compatible RL environment for the board game Blokus +Author: Blokus Gym Contributors +License: MIT +Project-URL: Homepage, https://github.com/blokus-gym/blokus-gym +Project-URL: Documentation, https://github.com/blokus-gym/blokus-gym#readme +Keywords: reinforcement-learning,gymnasium,blokus,board-game,pettingzoo +Classifier: Development Status :: 3 - Alpha +Classifier: Intended Audience :: Science/Research +Classifier: License :: OSI Approved :: MIT License +Classifier: Programming Language :: Python :: 3 +Classifier: Programming Language :: Python :: 3.10 +Classifier: Programming Language :: Python :: 3.11 +Classifier: Programming Language :: Python :: 3.12 +Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence +Requires-Python: >=3.10 +Description-Content-Type: text/markdown +Requires-Dist: numpy>=1.24 +Requires-Dist: gymnasium>=0.29 +Requires-Dist: pettingzoo>=1.26 +Provides-Extra: render +Requires-Dist: matplotlib>=3.7; extra == "render" +Provides-Extra: dev +Requires-Dist: pytest>=7.4; extra == "dev" +Requires-Dist: pytest-cov>=4.1; extra == "dev" +Requires-Dist: black>=23.7; extra == "dev" +Requires-Dist: ruff>=0.1.0; extra == "dev" +Requires-Dist: mypy>=1.5; extra == "dev" +Provides-Extra: train +Requires-Dist: stable-baselines3>=2.2; extra == "train" +Requires-Dist: torch>=2.0; extra == "train" diff --git a/src/blokus_gym.egg-info/SOURCES.txt b/src/blokus_gym.egg-info/SOURCES.txt new file mode 100644 index 0000000..e665378 --- /dev/null +++ b/src/blokus_gym.egg-info/SOURCES.txt @@ -0,0 +1,19 @@ +pyproject.toml +src/blokus_gym/__init__.py +src/blokus_gym.egg-info/PKG-INFO +src/blokus_gym.egg-info/SOURCES.txt +src/blokus_gym.egg-info/dependency_links.txt +src/blokus_gym.egg-info/requires.txt +src/blokus_gym.egg-info/top_level.txt +src/blokus_gym/core/__init__.py +src/blokus_gym/core/board.py +src/blokus_gym/core/bots.py +src/blokus_gym/core/game.py +src/blokus_gym/core/pieces.py +src/blokus_gym/envs/__init__.py +src/blokus_gym/envs/blokus_env.py +src/blokus_gym/envs/multiagent.py +src/blokus_gym/utils/__init__.py +src/blokus_gym/utils/render.py +src/blokus_gym/wrappers/__init__.py +src/blokus_gym/wrappers/action_mask.py \ No newline at end of file diff --git a/src/blokus_gym.egg-info/dependency_links.txt b/src/blokus_gym.egg-info/dependency_links.txt new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/blokus_gym.egg-info/dependency_links.txt @@ -0,0 +1 @@ + diff --git a/src/blokus_gym.egg-info/requires.txt b/src/blokus_gym.egg-info/requires.txt new file mode 100644 index 0000000..846a800 --- /dev/null +++ b/src/blokus_gym.egg-info/requires.txt @@ -0,0 +1,17 @@ +numpy>=1.24 +gymnasium>=0.29 +pettingzoo>=1.26 + +[dev] +pytest>=7.4 +pytest-cov>=4.1 +black>=23.7 +ruff>=0.1.0 +mypy>=1.5 + +[render] +matplotlib>=3.7 + +[train] +stable-baselines3>=2.2 +torch>=2.0 diff --git a/src/blokus_gym.egg-info/top_level.txt b/src/blokus_gym.egg-info/top_level.txt new file mode 100644 index 0000000..66d72bf --- /dev/null +++ b/src/blokus_gym.egg-info/top_level.txt @@ -0,0 +1 @@ +blokus_gym diff --git a/src/blokus_gym/__init__.py b/src/blokus_gym/__init__.py new file mode 100644 index 0000000..759f955 --- /dev/null +++ b/src/blokus_gym/__init__.py @@ -0,0 +1,39 @@ +from blokus_gym.core.board import Board +from blokus_gym.core.bots import Bot, GreedyBot, GreedyCornersBot, MinimaxBot, RandomBot +from blokus_gym.core.game import BlokusGame, Move +from blokus_gym.core.pieces import ( + DUO_PIECES, + JUNIOR_PIECES, + STANDARD_PIECES, + Piece, + PieceOrientation, + PieceSet, + generate_orientations, +) +from blokus_gym.envs.blokus_env import BlokusEnv +from blokus_gym.envs.multiagent import BlokusMultiAgentEnv +from blokus_gym.wrappers.action_mask import ActionMaskWrapper + +__version__ = "0.1.0" + +__all__ = [ + "BlokusEnv", + "BlokusMultiAgentEnv", + "ActionMaskWrapper", + "Board", + "BlokusGame", + "Move", + "Piece", + "PieceOrientation", + "PieceSet", + "STANDARD_PIECES", + "DUO_PIECES", + "JUNIOR_PIECES", + "generate_orientations", + "Bot", + "RandomBot", + "GreedyBot", + "GreedyCornersBot", + "MinimaxBot", + "__version__", +] diff --git a/src/blokus_gym/__pycache__/__init__.cpython-312.pyc b/src/blokus_gym/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..dddead7 Binary files /dev/null and b/src/blokus_gym/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/blokus_gym/core/__init__.py b/src/blokus_gym/core/__init__.py new file mode 100644 index 0000000..06513ef --- /dev/null +++ b/src/blokus_gym/core/__init__.py @@ -0,0 +1,30 @@ +from blokus_gym.core.board import Board +from blokus_gym.core.bots import Bot, GreedyBot, GreedyCornersBot, MinimaxBot, RandomBot +from blokus_gym.core.game import BlokusGame, Move +from blokus_gym.core.pieces import ( + DUO_PIECES, + JUNIOR_PIECES, + STANDARD_PIECES, + Piece, + PieceOrientation, + PieceSet, + generate_orientations, +) + +__all__ = [ + "Board", + "BlokusGame", + "Move", + "Piece", + "PieceOrientation", + "PieceSet", + "STANDARD_PIECES", + "DUO_PIECES", + "JUNIOR_PIECES", + "generate_orientations", + "Bot", + "RandomBot", + "GreedyBot", + "GreedyCornersBot", + "MinimaxBot", +] diff --git a/src/blokus_gym/core/__pycache__/__init__.cpython-312.pyc b/src/blokus_gym/core/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..bbbfec9 Binary files /dev/null and b/src/blokus_gym/core/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/blokus_gym/core/__pycache__/board.cpython-312.pyc b/src/blokus_gym/core/__pycache__/board.cpython-312.pyc new file mode 100644 index 0000000..c46bfac Binary files /dev/null and b/src/blokus_gym/core/__pycache__/board.cpython-312.pyc differ diff --git a/src/blokus_gym/core/__pycache__/bots.cpython-312.pyc b/src/blokus_gym/core/__pycache__/bots.cpython-312.pyc new file mode 100644 index 0000000..7cf3bbd Binary files /dev/null and b/src/blokus_gym/core/__pycache__/bots.cpython-312.pyc differ diff --git a/src/blokus_gym/core/__pycache__/game.cpython-312.pyc b/src/blokus_gym/core/__pycache__/game.cpython-312.pyc new file mode 100644 index 0000000..f633397 Binary files /dev/null and b/src/blokus_gym/core/__pycache__/game.cpython-312.pyc differ diff --git a/src/blokus_gym/core/__pycache__/pieces.cpython-312.pyc b/src/blokus_gym/core/__pycache__/pieces.cpython-312.pyc new file mode 100644 index 0000000..9b6c7d6 Binary files /dev/null and b/src/blokus_gym/core/__pycache__/pieces.cpython-312.pyc differ diff --git a/src/blokus_gym/core/board.py b/src/blokus_gym/core/board.py new file mode 100644 index 0000000..5dccba7 --- /dev/null +++ b/src/blokus_gym/core/board.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +import numpy as np + + +class Board: + """A square game board represented as a numpy array. + + Cell values: + 0 = empty + 1-4 = player index (1-based for consistency with Blokus colors) + """ + + EMPTY = 0 + + def __init__(self, size: int): + self.size = size + self.grid = np.zeros((size, size), dtype=np.int8) + + def place(self, player_idx: int, squares: list[tuple[int, int]]) -> None: + """Place a player's piece on the board.""" + for x, y in squares: + self.grid[y, x] = player_idx + + def clear(self) -> None: + """Reset the board to empty.""" + self.grid.fill(0) + + def is_empty(self, x: int, y: int) -> bool: + """Check if a cell is empty.""" + return self.grid[y, x] == 0 + + def in_bounds(self, x: int, y: int) -> bool: + """Check if coordinates are within the board.""" + return 0 <= x < self.size and 0 <= y < self.size + + def has_overlap(self, squares: list[tuple[int, int]]) -> bool: + """Check if any square in the list is already occupied.""" + for x, y in squares: + if not self.in_bounds(x, y) or self.grid[y, x] != 0: + return True + return False + + def get_cell(self, x: int, y: int) -> int: + """Get the player index at a cell (0 = empty).""" + return self.grid[y, x] + + def get_player_squares(self, player_idx: int) -> list[tuple[int, int]]: + """Get all squares occupied by a player.""" + ys, xs = np.where(self.grid == player_idx) + return [(int(x), int(y)) for x, y in zip(xs, ys, strict=True)] + + def get_player_corners(self, player_idx: int) -> set[tuple[int, int]]: + """Get all corner cells adjacent to a player's pieces. + + Corner cells are the diagonal neighbors of a player's pieces. + These are the cells where the player can place new pieces + (corner-to-corner contact rule). + """ + corners: set[tuple[int, int]] = set() + player_squares = self.get_player_squares(player_idx) + for x, y in player_squares: + for dx, dy in [(-1, -1), (1, -1), (-1, 1), (1, 1)]: + cx, cy = x + dx, y + dy + if self.in_bounds(cx, cy) and self.grid[cy, cx] == 0: + corners.add((cx, cy)) + return corners + + def get_occupied(self) -> set[tuple[int, int]]: + """Get all occupied cells.""" + ys, xs = np.where(self.grid != 0) + return {(int(x), int(y)) for x, y in zip(xs, ys, strict=True)} + + def is_full(self) -> bool: + """Check if the board is completely full.""" + return np.all(self.grid != 0) + + def coverage(self) -> float: + """Get the fraction of the board that is occupied.""" + return float(np.count_nonzero(self.grid)) / (self.size * self.size) + + def copy(self) -> Board: + """Create a deep copy of the board.""" + new_board = Board(self.size) + new_board.grid = self.grid.copy() + return new_board + + def __repr__(self) -> str: + return f"Board(size={self.size}, coverage={self.coverage():.2%})" diff --git a/src/blokus_gym/core/bots.py b/src/blokus_gym/core/bots.py new file mode 100644 index 0000000..27901b4 --- /dev/null +++ b/src/blokus_gym/core/bots.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import random +from abc import ABC, abstractmethod + +import numpy as np + +from blokus_gym.core.game import BlokusGame + + +class Bot(ABC): + """Base class for Blokus bots.""" + + def __init__(self, player_idx: int, seed: int | None = None): + self.player_idx = player_idx + self.rng = random.Random(seed) + + @abstractmethod + def select_action(self, game: BlokusGame, valid_actions: np.ndarray) -> int | None: + """Select an action from the valid actions. + + Args: + game: The current game state. + valid_actions: Boolean mask of valid actions. + + Returns: + Action index, or None if no valid moves. + """ + ... + + def __repr__(self) -> str: + return f"{self.__class__.__name__}(player={self.player_idx})" + + +class RandomBot(Bot): + """Randomly selects from valid actions.""" + + def select_action(self, game: BlokusGame, valid_actions: np.ndarray) -> int | None: + valid_indices = np.where(valid_actions)[0] + if len(valid_indices) == 0: + return None + return int(self.rng.choice(valid_indices)) + + +class GreedyBot(Bot): + """Selects the move that places the most squares (largest piece).""" + + def select_action(self, game: BlokusGame, valid_actions: np.ndarray) -> int | None: + valid_indices = np.where(valid_actions)[0] + if len(valid_indices) == 0: + return None + + best_action = -1 + best_size = -1 + for action in valid_indices: + move = game.get_move(int(action)) + piece = game.piece_set.get_piece(move.piece_id) + if piece.size > best_size: + best_size = piece.size + best_action = int(action) + + return best_action + + +class GreedyCornersBot(Bot): + """Greedy bot that prefers moves opening more corners for future play.""" + + def select_action(self, game: BlokusGame, valid_actions: np.ndarray) -> int | None: + valid_indices = np.where(valid_actions)[0] + if len(valid_indices) == 0: + return None + + best_action = -1 + best_score = float("-inf") + + for action in valid_indices: + move = game.get_move(int(action)) + piece = game.piece_set.get_piece(move.piece_id) + placed_corners = game._get_placed_corners(move) + + # Count how many new corners this move opens + new_corners = 0 + for cx, cy in placed_corners: + if game.board.in_bounds(cx, cy) and game.board.is_empty(cx, cy): + new_corners += 1 + + # Score = piece size + corner bonus + score = piece.size + new_corners * 0.5 + if score > best_score: + best_score = score + best_action = int(action) + + return best_action + + +class MinimaxBot(Bot): + """Minimax search bot (primarily for 2-player games). + + Uses a depth-limited minimax search with a simple heuristic. + """ + + def __init__(self, player_idx: int, depth: int = 2, seed: int | None = None): + super().__init__(player_idx, seed) + self.depth = depth + + def select_action(self, game: BlokusGame, valid_actions: np.ndarray) -> int | None: + valid_indices = np.where(valid_actions)[0] + if len(valid_indices) == 0: + return None + + if len(valid_indices) == 1: + return int(valid_indices[0]) + + best_action = -1 + best_value = float("-inf") + + for action in valid_indices: + if game.num_players != 2: + # For multi-player, just use greedy + return GreedyBot(self.player_idx, self.rng.randint(0, 2**31)).select_action( + game, valid_actions + ) + + cloned = game.copy() + success = cloned.play_move(self.player_idx, int(action)) + if not success: + continue + + value = self._minimax(cloned, self.depth - 1, False) + if value > best_value: + best_value = value + best_action = int(action) + + return best_action if best_action >= 0 else int(valid_indices[0]) + + def _minimax(self, game: BlokusGame, depth: int, maximizing: bool) -> float: + if depth == 0 or game.is_game_over(): + return self._evaluate(game) + + if maximizing: + value = float("-inf") + valid = game.get_valid_actions(self.player_idx) + valid_indices = np.where(valid)[0] + for action in valid_indices: + cloned = game.copy() + if cloned.play_move(self.player_idx, int(action)): + val = self._minimax(cloned, depth - 1, False) + value = max(value, val) + return value + else: + # Opponent's turn + opp_idx = 1 - self.player_idx + value = float("inf") + valid = game.get_valid_actions(opp_idx) + valid_indices = np.where(valid)[0] + for action in valid_indices: + cloned = game.copy() + if cloned.play_move(opp_idx, int(action)): + val = self._minimax(cloned, depth - 1, True) + value = min(value, val) + return value + + def _evaluate(self, game: BlokusGame) -> float: + """Simple heuristic: difference in placed squares.""" + my_score = game.players[self.player_idx].score + opp_idx = 1 - self.player_idx + opp_score = game.players[opp_idx].score if opp_idx < len(game.players) else 0 + return float(my_score - opp_score) diff --git a/src/blokus_gym/core/game.py b/src/blokus_gym/core/game.py new file mode 100644 index 0000000..27cb8ed --- /dev/null +++ b/src/blokus_gym/core/game.py @@ -0,0 +1,414 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + +import numpy as np + +from blokus_gym.core.board import Board +from blokus_gym.core.pieces import STANDARD_PIECES, Move, PieceSet + + +@dataclass +class PlayerState: + """Tracks the state of a single player in the game.""" + + idx: int # 0-based player index + available_pieces: set[str] = field(default_factory=set) # Piece names still available + score: int = 0 # Squares placed (positive = good) + corners: set[tuple[int, int]] = field(default_factory=set) # Valid placement corners + has_started: bool = False # Whether player has placed their first piece + can_move: bool = True # Whether player can still make any move + + +class BlokusGame: + """Core Blokus game logic. + + Manages the board state, player turns, move validation, and scoring. + This class is independent of the Gymnasium environment and can be + used standalone for game simulation or bot development. + """ + + # Player indices are 1-based for board representation + # (0 = empty, 1 = player 1, etc.) + PLAYER_OFFSET = 1 + + def __init__( + self, + board_size: int = 20, + pieces: PieceSet | None = None, + num_players: int = 4, + corner_rule: bool = True, + ): + self.board_size = board_size + self.piece_set = pieces or PieceSet(STANDARD_PIECES) + self.num_players = num_players + self.corner_rule = corner_rule + + self.board = Board(board_size) + self.players: list[PlayerState] = [] + self.current_player = 0 + self.rounds = 0 + self.game_over = False + + # Pre-compute all possible moves (action lookup table) + self._action_moves: list[Move] = [] + self._move_to_action: dict[tuple[int, int, int, int], int] = {} + self._generate_action_space() + + # Starting corners for each player (standard Blokus layout) + # 4 players: all four corners + # 2 players: opposite corners (0,0) and (max, max) + # 3 players: three corners + max_idx = board_size - 1 + self._starting_corners = [ + (0, 0), + (0, max_idx), + (max_idx, 0), + (max_idx, max_idx), + ] + if num_players == 2: + self._starting_corners = [(0, 0), (max_idx, max_idx)] + elif num_players == 3: + self._starting_corners = [(0, 0), (0, max_idx), (max_idx, 0)] + + # ------------------------------------------------------------------ + # Action space generation + # ------------------------------------------------------------------ + + def _generate_action_space(self) -> None: + """Pre-compute all possible (piece, orientation, position) moves. + + Each unique combination is assigned a stable integer action index. + """ + self._action_moves = [] + self._move_to_action = {} + + for piece_id in range(self.piece_set.num_pieces): + for orient_id, orient in enumerate(self.piece_set.get_orientations(piece_id)): + squares = orient.squares + # Compute bounding box of the piece + max_x = max(x for x, _ in squares) + max_y = max(y for _, y in squares) + + for px in range(self.board_size - max_x): + for py in range(self.board_size - max_y): + move = Move( + piece_id=piece_id, + orientation_id=orient_id, + x=px, + y=py, + ) + action_idx = len(self._action_moves) + self._action_moves.append(move) + self._move_to_action[(piece_id, orient_id, px, py)] = action_idx + + @property + def num_actions(self) -> int: + return len(self._action_moves) + + def get_move(self, action: int) -> Move: + """Get the Move object for a given action index.""" + return self._action_moves[action] + + def get_action(self, move: Move) -> int: + """Get the action index for a given Move.""" + return self._move_to_action[(move.piece_id, move.orientation_id, move.x, move.y)] + + # ------------------------------------------------------------------ + # Game initialization + # ------------------------------------------------------------------ + + def reset(self) -> None: + """Reset the game to its initial state.""" + self.board = Board(self.board_size) + self.players = [] + self.current_player = 0 + self.rounds = 0 + self.game_over = False + + for i in range(self.num_players): + player = PlayerState(idx=i) + player.available_pieces = set(self.piece_set.piece_names) + if self.corner_rule and i < len(self._starting_corners): + player.corners = {self._starting_corners[i]} + else: + player.corners = set() + self.players.append(player) + + # ------------------------------------------------------------------ + # Move validation + # ------------------------------------------------------------------ + + def _get_placed_squares(self, move: Move) -> list[tuple[int, int]]: + """Get the absolute board coordinates for a move.""" + orient = self.piece_set.get_orientations(move.piece_id)[move.orientation_id] + return [(move.x + dx, move.y + dy) for dx, dy in orient.squares] + + def _get_placed_corners(self, move: Move) -> list[tuple[int, int]]: + """Get the absolute corner coordinates for a move.""" + orient = self.piece_set.get_orientations(move.piece_id)[move.orientation_id] + return [(move.x + dx, move.y + dy) for dx, dy in orient.corners] + + def valid_move(self, player_idx: int, move: Move) -> bool: + """Check if a move is valid for the given player. + + Validation rules: + 1. Player must have the piece available + 2. All squares must be in bounds + 3. No overlap with existing pieces + 4. Corner rule: piece must touch same-color piece at a corner + 5. No edge adjacency with same-color pieces + 6. First move must be in a corner (if corner_rule is enabled) + """ + player = self.players[player_idx] + piece = self.piece_set.get_piece(move.piece_id) + + # Rule 1: Check if player has the piece + if piece.name not in player.available_pieces: + return False + + placed_squares = self._get_placed_squares(move) + + # Rule 2: Check bounds + for x, y in placed_squares: + if not self.board.in_bounds(x, y): + return False + + # Rule 3: Check overlap + if self.board.has_overlap(placed_squares): + return False + + player_board_idx = player_idx + self.PLAYER_OFFSET + + # Rule 6: First move must be in a corner + if self.corner_rule and not player.has_started: + if len(self.players) <= len(self._starting_corners): + start_corner = self._starting_corners[player_idx] + if (move.x, move.y) != start_corner: + return False + + # Rule 5: No edge adjacency with same-color pieces + for x, y in placed_squares: + for dx, dy in [(-1, 0), (1, 0), (0, -1), (0, 1)]: + nx, ny = x + dx, y + dy + if self.board.in_bounds(nx, ny): + if self.board.get_cell(nx, ny) == player_board_idx: + return False + + # Rule 4: Corner rule (must touch same-color at a corner) + if player.has_started: + # Must touch at least one same-color piece at a corner + placed_corners = self._get_placed_corners(move) + touches_corner = False + for cx, cy in placed_corners: + if self.board.in_bounds(cx, cy): + if self.board.get_cell(cx, cy) == player_board_idx: + touches_corner = True + break + if not touches_corner: + return False + # If not started yet, first move doesn't need to touch (it's the first piece) + + return True + + def get_valid_actions(self, player_idx: int) -> np.ndarray: + """Get a boolean mask of valid actions for the given player.""" + mask = np.zeros(self.num_actions, dtype=bool) + player = self.players[player_idx] + + if not player.can_move: + return mask + + for action_idx in range(self.num_actions): + move = self._action_moves[action_idx] + piece = self.piece_set.get_piece(move.piece_id) + if piece.name not in player.available_pieces: + continue + if self.valid_move(player_idx, move): + mask[action_idx] = True + + return mask + + def get_valid_action_indices(self, player_idx: int) -> list[int]: + """Get a list of valid action indices for the given player.""" + mask = self.get_valid_actions(player_idx) + return [int(i) for i in np.where(mask)[0]] + + def has_valid_moves(self, player_idx: int) -> bool: + """Check if a player has any valid moves.""" + player = self.players[player_idx] + if not player.can_move: + return False + + for action_idx in range(self.num_actions): + move = self._action_moves[action_idx] + piece = self.piece_set.get_piece(move.piece_id) + if piece.name in player.available_pieces: + if self.valid_move(player_idx, move): + return True + return False + + # ------------------------------------------------------------------ + # Move execution + # ------------------------------------------------------------------ + + def apply_move(self, player_idx: int, move: Move) -> None: + """Apply a move to the game state. Does not validate.""" + player = self.players[player_idx] + piece = self.piece_set.get_piece(move.piece_id) + placed_squares = self._get_placed_squares(move) + placed_corners = self._get_placed_corners(move) + + # Place on board + player_board_idx = player_idx + self.PLAYER_OFFSET + self.board.place(player_board_idx, placed_squares) + + # Update player state + player.score += piece.size + player.available_pieces.discard(piece.name) + player.has_started = True + + # Update corners: remove covered squares, add new corners + for cx, cy in placed_corners: + if self.board.in_bounds(cx, cy) and self.board.is_empty(cx, cy): + player.corners.add((cx, cy)) + + # Remove corners that are now occupied + player.corners = { + (x, y) for x, y in player.corners if self.board.is_empty(x, y) + } + + def play_move(self, player_idx: int, action: int) -> bool: + """Validate and apply a move. Returns True if successful.""" + move = self.get_move(action) + if not self.valid_move(player_idx, move): + return False + self.apply_move(player_idx, move) + return True + + # ------------------------------------------------------------------ + # Turn management + # ------------------------------------------------------------------ + + def next_player(self) -> int: + """Advance to the next player who can still move. + + If no players can move, the game is over. + """ + if self.game_over: + return self.current_player + + for offset in range(1, self.num_players + 1): + next_idx = (self.current_player + offset) % self.num_players + next_player = self.players[next_idx] + if next_player.can_move and self.has_valid_moves(next_idx): + self.current_player = next_idx + return next_idx + + # No one can move — game over + self.game_over = True + return self.current_player + + def advance_turn(self) -> None: + """Apply a move for the current player, then advance. + + This is a convenience method for bot-driven games. + """ + self.next_player() + + # ------------------------------------------------------------------ + # Game state queries + # ------------------------------------------------------------------ + + def is_game_over(self) -> bool: + """Check if the game is over (no player can make a move).""" + if self.game_over: + return True + # Check if any player can still move + for i in range(self.num_players): + if self.players[i].can_move and self.has_valid_moves(i): + return False + self.game_over = True + return True + + def get_scores(self) -> list[int]: + """Get the score for each player. + + Score = squares placed (positive) - unplaced squares (negative). + A player who placed all pieces gets a +15 bonus. + If the last piece was the monomino, an additional +5 bonus. + """ + scores = [] + for player in self.players: + # Sum sizes of remaining pieces + unplaced = sum( + self.piece_set.pieces[pid].size + for pid, piece in enumerate(self.piece_set.pieces) + if piece.name in player.available_pieces + ) + score = player.score - unplaced + + # Bonus for placing all pieces + if len(player.available_pieces) == 0: + score += 15 + + scores.append(score) + return scores + + def get_winners(self) -> list[int] | None: + """Get the winning player(s). Returns None if game is not over.""" + if not self.is_game_over(): + return None + + scores = self.get_scores() + max_score = max(scores) + winners = [i for i, s in enumerate(scores) if s == max_score] + return winners + + def get_current_observation(self) -> dict: + """Get the current game state as an observation dict.""" + player_idx = self.current_player + player = self.players[player_idx] + + # Board as player indices (1-based) + board_obs = self.board.grid.copy() + + # Available pieces as binary vector + pieces_obs = np.zeros(self.piece_set.num_pieces, dtype=bool) + for name in player.available_pieces: + pieces_obs[self.piece_set.get_piece_id(name)] = True + + # Corners as boolean grid + corners_obs = np.zeros((self.board_size, self.board_size), dtype=bool) + for x, y in player.corners: + if self.board.in_bounds(x, y): + corners_obs[y, x] = True + + return { + "board": board_obs.astype(np.int8), + "pieces": pieces_obs, + "corners": corners_obs, + } + + def get_action_mask(self, player_idx: int) -> np.ndarray: + """Get the action mask for a player.""" + return self.get_valid_actions(player_idx) + + def copy(self) -> BlokusGame: + """Create a deep copy of the game state.""" + import copy + + new_game = BlokusGame.__new__(BlokusGame) + new_game.board_size = self.board_size + new_game.piece_set = self.piece_set + new_game.num_players = self.num_players + new_game.corner_rule = self.corner_rule + new_game.board = self.board.copy() + new_game.players = copy.deepcopy(self.players) + new_game.current_player = self.current_player + new_game.rounds = self.rounds + new_game.game_over = self.game_over + new_game._action_moves = self._action_moves + new_game._move_to_action = self._move_to_action + new_game._starting_corners = self._starting_corners + return new_game diff --git a/src/blokus_gym/core/pieces.py b/src/blokus_gym/core/pieces.py new file mode 100644 index 0000000..51895c6 --- /dev/null +++ b/src/blokus_gym/core/pieces.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Piece: + """A polyomino piece defined by a set of (x, y) coordinates. + + Coordinates are relative to a reference point (typically the + top-left corner of the bounding box after normalization). + """ + + name: str + squares: frozenset[tuple[int, int]] + + @property + def size(self) -> int: + return len(self.squares) + + def to_dict(self) -> dict: + return { + "name": self.name, + "squares": sorted(self.squares), + } + + @classmethod + def from_dict(cls, data: dict) -> Piece: + return cls( + name=data["name"], + squares=frozenset(tuple(s) for s in data["squares"]), + ) + + +@dataclass +class OrientedPiece: + """A piece in a specific orientation placed at a specific position.""" + + piece_id: int + orientation_id: int + squares: list[tuple[int, int]] + corners: list[tuple[int, int]] + + +@dataclass +class Move: + """A complete move: which piece, which orientation, where placed.""" + + piece_id: int + orientation_id: int + x: int + y: int + + +@dataclass +class PieceOrientation: + """A piece in a specific orientation (before placement on board).""" + + piece_id: int + orientation_id: int + squares: list[tuple[int, int]] # normalized relative coordinates + corners: list[tuple[int, int]] # corner cells relative to squares + + +def _normalize(squares: list[tuple[int, int]]) -> list[tuple[int, int]]: + """Shift coordinates so the minimum x and y are 0.""" + min_x = min(x for x, _ in squares) + min_y = min(y for _, y in squares) + return [(x - min_x, y - min_y) for x, y in squares] + + +def _rotate(squares: list[tuple[int, int]]) -> list[tuple[int, int]]: + """Rotate 90 degrees clockwise: (x, y) -> (y, -x).""" + return [(y, -x) for x, y in squares] + + +def _flip(squares: list[tuple[int, int]]) -> list[tuple[int, int]]: + """Flip horizontally: (x, y) -> (-x, y).""" + return [(-x, y) for x, y in squares] + + +def _compute_corners(squares: list[tuple[int, int]]) -> list[tuple[int, int]]: + """Compute the corner cells for a piece. + + Corner cells are the diagonal neighbors of each square that are not + edge-adjacent to any other square in the piece. These are the cells + where a same-color piece must touch (corner-to-corner). + """ + square_set = set(squares) + corners = set() + for x, y in squares: + for dx, dy in [(-1, -1), (1, -1), (-1, 1), (1, 1)]: + cx, cy = x + dx, y + dy + if (cx, cy) not in square_set: + # Check that this corner is not edge-adjacent to another square + edge_neighbors = [ + (cx + 1, cy), (cx - 1, cy), (cx, cy + 1), (cx, cy - 1) + ] + if not any(en in square_set for en in edge_neighbors): + corners.add((cx, cy)) + return sorted(corners) + + +def generate_orientations(piece: Piece) -> list[PieceOrientation]: + """Generate all unique orientations of a piece. + + Applies 4 rotations and 2 flips, then deduplicates by comparing + normalized coordinate sets. + """ + orientations: list[PieceOrientation] = [] + seen: set[tuple[tuple[int, int], ...]] = set() + + squares = sorted(piece.squares) + for flip_count in range(2): + for rotation_count in range(4): + current = list(squares) + for _ in range(rotation_count): + current = _rotate(current) + for _ in range(flip_count): + current = _flip(current) + normalized = tuple(sorted(_normalize(current))) + if normalized not in seen: + seen.add(normalized) + corners = _compute_corners(list(normalized)) + orientations.append( + PieceOrientation( + piece_id=-1, # Set by PieceSet + orientation_id=len(orientations), + squares=list(normalized), + corners=corners, + ) + ) + return orientations + + +# --------------------------------------------------------------------------- +# Standard Blokus piece sets +# --------------------------------------------------------------------------- + +# The 21 free polyominoes of size 1-5 +# Naming follows standard pentomino/tetromino conventions + +_I1 = Piece("I1", frozenset([(0, 0)])) + +_I2 = Piece("I2", frozenset([(0, 0), (0, 1)])) + +_I3 = Piece("I3", frozenset([(0, 0), (0, 1), (0, 2)])) + +_V3 = Piece("V3", frozenset([(0, 0), (1, 0), (0, 1)])) + +_I4 = Piece("I4", frozenset([(0, 0), (0, 1), (0, 2), (0, 3)])) + +_L4 = Piece("L4", frozenset([(0, 0), (0, 1), (0, 2), (1, 0)])) + +_T4 = Piece("T4", frozenset([(0, 0), (1, 0), (2, 0), (1, 1)])) + +_S4 = Piece("S4", frozenset([(0, 0), (1, 0), (1, 1), (2, 1)])) + +_O4 = Piece("O4", frozenset([(0, 0), (1, 0), (0, 1), (1, 1)])) + +_I5 = Piece("I5", frozenset([(0, 0), (0, 1), (0, 2), (0, 3), (0, 4)])) + +_L5 = Piece("L5", frozenset([(0, 0), (0, 1), (0, 2), (0, 3), (1, 0)])) + +_Y5 = Piece("Y5", frozenset([(0, 0), (0, 1), (0, 2), (0, 3), (1, 1)])) + +_N5 = Piece("N5", frozenset([(0, 0), (1, 0), (2, 0), (2, 1), (3, 1)])) + +_T5 = Piece("T5", frozenset([(0, 0), (1, 0), (2, 0), (1, 1), (1, 2)])) + +_U5 = Piece("U5", frozenset([(0, 0), (2, 0), (0, 1), (1, 1), (2, 1)])) + +_V5 = Piece("V5", frozenset([(0, 0), (0, 1), (0, 2), (1, 0), (2, 0)])) + +_W5 = Piece("W5", frozenset([(0, 0), (0, 1), (1, 0), (1, 1), (2, 0)])) + +_Z5 = Piece("Z5", frozenset([(0, 0), (1, 0), (1, 1), (1, 2), (2, 2)])) + +_F5 = Piece("F5", frozenset([(0, 0), (1, 0), (1, 1), (2, 1), (1, 2)])) + +_X5 = Piece("X5", frozenset([(0, 0), (1, 0), (2, 0), (1, 1), (1, -1)])) + +_P5 = Piece("P5", frozenset([(0, 0), (1, 0), (0, 1), (1, 1), (0, 2)])) + +STANDARD_PIECES: list[Piece] = [ + _I1, _I2, _I3, _V3, _I4, _L4, _T4, _S4, _O4, + _I5, _L5, _Y5, _N5, _T5, _U5, _V5, _W5, _Z5, _F5, _X5, _P5, +] + +# Blokus Duo uses the same pieces but on a 14x14 board +DUO_PIECES: list[Piece] = STANDARD_PIECES + +# Blokus Junior: simplified pieces (only 12 unique pieces, 2 copies each) +# Uses only pieces with size <= 4 for simplicity +JUNIOR_PIECES: list[Piece] = [ + _I1, _I2, _I3, _V3, _I4, _L4, _T4, _S4, _O4, +] + + +class PieceSet: + """A collection of pieces with pre-computed orientations. + + This class manages the piece set used in a Blokus game, including + all unique orientations for each piece and a lookup table for + generating moves. + """ + + def __init__(self, pieces: list[Piece]): + self.pieces: list[Piece] = pieces + self.piece_names: list[str] = [p.name for p in pieces] + self.piece_id_map: dict[str, int] = {name: i for i, name in enumerate(self.piece_names)} + + # Pre-compute all orientations for each piece + self.orientations: list[list[PieceOrientation]] = [] + for piece_id, piece in enumerate(pieces): + orients = generate_orientations(piece) + for orient in orients: + orient.piece_id = piece_id + self.orientations.append(orients) + + # Total number of (piece, orientation) combinations + self.num_orientations: int = sum(len(o) for o in self.orientations) + + @property + def num_pieces(self) -> int: + return len(self.pieces) + + def get_orientations(self, piece_id: int) -> list[PieceOrientation]: + return self.orientations[piece_id] + + def get_piece(self, piece_id: int) -> Piece: + return self.pieces[piece_id] + + def get_piece_id(self, name: str) -> int: + return self.piece_id_map[name] + + def to_dict(self) -> dict: + return { + "pieces": [p.to_dict() for p in self.pieces], + } + + @classmethod + def from_dict(cls, data: dict) -> PieceSet: + pieces = [Piece.from_dict(p) for p in data["pieces"]] + return cls(pieces) + + @classmethod + def from_json(cls, path: str) -> PieceSet: + import json + + with open(path) as f: + data = json.load(f) + return cls.from_dict(data) + + def save_json(self, path: str) -> None: + import json + + with open(path, "w") as f: + json.dump(self.to_dict(), f, indent=2) + + def __len__(self) -> int: + return len(self.pieces) + + def __repr__(self) -> str: + return f"PieceSet(num_pieces={self.num_pieces}, num_orientations={self.num_orientations})" diff --git a/src/blokus_gym/envs/__init__.py b/src/blokus_gym/envs/__init__.py new file mode 100644 index 0000000..e7fd5e2 --- /dev/null +++ b/src/blokus_gym/envs/__init__.py @@ -0,0 +1,97 @@ +from blokus_gym.core.bots import GreedyBot, GreedyCornersBot, MinimaxBot, RandomBot +from blokus_gym.core.pieces import DUO_PIECES, JUNIOR_PIECES, STANDARD_PIECES +from blokus_gym.envs.blokus_env import BlokusEnv +from blokus_gym.envs.multiagent import BlokusMultiAgentEnv + +__all__ = [ + "BlokusEnv", + "BlokusMultiAgentEnv", + "STANDARD_PIECES", + "DUO_PIECES", + "JUNIOR_PIECES", + "RandomBot", + "GreedyBot", + "GreedyCornersBot", + "MinimaxBot", +] + + +def _register_envs(): + """Register all predefined environment configurations.""" + import gymnasium as gym + + # Standard 4-player Blokus (20x20) + gym.register( + id="Blokus-v0", + entry_point="blokus_gym.envs:BlokusEnv", + kwargs={ + "num_players": 4, + "board_size": 20, + "pieces": STANDARD_PIECES, + }, + max_episode_steps=500, + ) + + # Blokus Duo (2 players, 14x14) + gym.register( + id="BlokusDuo-v0", + entry_point="blokus_gym.envs:BlokusEnv", + kwargs={ + "num_players": 2, + "board_size": 14, + "pieces": DUO_PIECES, + }, + max_episode_steps=300, + ) + + # Blokus Junior (2 players, 14x14, simplified pieces) + gym.register( + id="BlokusJunior-v0", + entry_point="blokus_gym.envs:BlokusEnv", + kwargs={ + "num_players": 2, + "board_size": 14, + "pieces": JUNIOR_PIECES, + }, + max_episode_steps=200, + ) + + # Simple test env (2 players, 7x7, only small pieces) + gym.register( + id="BlokusSimple-v0", + entry_point="blokus_gym.envs:BlokusEnv", + kwargs={ + "num_players": 2, + "board_size": 7, + "pieces": [p for p in STANDARD_PIECES if p.size < 5], + }, + max_episode_steps=100, + ) + + # Greedy bot variants + gym.register( + id="BlokusGreedy-v0", + entry_point="blokus_gym.envs:BlokusEnv", + kwargs={ + "num_players": 4, + "board_size": 20, + "pieces": STANDARD_PIECES, + "bot_type": GreedyBot, + }, + max_episode_steps=500, + ) + + gym.register( + id="BlokusDuoGreedy-v0", + entry_point="blokus_gym.envs:BlokusEnv", + kwargs={ + "num_players": 2, + "board_size": 14, + "pieces": DUO_PIECES, + "bot_type": GreedyBot, + }, + max_episode_steps=300, + ) + + +_register_envs() diff --git a/src/blokus_gym/envs/__pycache__/__init__.cpython-312.pyc b/src/blokus_gym/envs/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..6083fcc Binary files /dev/null and b/src/blokus_gym/envs/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/blokus_gym/envs/__pycache__/blokus_env.cpython-312.pyc b/src/blokus_gym/envs/__pycache__/blokus_env.cpython-312.pyc new file mode 100644 index 0000000..317c7b0 Binary files /dev/null and b/src/blokus_gym/envs/__pycache__/blokus_env.cpython-312.pyc differ diff --git a/src/blokus_gym/envs/__pycache__/multiagent.cpython-312.pyc b/src/blokus_gym/envs/__pycache__/multiagent.cpython-312.pyc new file mode 100644 index 0000000..93985de Binary files /dev/null and b/src/blokus_gym/envs/__pycache__/multiagent.cpython-312.pyc differ diff --git a/src/blokus_gym/envs/blokus_env.py b/src/blokus_gym/envs/blokus_env.py new file mode 100644 index 0000000..f6209f8 --- /dev/null +++ b/src/blokus_gym/envs/blokus_env.py @@ -0,0 +1,366 @@ +from __future__ import annotations + +from typing import Any + +import gymnasium as gym +import numpy as np +from gymnasium import spaces + +from blokus_gym.core.bots import Bot, RandomBot +from blokus_gym.core.game import BlokusGame +from blokus_gym.core.pieces import STANDARD_PIECES, Piece, PieceSet +from blokus_gym.utils.render import render_ansi_board, render_text_board + + +class BlokusEnv(gym.Env): + """Gymnasium environment for the board game Blokus. + + Supports 2-4 players on customizable board sizes with custom piece sets. + In single-agent mode, the agent controls player 0 and opponents are bots. + + Action Space: Discrete(N) where N is the total number of possible + (piece, orientation, position) combinations. An action mask is provided + in the info dict to indicate valid actions. + + Observation Space: Dict with: + - "board": (board_size, board_size) int8 array (0=empty, 1-4=player) + - "pieces": MultiBinary(num_pieces) - 1 if piece still available + - "corners": (board_size, board_size) bool array - valid corner cells + + Example: + >>> import gymnasium as gym + >>> from blokus_gym import BlokusEnv + >>> env = BlokusEnv(num_players=4, board_size=20) + >>> obs, info = env.reset(seed=42) + >>> action = env.action_space.sample() # Use info["action_mask"] instead + >>> obs, reward, terminated, truncated, info = env.step(action) + """ + + metadata = {"render_modes": ["human", "ansi", "rgb_array"]} + + # Player colors for rendering (index 0 = empty) + PLAYER_COLORS = ["empty", "red", "blue", "yellow", "green"] + + def __init__( + self, + num_players: int = 4, + board_size: int = 20, + pieces: list[Piece] | None = None, + render_mode: str | None = None, + bot_type: type[Bot] = RandomBot, + bot_strength: int = 1, + reward_shaping: bool = False, + corner_rule: bool = True, + max_steps: int | None = None, + seed: int | None = None, + **kwargs: Any, + ): + super().__init__() + + # Validate parameters + assert 2 <= num_players <= 4, f"num_players must be 2-4, got {num_players}" + assert board_size >= 5, f"board_size must be >= 5, got {board_size}" + + self.num_players = num_players + self.board_size = board_size + self.render_mode = render_mode + self.bot_type = bot_type + self.bot_strength = bot_strength + self.reward_shaping = reward_shaping + self.corner_rule = corner_rule + self.max_steps = max_steps + self._seed = seed + + # Initialize piece set + self.piece_set = PieceSet(pieces or STANDARD_PIECES) + + # Initialize game + self.game = BlokusGame( + board_size=board_size, + pieces=self.piece_set, + num_players=num_players, + corner_rule=corner_rule, + ) + + # Define spaces + self.observation_space = spaces.Dict({ + "board": spaces.Box(0, num_players, (board_size, board_size), dtype=np.int8), + "pieces": spaces.MultiBinary(self.piece_set.num_pieces), + "corners": spaces.Box(0, 1, (board_size, board_size), dtype=bool), + }) + + self.action_space = spaces.Discrete(self.game.num_actions) + + # Bot instances (created during reset) + self.bots: list[Bot | None] = [None] * num_players + + # Episode tracking + self.current_step = 0 + self._last_observation: dict | None = None + + def _get_obs(self) -> dict: + """Get the current observation from the agent's perspective (player 0).""" + obs = self.game.get_current_observation() + return { + "board": obs["board"], + "pieces": obs["pieces"], + "corners": obs["corners"], + } + + def _get_info(self) -> dict: + """Get auxiliary information.""" + return { + "action_mask": self.game.get_action_mask(0), + "current_player": self.game.current_player, + "players_with_moves": [ + i for i in range(self.num_players) + if self.game.players[i].can_move and self.game.has_valid_moves(i) + ], + "step_count": self.current_step, + "scores": self.game.get_scores(), + } + + def reset( + self, + seed: int | None = None, + options: dict | None = None, + ) -> tuple[dict, dict]: + """Reset the environment to a new episode. + + Args: + seed: Random seed for reproducibility. + options: Additional options (unused). + + Returns: + Tuple of (observation, info). + """ + super().reset(seed=seed) + + # Re-initialize game + self.game = BlokusGame( + board_size=self.board_size, + pieces=self.piece_set, + num_players=self.num_players, + corner_rule=self.corner_rule, + ) + self.game.reset() + + # Create bots for opponents + bot_seed = seed or 0 + for i in range(1, self.num_players): + self.bots[i] = self.bot_type( + player_idx=i, + seed=bot_seed + i * 1000, + ) + + self.current_step = 0 + + # Let bots play until it's the agent's turn + # The agent (player 0) starts first if corner_rule is enabled + self._play_bots_until_agent() + + obs = self._get_obs() + info = self._get_info() + self._last_observation = obs + + return obs, info + + def _play_bots_until_agent(self) -> None: + """Let bot players take their turns until it's the agent's turn.""" + max_bot_turns = self.num_players * 10 # Safety limit + turns = 0 + + while self.game.current_player != 0 and not self.game.is_game_over(): + if turns > max_bot_turns: + break + bot = self.bots[self.game.current_player] + if bot is None: + # Shouldn't happen, but safety + self.game.next_player() + turns += 1 + continue + + valid_actions = self.game.get_valid_actions(self.game.current_player) + if not np.any(valid_actions): + self.game.players[self.game.current_player].can_move = False + self.game.next_player() + turns += 1 + continue + + action = bot.select_action(self.game, valid_actions) + if action is not None: + self.game.play_move(self.game.current_player, action) + else: + self.game.players[self.game.current_player].can_move = False + + self.game.next_player() + turns += 1 + + def step(self, action: int) -> tuple[dict, float, bool, bool, dict]: + """Execute one step in the environment. + + The agent (player 0) takes an action, then all bots play until + it's the agent's turn again. + + Args: + action: Action index (piece, orientation, position). + + Returns: + Tuple of (observation, reward, terminated, truncated, info). + """ + self.current_step += 1 + + # Apply agent's action + success = self.game.play_move(0, action) + + if not success: + # Invalid action - penalize and end episode + reward = -10.0 + obs = self._get_obs() + info = self._get_info() + info["invalid_action"] = True + self._last_observation = obs + return obs, reward, True, False, info + + # Advance to next player + self.game.next_player() + + # Let bots play until it's the agent's turn or game is over + self._play_bots_until_agent() + + # Check if game is over + terminated = self.game.is_game_over() + truncated = False + if self.max_steps is not None and self.current_step >= self.max_steps: + truncated = True + + # Compute reward + if terminated or truncated: + reward = self._compute_terminal_reward() + else: + reward = self._compute_step_reward() + + obs = self._get_obs() + info = self._get_info() + self._last_observation = obs + + return obs, reward, terminated, truncated, info + + def _compute_step_reward(self) -> float: + """Compute reward for a non-terminal step.""" + if self.reward_shaping: + # Small reward for each square placed + return 0.01 * self.game.players[0].score + return 0.0 + + def _compute_terminal_reward(self) -> float: + """Compute reward when the game ends.""" + scores = self.game.get_scores() + agent_score = scores[0] + + if self.reward_shaping: + # Normalized score: agent score / max possible score + max_possible = sum(p.size for p in self.piece_set.pieces) + return agent_score / max_possible + + # Win/loss reward + max_score = max(scores) + if agent_score == max_score: + # Check if it's a tie + winners = [i for i, s in enumerate(scores) if s == max_score] + if len(winners) == 1: + return 1.0 # Win + else: + return 0.0 # Tie + else: + return -1.0 # Loss + + def render(self) -> str | np.ndarray | None: + """Render the environment. + + Args: + No arguments - uses self.render_mode. + + Returns: + For "human": None (prints to stdout) + For "ansi": string representation + For "rgb_array": numpy array of shape (H, W, 3) + """ + if self.render_mode is None: + return None + + if self.render_mode == "human": + render_text_board(self.game) + return None + + elif self.render_mode == "ansi": + return render_ansi_board(self.game) + + elif self.render_mode == "rgb_array": + return self._render_rgb_array() + + else: + raise ValueError(f"Unknown render_mode: {self.render_mode}") + + def _render_rgb_array(self) -> np.ndarray: + """Render the board as an RGB array using matplotlib.""" + try: + import matplotlib + matplotlib.use("Agg") + import matplotlib.pyplot as plt + except ImportError as exc: + raise ImportError( + "matplotlib is required for rgb_array rendering. " + "Install with: pip install matplotlib" + ) from exc + + fig, ax = plt.subplots(figsize=(8, 8)) + + # Color map: 0=empty(light grey), 1=red, 2=blue, 3=yellow, 4=green + colors = { + 0: "#d3d3d3", # light grey + 1: "#ff6b6b", # red + 2: "#4dabf7", # blue + 3: "#ffd43b", # yellow + 4: "#51cf66", # green + } + + grid = self.game.board.grid.astype(float) + colored = np.zeros((self.board_size, self.board_size, 3)) + for i in range(self.board_size): + for j in range(self.board_size): + val = int(grid[i, j]) + hex_color = colors.get(val, "#d3d3d3") + # Parse hex color + r = int(hex_color[1:3], 16) / 255 + g = int(hex_color[3:5], 16) / 255 + b = int(hex_color[5:7], 16) / 255 + colored[i, j] = [r, g, b] + + ax.imshow(colored, interpolation="nearest") + ax.set_xticks(np.arange(-0.5, self.board_size, 1), minor=True) + ax.set_yticks(np.arange(-0.5, self.board_size, 1), minor=True) + ax.grid(True, which="minor", color="black", linewidth=0.5) + ax.set_xticks([]) + ax.set_yticks([]) + + # Title with current player and scores + scores = self.game.get_scores() + title = f"Current player: {self.game.current_player + 1}\n" + title += "Scores: " + " | ".join(f"P{i+1}: {s}" for i, s in enumerate(scores)) + ax.set_title(title, fontsize=10) + + plt.tight_layout() + fig.canvas.draw() + image = np.frombuffer(fig.canvas.tostring_rgb(), dtype=np.uint8) + image = image.reshape(fig.canvas.get_width_height()[::-1] + (3,)) + plt.close(fig) + return image + + def close(self) -> None: + """Clean up resources.""" + pass + + def get_action_mask(self) -> np.ndarray: + """Get the action mask for the current player (player 0).""" + return self.game.get_action_mask(0) diff --git a/src/blokus_gym/envs/multiagent.py b/src/blokus_gym/envs/multiagent.py new file mode 100644 index 0000000..670f811 --- /dev/null +++ b/src/blokus_gym/envs/multiagent.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +import numpy as np +from gymnasium import spaces + +from blokus_gym.core.game import BlokusGame +from blokus_gym.core.pieces import STANDARD_PIECES, PieceSet + + +class BlokusMultiAgentEnv: + """PettingZoo-style AEC (Actor-Environment-Cycle) environment for Blokus. + + This wraps the BlokusGame core to provide a multi-agent interface where + each player acts independently. Compatible with PettingZoo's API and + RLlib's multi-agent RL. + + Unlike the single-agent BlokusEnv, all players are controlled by the + caller (no built-in bots). This allows for self-play training and + evaluation of multi-agent policies. + + Attributes: + num_players: Number of players (2-4). + board_size: Size of the square board. + agents: List of agent names (e.g., ["player_0", "player_1", ...]). + possible_agents: Same as agents (never changes). + agent_order: Order in which agents take turns. + + Example: + >>> from blokus_gym import BlokusMultiAgentEnv + >>> env = BlokusMultiAgentEnv(num_players=4) + >>> obs, info = env.reset() + >>> for agent in env.agent_iter(): + ... obs, reward, terminated, truncated, info = env.last() + ... action = env.action_space(agent).sample() # Use action mask + ... env.step(action) + """ + + metadata = {"render_modes": ["human", "ansi"]} + + def __init__( + self, + num_players: int = 4, + board_size: int = 20, + pieces: list | None = None, + render_mode: str | None = None, + corner_rule: bool = True, + seed: int | None = None, + ): + assert 2 <= num_players <= 4, f"num_players must be 2-4, got {num_players}" + assert board_size >= 5, f"board_size must be >= 5, got {board_size}" + + self.num_players = num_players + self.board_size = board_size + self.render_mode = render_mode + self.corner_rule = corner_rule + self._seed = seed + + self.piece_set = PieceSet(pieces or STANDARD_PIECES) + self.game: BlokusGame | None = None + + self.agents = [f"player_{i}" for i in range(num_players)] + self.possible_agents = list(self.agents) + self.agent_order = list(self.agents) + + # Observation and action spaces (same for all agents) + self.observation_space = spaces.Dict({ + "board": spaces.Box(0, num_players, (board_size, board_size), dtype=np.int8), + "pieces": spaces.MultiBinary(self.piece_set.num_pieces), + "corners": spaces.Box(0, 1, (board_size, board_size), dtype=bool), + }) + + self._cur_step = 0 + self._current_agent_idx = 0 + self._agent_dones: dict[str, bool] = {} + + def reset(self, seed: int | None = None, options: dict | None = None) -> tuple[dict, dict]: + """Reset the environment. + + Returns: + obs: Dict mapping agent name to observation. + info: Dict with auxiliary information. + """ + self.game = BlokusGame( + board_size=self.board_size, + pieces=self.piece_set, + num_players=self.num_players, + corner_rule=self.corner_rule, + ) + self.game.reset() + + self._cur_step = 0 + self._current_agent_idx = 0 + self._agent_dones = dict.fromkeys(self.agents, False) + + # Get observation for the first agent + first_agent = self.agent_order[0] + obs = self._get_obs(first_agent) + info = self._get_info(first_agent) + + return {first_agent: obs}, info + + def step(self, action: int) -> tuple[dict, dict, dict, dict, dict]: + """Execute one step. Must be called for the current agent in agent_iter(). + + Returns: + observations: Dict mapping agent -> observation + rewards: Dict mapping agent -> reward + terminations: Dict mapping agent -> bool + truncations: Dict mapping agent -> bool + infos: Dict mapping agent -> info + """ + self._cur_step += 1 + agent = self.agent_order[self._current_agent_idx] + player_idx = int(agent.split("_")[1]) + + # Apply the action + success = self.game.play_move(player_idx, action) + + # Compute rewards for this player + reward = 0.0 + if not success: + reward = -10.0 # Invalid action penalty + self._agent_dones[agent] = True + + # Check game over + game_over = self.game.is_game_over() + + # Advance to next player + if not game_over: + self.game.next_player() + # Skip players who can't move + while not self.game.is_game_over(): + player = self.game.current_player + agent_name = self.agents[player] + if self._agent_dones.get(agent_name, False): + self.game.next_player() + continue + if not self.game.has_valid_moves(player): + self.game.players[player].can_move = False + self.game.next_player() + continue + break + + # Determine next agent + next_player = self.game.current_player + next_agent = self.agents[next_player] + + # Check if game is over + if game_over: + # Final rewards based on scores + scores = self.game.get_scores() + max_score = max(scores) + terminations = {} + rewards = {} + for i, agent_name in enumerate(self.agents): + if scores[i] == max_score: + winners = [j for j, s in enumerate(scores) if s == max_score] + if len(winners) == 1: + rewards[agent_name] = 1.0 + else: + rewards[agent_name] = 0.0 + else: + rewards[agent_name] = -1.0 + terminations[agent_name] = True + self._agent_dones[agent_name] = True + + infos = {agent_name: {"step_count": self._cur_step} for agent_name in self.agents} + observations = {agent_name: self._get_obs(agent_name) for agent_name in self.agents} + + return observations, rewards, terminations, dict.fromkeys(self.agents, False), infos + + # Normal step + rewards = dict.fromkeys(self.agents, 0.0) + rewards[agent] = reward + terminations = { + agent_name: self._agent_dones.get(agent_name, False) + for agent_name in self.agents + } + truncations = dict.fromkeys(self.agents, False) + + observations = {next_agent: self._get_obs(next_agent)} + infos = {next_agent: self._get_info(next_agent)} + + # Advance current agent index + self._current_agent_idx = self.agents.index(next_agent) + + return observations, rewards, terminations, truncations, infos + + def observe(self, agent: str) -> dict: + """Get observation for a specific agent.""" + return self._get_obs(agent) + + def get_action_mask(self, agent: str) -> np.ndarray: + """Get valid action mask for a specific agent.""" + player_idx = int(agent.split("_")[1]) + return self.game.get_action_mask(player_idx) + + def action_space(self, agent: str) -> spaces.Discrete: + """Get action space for a specific agent.""" + return spaces.Discrete(self.game.num_actions) + + def observation_space(self, agent: str) -> spaces.Dict: + """Get observation space for a specific agent.""" + return self.observation_space + + def render(self) -> str | None: + """Render the current state.""" + if self.render_mode is None: + return None + + from blokus_gym.utils.render import render_ansi_board, render_text_board + + if self.render_mode == "human": + print(render_text_board(self.game)) + return None + elif self.render_mode == "ansi": + return render_ansi_board(self.game) + else: + raise ValueError(f"Unknown render_mode: {self.render_mode}") + + def close(self) -> None: + """Clean up resources.""" + pass + + def _get_obs(self, agent: str) -> dict: + """Get observation from an agent's perspective.""" + obs = self.game.get_current_observation() + return { + "board": obs["board"], + "pieces": obs["pieces"], + "corners": obs["corners"], + } + + def _get_info(self, agent: str) -> dict: + """Get auxiliary info for an agent.""" + player_idx = int(agent.split("_")[1]) + return { + "action_mask": self.game.get_action_mask(player_idx), + "current_player": self.game.current_player, + "step_count": self._cur_step, + "scores": self.game.get_scores(), + } + + def agent_iter(self): + """Iterate over agents in turn order.""" + # This is a generator that yields the current agent + while True: + agent = self.agents[self.game.current_player] + if self._agent_dones.get(agent, False): + # Skip done agents + if all(self._agent_dones.values()): + break + continue + yield agent diff --git a/src/blokus_gym/utils/__init__.py b/src/blokus_gym/utils/__init__.py new file mode 100644 index 0000000..0d8ff89 --- /dev/null +++ b/src/blokus_gym/utils/__init__.py @@ -0,0 +1,17 @@ +from blokus_gym.utils.render import ( + ANSI_COLORS, + ANSI_RESET, + PLAYER_NAMES, + print_board, + render_ansi_board, + render_text_board, +) + +__all__ = [ + "ANSI_COLORS", + "ANSI_RESET", + "PLAYER_NAMES", + "render_ansi_board", + "render_text_board", + "print_board", +] diff --git a/src/blokus_gym/utils/__pycache__/__init__.cpython-312.pyc b/src/blokus_gym/utils/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..415c606 Binary files /dev/null and b/src/blokus_gym/utils/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/blokus_gym/utils/__pycache__/render.cpython-312.pyc b/src/blokus_gym/utils/__pycache__/render.cpython-312.pyc new file mode 100644 index 0000000..7f4fb6a Binary files /dev/null and b/src/blokus_gym/utils/__pycache__/render.cpython-312.pyc differ diff --git a/src/blokus_gym/utils/render.py b/src/blokus_gym/utils/render.py new file mode 100644 index 0000000..ff8ccb6 --- /dev/null +++ b/src/blokus_gym/utils/render.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +from blokus_gym.core.game import BlokusGame + +# ANSI color codes for terminal rendering +ANSI_COLORS = { + 0: "\033[90m", # grey (empty) + 1: "\033[91m", # red + 2: "\033[94m", # blue + 3: "\033[93m", # yellow + 4: "\033[92m", # green +} +ANSI_RESET = "\033[0m" + +# Player names +PLAYER_NAMES = {1: "Red", 2: "Blue", 3: "Yellow", 4: "Green"} + + +def render_text_board(game: BlokusGame) -> str: + """Render the board as a text string (no ANSI colors). + + Returns a string representation of the board. + """ + lines = [] + size = game.board_size + + # Column headers + header = " " + " ".join(str(i % 10) for i in range(size)) + lines.append(header) + + for y in range(size): + row = f"{y % 10} " + for x in range(size): + cell = game.board.get_cell(x, y) + if cell == 0: + row += ". " + else: + row += f"{cell} " + lines.append(row) + + # Player info + lines.append("") + for i, player in enumerate(game.players): + remaining = len(player.available_pieces) + total = len(game.piece_set.pieces) + name = PLAYER_NAMES.get(i + 1, f"Player {i + 1}") + lines.append(f"{name}: {remaining}/{total} pieces remaining, score: {player.score}") + + cur_name = PLAYER_NAMES.get(game.current_player + 1, game.current_player + 1) + lines.append(f"Current player: {cur_name}") + + return "\n".join(lines) + + +def render_ansi_board(game: BlokusGame) -> str: + """Render the board with ANSI color codes. + + Returns a string with ANSI escape sequences for colored output. + """ + lines = [] + size = game.board_size + + # Column headers + header = " " + " ".join(str(i % 10) for i in range(size)) + lines.append(header) + + for y in range(size): + row = f"{y % 10} " + for x in range(size): + cell = game.board.get_cell(x, y) + color = ANSI_COLORS.get(cell, ANSI_RESET) + if cell == 0: + row += f"{color}. {ANSI_RESET}" + else: + row += f"{color}{cell} {ANSI_RESET}" + lines.append(row) + + # Player info + lines.append("") + for i, player in enumerate(game.players): + remaining = len(player.available_pieces) + total = len(game.piece_set.pieces) + name = PLAYER_NAMES.get(i + 1, f"Player {i + 1}") + color = ANSI_COLORS.get(i + 1, ANSI_RESET) + lines.append( + f"{color}{name}{ANSI_RESET}: {remaining}/{total} pieces, " + f"score: {player.score}" + ) + + lines.append( + f"Current player: " + f"{ANSI_COLORS.get(game.current_player + 1, ANSI_RESET)}" + f"{PLAYER_NAMES.get(game.current_player + 1, game.current_player + 1)}" + f"{ANSI_RESET}" + ) + + return "\n".join(lines) + + +def print_board(game: BlokusGame, use_ansi: bool = True) -> None: + """Print the board to stdout.""" + if use_ansi: + print(render_ansi_board(game)) + else: + print(render_text_board(game)) diff --git a/src/blokus_gym/wrappers/__init__.py b/src/blokus_gym/wrappers/__init__.py new file mode 100644 index 0000000..27085c7 --- /dev/null +++ b/src/blokus_gym/wrappers/__init__.py @@ -0,0 +1,3 @@ +from blokus_gym.wrappers.action_mask import ActionMaskWrapper + +__all__ = ["ActionMaskWrapper"] diff --git a/src/blokus_gym/wrappers/__pycache__/__init__.cpython-312.pyc b/src/blokus_gym/wrappers/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..3d70799 Binary files /dev/null and b/src/blokus_gym/wrappers/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/blokus_gym/wrappers/__pycache__/action_mask.cpython-312.pyc b/src/blokus_gym/wrappers/__pycache__/action_mask.cpython-312.pyc new file mode 100644 index 0000000..5e117f7 Binary files /dev/null and b/src/blokus_gym/wrappers/__pycache__/action_mask.cpython-312.pyc differ diff --git a/src/blokus_gym/wrappers/action_mask.py b/src/blokus_gym/wrappers/action_mask.py new file mode 100644 index 0000000..c7b2b8e --- /dev/null +++ b/src/blokus_gym/wrappers/action_mask.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import numpy as np +from gymnasium import spaces +from gymnasium.core import Env, Wrapper + + +class ActionMaskWrapper(Wrapper): + """Wrapper that moves the action mask from info into the observation space. + + This is useful for RL frameworks that expect the action mask to be part + of the observation (e.g., Stable-Baselines3 with custom policies, + RLlib, etc.). + + After wrapping, the observation becomes a Dict with: + - "observation": the original observation + - "action_mask": boolean array of valid actions + + Example: + >>> from blokus_gym import BlokusEnv + >>> from blokus_gym.wrappers import ActionMaskWrapper + >>> env = BlokusEnv() + >>> env = ActionMaskWrapper(env) + >>> obs, info = env.reset() + >>> obs["action_mask"] # Boolean mask of valid actions + """ + + def __init__(self, env: Env): + super().__init__(env) + + # Build new observation space + original_obs_space = env.observation_space + action_dim = env.action_space.n + + self.observation_space = spaces.Dict({ + "observation": original_obs_space, + "action_mask": spaces.Box(0, 1, (action_dim,), dtype=bool), + }) + + def reset(self, *, seed=None, options=None): + obs, info = self.env.reset(seed=seed, options=options) + action_mask = info.get("action_mask", np.zeros(self.env.action_space.n, dtype=bool)) + return {"observation": obs, "action_mask": action_mask}, info + + def step(self, action): + obs, reward, terminated, truncated, info = self.env.step(action) + action_mask = info.get("action_mask", np.zeros(self.env.action_space.n, dtype=bool)) + return {"observation": obs, "action_mask": action_mask}, reward, terminated, truncated, info diff --git a/tests/__pycache__/test_board.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_board.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..5f781db Binary files /dev/null and b/tests/__pycache__/test_board.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_envs.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_envs.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..0a5e3bf Binary files /dev/null and b/tests/__pycache__/test_envs.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_game.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_game.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..1323a3f Binary files /dev/null and b/tests/__pycache__/test_game.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_imports.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_imports.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..353d630 Binary files /dev/null and b/tests/__pycache__/test_imports.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_pieces.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_pieces.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..39727a8 Binary files /dev/null and b/tests/__pycache__/test_pieces.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/test_board.py b/tests/test_board.py new file mode 100644 index 0000000..3250fc8 --- /dev/null +++ b/tests/test_board.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import numpy as np + +from blokus_gym.core.board import Board + + +class TestBoard: + def test_board_creation(self): + board = Board(20) + assert board.size == 20 + assert board.grid.shape == (20, 20) + assert np.all(board.grid == 0) + + def test_in_bounds(self): + board = Board(20) + assert board.in_bounds(0, 0) + assert board.in_bounds(19, 19) + assert not board.in_bounds(-1, 0) + assert not board.in_bounds(20, 0) + assert not board.in_bounds(0, 20) + + def test_place_and_get(self): + board = Board(20) + cells = [(0, 0), (1, 0), (2, 0)] + board.place(1, cells) + assert board.get_cell(0, 0) == 1 + assert board.get_cell(1, 0) == 1 + assert board.get_cell(2, 0) == 1 + assert board.get_cell(3, 0) == 0 + + def test_has_overlap(self): + board = Board(20) + board.place(1, [(0, 0), (1, 0)]) + assert board.has_overlap([(0, 0), (1, 0)]) # overlap + assert not board.has_overlap([(2, 0), (3, 0)]) # no overlap + + def test_has_overlap_out_of_bounds(self): + board = Board(20) + assert board.has_overlap([(20, 0)]) # out of bounds + + def test_clear(self): + board = Board(20) + board.place(1, [(0, 0)]) + board.clear() + assert np.all(board.grid == 0) + + def test_get_player_squares(self): + board = Board(20) + board.place(1, [(0, 0), (1, 0)]) + squares = board.get_player_squares(1) + assert len(squares) == 2 + assert (0, 0) in squares + assert (1, 0) in squares + + def test_get_player_corners(self): + board = Board(20) + board.place(1, [(0, 0)]) + corners = board.get_player_corners(1) + assert (1, 1) in corners # diagonal + assert (0, 0) not in corners # occupied + + def test_is_full(self): + board = Board(2) + assert not board.is_full() + board.place(1, [(0, 0), (0, 1), (1, 0), (1, 1)]) + assert board.is_full() + + def test_copy(self): + board = Board(20) + board.place(1, [(0, 0)]) + board_copy = board.copy() + assert board_copy.get_cell(0, 0) == 1 + board_copy.place(2, [(1, 0)]) + assert board.get_cell(1, 0) == 0 # original unchanged + + def test_is_empty(self): + board = Board(20) + assert board.is_empty(0, 0) + board.place(1, [(0, 0)]) + assert not board.is_empty(0, 0) + + def test_coverage(self): + board = Board(2) + assert board.coverage() == 0.0 + board.place(1, [(0, 0)]) + assert board.coverage() == 0.25 + + def test_get_occupied(self): + board = Board(20) + board.place(1, [(0, 0), (1, 0)]) + board.place(2, [(5, 5)]) + occupied = board.get_occupied() + assert (0, 0) in occupied + assert (1, 0) in occupied + assert (5, 5) in occupied + assert (2, 0) not in occupied diff --git a/tests/test_envs.py b/tests/test_envs.py new file mode 100644 index 0000000..c4792b5 --- /dev/null +++ b/tests/test_envs.py @@ -0,0 +1,197 @@ +from __future__ import annotations + +import gymnasium as gym +import numpy as np + +from blokus_gym import ( + STANDARD_PIECES, + ActionMaskWrapper, + BlokusEnv, + BlokusMultiAgentEnv, + GreedyBot, +) + + +class TestBlokusEnv: + def test_env_creation(self): + env = BlokusEnv(num_players=4, board_size=20) + assert env.num_players == 4 + assert env.board_size == 20 + + def test_observation_space(self): + env = BlokusEnv(num_players=4, board_size=20) + obs, _ = env.reset(seed=42) + assert "board" in obs + assert "pieces" in obs + assert "corners" in obs + assert obs["board"].shape == (20, 20) + assert obs["pieces"].shape == (21,) + assert obs["corners"].shape == (20, 20) + + def test_action_space(self): + env = BlokusEnv(num_players=4, board_size=20) + assert env.action_space.shape == () + assert env.action_space.n > 0 + + def test_reset_returns_obs_and_info(self): + env = BlokusEnv(num_players=4, board_size=20) + obs, info = env.reset(seed=42) + assert isinstance(obs, dict) + assert isinstance(info, dict) + assert "action_mask" in info + + def test_action_mask_valid(self): + env = BlokusEnv(num_players=4, board_size=20) + obs, info = env.reset(seed=42) + mask = info["action_mask"] + assert mask.dtype == bool + assert mask.sum() > 0 # Should have valid moves + + def test_step_with_valid_action(self): + env = BlokusEnv(num_players=4, board_size=20) + obs, info = env.reset(seed=42) + valid_actions = np.where(info["action_mask"])[0] + action = valid_actions[0] + obs, reward, terminated, truncated, info = env.step(action) + assert isinstance(reward, float) + assert isinstance(terminated, bool) + assert isinstance(truncated, bool) + + def test_step_with_invalid_action(self): + env = BlokusEnv(num_players=4, board_size=20) + obs, info = env.reset(seed=42) + # Use an action that's likely invalid + obs, reward, terminated, truncated, info = env.step(env.action_space.n - 1) + assert reward < 0 # Penalty for invalid action + assert terminated # Should end episode + + def test_two_player_env(self): + env = BlokusEnv(num_players=2, board_size=14) + obs, info = env.reset(seed=42) + assert env.num_players == 2 + + def test_three_player_env(self): + env = BlokusEnv(num_players=3, board_size=20) + obs, info = env.reset(seed=42) + assert env.num_players == 3 + + def test_custom_pieces(self): + custom = [p for p in STANDARD_PIECES if p.size < 5] + env = BlokusEnv(num_players=2, board_size=10, pieces=custom) + obs, info = env.reset(seed=42) + assert obs["pieces"].shape == (len(custom),) + + def test_greedy_bot_opponent(self): + env = BlokusEnv(num_players=2, board_size=7, bot_type=GreedyBot) + obs, info = env.reset(seed=42) + assert env.bots[1] is not None + + def test_game_over(self): + env = BlokusEnv(num_players=2, board_size=5, max_steps=10) + obs, info = env.reset(seed=42) + terminated = False + truncated = False + steps = 0 + while not (terminated or truncated) and steps < 50: + valid = np.where(info["action_mask"])[0] + if len(valid) == 0: + break + obs, reward, terminated, truncated, info = env.step(valid[0]) + steps += 1 + + def test_seed_reproducibility(self): + env1 = BlokusEnv(num_players=2, board_size=7) + env2 = BlokusEnv(num_players=2, board_size=7) + obs1, _ = env1.reset(seed=42) + obs2, _ = env2.reset(seed=42) + np.testing.assert_array_equal(obs1["board"], obs2["board"]) + + def test_render_text(self): + env = BlokusEnv(num_players=2, board_size=7, render_mode="ansi") + obs, info = env.reset(seed=42) + result = env.render() + assert isinstance(result, str) + + +class TestActionMaskWrapper: + def test_wrapper_creation(self): + env = BlokusEnv(num_players=2, board_size=7) + env = ActionMaskWrapper(env) + obs, info = env.reset(seed=42) + assert "observation" in obs + assert "action_mask" in obs + + def test_wrapper_step(self): + env = BlokusEnv(num_players=2, board_size=7) + env = ActionMaskWrapper(env) + obs, info = env.reset(seed=42) + valid = np.where(obs["action_mask"])[0] + obs, reward, terminated, truncated, info = env.step(valid[0]) + assert "observation" in obs + assert "action_mask" in obs + + +class TestMultiAgentEnv: + def test_env_creation(self): + env = BlokusMultiAgentEnv(num_players=2, board_size=7) + assert env.num_players == 2 + assert len(env.agents) == 2 + + def test_reset(self): + env = BlokusMultiAgentEnv(num_players=2, board_size=7) + obs, info = env.reset(seed=42) + assert "player_0" in obs + + def test_step(self): + env = BlokusMultiAgentEnv(num_players=2, board_size=7) + obs, info = env.reset(seed=42) + mask = env.get_action_mask("player_0") + valid = np.where(mask)[0] + obs, rewards, terminations, truncations, infos = env.step(valid[0]) + assert isinstance(rewards, dict) + assert isinstance(terminations, dict) + + def test_observation_space(self): + env = BlokusMultiAgentEnv(num_players=2, board_size=7) + obs, _ = env.reset(seed=42) + for agent in env.agents: + if agent in obs: + assert "board" in obs[agent] + + def test_action_space(self): + env = BlokusMultiAgentEnv(num_players=2, board_size=7) + env.reset(seed=42) + space = env.action_space("player_0") + assert space.n > 0 + + def test_four_player(self): + env = BlokusMultiAgentEnv(num_players=4, board_size=7) + obs, info = env.reset(seed=42) + assert len(env.agents) == 4 + + +class TestRegisteredEnvs: + def test_blokus_v0(self): + env = gym.make("Blokus-v0") + obs, info = env.reset(seed=42) + assert obs["board"].shape == (20, 20) + + def test_blokus_duo_v0(self): + env = gym.make("BlokusDuo-v0") + obs, info = env.reset(seed=42) + assert obs["board"].shape == (14, 14) + + def test_blokus_junior_v0(self): + env = gym.make("BlokusJunior-v0") + obs, info = env.reset(seed=42) + assert obs["board"].shape == (14, 14) + + def test_blokus_simple_v0(self): + env = gym.make("BlokusSimple-v0") + obs, info = env.reset(seed=42) + assert obs["board"].shape == (7, 7) + + def test_blokus_greedy_v0(self): + env = gym.make("BlokusGreedy-v0") + obs, info = env.reset(seed=42) + assert obs["board"].shape == (20, 20) diff --git a/tests/test_game.py b/tests/test_game.py new file mode 100644 index 0000000..19e27f2 --- /dev/null +++ b/tests/test_game.py @@ -0,0 +1,196 @@ +from __future__ import annotations + +import numpy as np + +from blokus_gym.core.bots import GreedyBot, GreedyCornersBot, RandomBot +from blokus_gym.core.game import BlokusGame + + +class TestBlokusGame: + def test_game_creation(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + assert game.board_size == 20 + assert game.num_players == 4 + assert game.current_player == 0 + assert len(game.players) == 4 + + def test_reset(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + assert game.current_player == 0 + for player in game.players: + assert len(player.available_pieces) == 21 + + def test_action_space_size(self): + game = BlokusGame(board_size=20, num_players=4) + assert game.num_actions > 0 + + def test_valid_actions_initial(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + valid = game.get_valid_actions(0) + assert np.any(valid) + + def test_play_first_move_corner(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + valid = game.get_valid_actions(0) + valid_indices = np.where(valid)[0] + action = valid_indices[0] + assert game.play_move(0, action) + assert game.board.grid.sum() > 0 + + def test_invalid_action(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + # Use an action that's likely invalid (very high index) + assert not game.play_move(0, game.num_actions - 1) + + def test_next_player(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + assert game.current_player == 0 + game.next_player() + assert game.current_player == 1 + game.next_player() + assert game.current_player == 2 + game.next_player() + assert game.current_player == 3 + game.next_player() + assert game.current_player == 0 + + def test_has_valid_moves(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + assert game.has_valid_moves(0) + + def test_is_game_over(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + assert not game.is_game_over() + + def test_get_scores(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + scores = game.get_scores() + assert len(scores) == 4 + for score in scores: + assert isinstance(score, (int, float)) + + def test_get_action_mask(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + mask = game.get_action_mask(0) + assert mask.shape == (game.num_actions,) + assert mask.dtype == bool + + def test_get_current_observation(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + obs = game.get_current_observation() + assert "board" in obs + assert "pieces" in obs + assert "corners" in obs + assert obs["board"].shape == (20, 20) + assert obs["pieces"].shape == (21,) + assert obs["corners"].shape == (20, 20) + + def test_two_player_game(self): + game = BlokusGame(board_size=14, num_players=2) + game.reset() + assert game.num_players == 2 + assert game.has_valid_moves(0) + + def test_three_player_game(self): + game = BlokusGame(board_size=20, num_players=3) + game.reset() + assert game.num_players == 3 + + def test_copy(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + game.play_move(0, 0) + game_copy = game.copy() + assert game_copy.current_player == game.current_player + + def test_play_multiple_moves(self): + game = BlokusGame(board_size=20, num_players=2) + game.reset() + # Play a few moves + for _ in range(5): + valid = game.get_valid_actions(game.current_player) + valid_indices = np.where(valid)[0] + if len(valid_indices) > 0: + game.play_move(game.current_player, valid_indices[0]) + game.next_player() + else: + break + + def test_get_winners_not_over(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + assert game.get_winners() is None + + def test_valid_move_first_corner_only(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + from blokus_gym.core.pieces import Move + # First move must be in corner (0, 0) for player 0 + move = Move(piece_id=0, orientation_id=0, x=0, y=0) + assert game.valid_move(0, move) + # Try a non-corner first move + move2 = Move(piece_id=0, orientation_id=0, x=5, y=5) + assert not game.valid_move(0, move2) + + +class TestBots: + def test_random_bot(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + bot = RandomBot(player_idx=0, seed=42) + valid = game.get_valid_actions(0) + action = bot.select_action(game, valid) + assert action is not None + assert valid[action] + + def test_greedy_bot(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + bot = GreedyBot(player_idx=0) + valid = game.get_valid_actions(0) + action = bot.select_action(game, valid) + assert action is not None + assert valid[action] + + def test_greedy_corners_bot(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + bot = GreedyCornersBot(player_idx=0) + valid = game.get_valid_actions(0) + action = bot.select_action(game, valid) + assert action is not None + assert valid[action] + + def test_bot_no_valid_moves(self): + game = BlokusGame(board_size=20, num_players=4) + game.reset() + bot = RandomBot(player_idx=0, seed=42) + valid = np.zeros(game.num_actions, dtype=bool) + action = bot.select_action(game, valid) + assert action is None + + def test_bots_play_game(self): + game = BlokusGame(board_size=7, num_players=2) + game.reset() + bots = [RandomBot(player_idx=i, seed=i) for i in range(2)] + steps = 0 + while not game.is_game_over() and steps < 100: + bot = bots[game.current_player] + valid = game.get_valid_actions(game.current_player) + action = bot.select_action(game, valid) + if action is not None: + game.play_move(game.current_player, action) + game.next_player() + steps += 1 + assert game.is_game_over() diff --git a/tests/test_imports.py b/tests/test_imports.py new file mode 100644 index 0000000..e74bf56 --- /dev/null +++ b/tests/test_imports.py @@ -0,0 +1,74 @@ +from __future__ import annotations + + +def test_package_import(): + import blokus_gym + assert blokus_gym.__version__ == "0.1.0" + + +def test_core_imports(): + from blokus_gym import BlokusGame, Board, Move + assert Board is not None + assert BlokusGame is not None + assert Move is not None + + +def test_env_imports(): + from blokus_gym import BlokusEnv, BlokusMultiAgentEnv + assert BlokusEnv is not None + assert BlokusMultiAgentEnv is not None + + +def test_wrapper_imports(): + from blokus_gym import ActionMaskWrapper + assert ActionMaskWrapper is not None + + +def test_piece_imports(): + from blokus_gym import ( + DUO_PIECES, + JUNIOR_PIECES, + STANDARD_PIECES, + Piece, + PieceSet, + ) + assert len(STANDARD_PIECES) == 21 + assert len(DUO_PIECES) > 0 + assert len(JUNIOR_PIECES) > 0 + assert Piece is not None + assert PieceSet is not None + + +def test_bot_imports(): + from blokus_gym import GreedyBot, GreedyCornersBot, MinimaxBot, RandomBot + assert RandomBot is not None + assert GreedyBot is not None + assert GreedyCornersBot is not None + assert MinimaxBot is not None + + +def test_all_exports(): + import blokus_gym + expected = [ + "BlokusEnv", + "BlokusMultiAgentEnv", + "ActionMaskWrapper", + "Board", + "BlokusGame", + "Move", + "Piece", + "PieceOrientation", + "PieceSet", + "STANDARD_PIECES", + "DUO_PIECES", + "JUNIOR_PIECES", + "generate_orientations", + "Bot", + "RandomBot", + "GreedyBot", + "GreedyCornersBot", + "MinimaxBot", + "__version__", + ] + for name in expected: + assert hasattr(blokus_gym, name), f"Missing export: {name}" diff --git a/tests/test_pieces.py b/tests/test_pieces.py new file mode 100644 index 0000000..b142677 --- /dev/null +++ b/tests/test_pieces.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +from blokus_gym.core.pieces import ( + STANDARD_PIECES, + Piece, + PieceOrientation, + PieceSet, + generate_orientations, +) + + +class TestPiece: + def test_piece_creation(self): + piece = Piece(name="F", squares=frozenset([(0, 0), (1, 0), (1, 1), (2, 1)])) + assert piece.name == "F" + assert piece.size == 4 + assert len(piece.squares) == 4 + + def test_piece_size(self): + piece = Piece(name="I4", squares=frozenset([(0, 0), (1, 0), (2, 0), (3, 0)])) + assert piece.size == 4 + + def test_standard_pieces_count(self): + assert len(STANDARD_PIECES) == 21 + + def test_standard_pieces_total_squares(self): + total = sum(p.size for p in STANDARD_PIECES) + assert total == 89 + + def test_piece_orientations_via_pieceset(self): + piece_set = PieceSet(STANDARD_PIECES) + orientations = piece_set.get_orientations(0) + assert len(orientations) > 0 + for orient in orientations: + assert isinstance(orient, PieceOrientation) + + +class TestPieceSet: + def test_piece_set_creation(self): + piece_set = PieceSet(STANDARD_PIECES) + assert piece_set.num_pieces == 21 + assert len(piece_set.pieces) == 21 + + def test_piece_set_custom(self): + custom = [p for p in STANDARD_PIECES if p.size <= 3] + piece_set = PieceSet(custom) + assert piece_set.num_pieces == len(custom) + + def test_piece_set_get_piece_id(self): + piece_set = PieceSet(STANDARD_PIECES) + piece_id = piece_set.get_piece_id("F5") + assert piece_set.get_piece(piece_id).name == "F5" + + def test_piece_set_get_orientations(self): + piece_set = PieceSet(STANDARD_PIECES) + for i in range(piece_set.num_pieces): + orientations = piece_set.get_orientations(i) + assert len(orientations) > 0 + + +class TestGenerateOrientations: + def test_square_piece_one_orientation(self): + piece = Piece("O4", frozenset([(0, 0), (0, 1), (1, 0), (1, 1)])) + orientations = generate_orientations(piece) + assert len(orientations) == 1 + + def test_line_piece_two_orientations(self): + piece = Piece("I3", frozenset([(0, 0), (1, 0), (2, 0)])) + orientations = generate_orientations(piece) + assert len(orientations) == 2 + + def test_L_piece_four_orientations(self): + piece = Piece("L3", frozenset([(0, 0), (0, 1), (1, 0)])) + orientations = generate_orientations(piece) + assert len(orientations) == 4 + + def test_all_orientations_unique(self): + piece_set = PieceSet(STANDARD_PIECES) + for piece_id in range(piece_set.num_pieces): + orientations = piece_set.get_orientations(piece_id) + unique_cells = set() + for orient in orientations: + key = tuple(sorted(orient.squares)) + assert key not in unique_cells, f"Duplicate orientation in piece {piece_id}" + unique_cells.add(key)