diff --git a/README.md b/README.md index 1c86571..4af67ec 100644 --- a/README.md +++ b/README.md @@ -66,26 +66,44 @@ Notes: ## Quickstart ```python -from hexo import Hexo +from hexo import GameStatus, Hexo -game = Hexo.new() -print(game.turn()) # Player.P2 -legal, reason = game.is_legal_move((1, 0)) -if not legal: - raise ValueError(reason) -game.push((1, 0)) +def move_key(coord: tuple[int, int]) -> tuple[int, int, int]: + # Deterministic "closest to center" move ordering. + return ( + abs(coord[0]) + abs(coord[1]) + abs(coord[0] + coord[1]), + coord[0], + coord[1], + ) -# Hook point for engines: run a search after first placement. -# ... search code here ... -record = game.push((0, 1)) # turn completes here -print(record) # TurnRecord -print(game.moves_left_in_turn()) # 2 +def play_turn(game: Hexo): + first = min(game.legal_moves, key=move_key) + record = game.push(first) + if record is not None: # win can happen on first placement + return record -state = game.to_state() -restored = Hexo.from_state(state) -print(restored.turn()) + second = min(game.legal_moves, key=move_key) + return game.push(second) + + +game = Hexo.new() # P1 center stone is pre-applied at (0, 0) + +for turn_no in range(1, 6): + if game.status() is not GameStatus.ONGOING: + break + record = play_turn(game) + print(f"turn {turn_no}: {record.player.name} {record.placements} won={record.won}") + +print("stones on board:", len(game.board())) +print("next player:", game.turn().name) + +# Useful for bot search trees: snapshot, restore, undo +snapshot = game.to_state() +restored = Hexo.from_state(snapshot) +affected = restored.undo() +print("undo affected turn:", affected) ``` ## Tests, linting and formatting diff --git a/examples/random_match.py b/examples/random_match.py index deefa9f..62332cd 100644 --- a/examples/random_match.py +++ b/examples/random_match.py @@ -6,28 +6,16 @@ from hexo import GameStatus, Hexo -Coord = tuple[int, int] - -def random_engine(game: Hexo, rng: random.Random) -> Coord: - candidates = game.legal_moves - if not candidates: - raise RuntimeError("no legal candidate placements found") - return rng.choice(tuple(candidates)) - - -def play_turn_random(game: Hexo, rng: random.Random) -> tuple[Coord, ...]: - chosen: list[Coord] = [] +def play_turn_random(game: Hexo, rng: random.Random): player = game.turn() - while game.turn() is player and game.status() is GameStatus.ONGOING: - coord = random_engine(game, rng) - game.push(coord) - chosen.append(coord) - return tuple(chosen) + while game.turn() is player: + move = rng.choice(tuple(game.legal_moves)) + game.push(move) @timed -def run_demo_match(seed: int = 42, max_turns: int = 100) -> GameStatus: +def run_demo_match(seed: int = 42, max_turns: int = 2000): logger = get_logger() rng = random.Random(seed) game = Hexo.new() @@ -51,7 +39,6 @@ def run_demo_match(seed: int = 42, max_turns: int = 100) -> GameStatus: ) else: logger.warning(f"Reached max turns ({max_turns}) with status={status.name}") - return status if __name__ == "__main__": diff --git a/pyproject.toml b/pyproject.toml index 05f500c..8340e12 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "hexo" -version = "0.2.0" +version = "0.2.1" description = "A fast, reusable Python engine for Hexo on an infinite hex grid." readme = "README.md" license = {file = "LICENSE"} diff --git a/src/hexo/_engine_utils.py b/src/hexo/_engine_utils.py new file mode 100644 index 0000000..25d1aad --- /dev/null +++ b/src/hexo/_engine_utils.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +from collections.abc import Sequence + +from hexo.types import AXES, Coord, Player + +AXIS_PAIRS: tuple[tuple[Coord, Coord], ...] = tuple( + (axis, (-axis[0], -axis[1])) for axis in AXES +) + + +def hex_distance(a: Coord, b: Coord) -> int: + """ + Compute hex-grid distance between two axial coordinates. + + :param a: + First coordinate as `(q, r)`. + :param b: + Second coordinate as `(q, r)`. + :return: + Hex distance between `a` and `b`. + """ + aq, ar = a + bq, br = b + dq = aq - bq + dr = ar - br + return (abs(dq) + abs(dr) + abs(dq + dr)) // 2 + + +def build_radius_offsets(radius: int) -> tuple[Coord, ...]: + """ + Return all axial offsets within a hex-disc radius. + """ + offsets: list[Coord] = [] + for dq in range(-radius, radius + 1): + dr_min = max(-radius, -dq - radius) + dr_max = min(radius, -dq + radius) + for dr in range(dr_min, dr_max + 1): + offsets.append((dq, dr)) + return tuple(offsets) + + +def expand_legal_cache_from( + anchor: Coord, + radius_offsets: Sequence[Coord], + board: dict[Coord, Player], + legal_moves_cache: set[Coord], +) -> tuple[Coord, ...]: + """ + Expand legal moves around an occupied anchor and return only newly-added cells. + """ + aq, ar = anchor + added: list[Coord] = [] + for dq, dr in radius_offsets: + candidate = (aq + dq, ar + dr) + if candidate in board or candidate in legal_moves_cache: + continue + legal_moves_cache.add(candidate) + added.append(candidate) + return tuple(added) + + +def has_winning_line( + player_cells: set[Coord], newly_placed: Sequence[Coord], win_length: int +) -> bool: + """ + Check whether any newly placed stone completes a line of at least `win_length`. + """ + for center in newly_placed: + if has_winning_line_at(player_cells, center, win_length): + return True + return False + + +def has_winning_line_at( + player_cells: set[Coord], center: Coord, win_length: int +) -> bool: + """ + Check whether one anchor coordinate completes a line of at least `win_length`. + """ + contains = player_cells.__contains__ + center_q, center_r = center + for (fdq, fdr), (bdq, bdr) in AXIS_PAIRS: + run = 1 + + q = center_q + fdq + r = center_r + fdr + while contains((q, r)): + run += 1 + q += fdq + r += fdr + if run >= win_length: + return True + + q = center_q + bdq + r = center_r + bdr + while contains((q, r)): + run += 1 + q += bdq + r += bdr + if run >= win_length: + return True + return False diff --git a/src/hexo/engine.py b/src/hexo/engine.py index 2bc322c..150fb35 100644 --- a/src/hexo/engine.py +++ b/src/hexo/engine.py @@ -2,10 +2,27 @@ from collections.abc import Collection, Iterator, Sequence -from hexo.errors import IllegalTurnError -from hexo.geometry import hex_distance +from hexo._engine_utils import ( + hex_distance, + build_radius_offsets, + expand_legal_cache_from, + has_winning_line_at, +) +from hexo.errors import ( + DUPLICATE_COORDINATES, + GAME_IS_OVER, + ILLEGAL_MOVE, + ILLEGAL_PLACEMENT, + NO_PLACEMENT_TO_UNDO, + PARTIAL_TURN_IN_PROGRESS, + STATE_INVALID_TURN_SIZE, + TURN_ALREADY_HAS_TWO, + TURN_MUST_PLACE_TWO, + IllegalTurnError, + occupied_cell, + placement_out_of_radius, +) from hexo.types import ( - AXES, Coord, EngineConfig, GameStatus, @@ -15,6 +32,8 @@ UndoSnapshot, ) +SUBMOVES_PER_TURN = 2 + class LegalMovesView(Collection[Coord]): """ @@ -99,8 +118,13 @@ def __init__(self, config: EngineConfig | None = None) -> None: self._to_move = Player.P2 self._legal_moves_cache: set[Coord] = set() self._legal_moves_view = LegalMovesView(self) - self._radius_offsets = self._build_radius_offsets(self.config.placement_radius) - self._expand_legal_cache_from(self.config.opening_center) + self._radius_offsets = build_radius_offsets(self.config.placement_radius) + expand_legal_cache_from( + self.config.opening_center, + self._radius_offsets, + self._board, + self._legal_moves_cache, + ) @classmethod def new(cls, config: EngineConfig | None = None) -> Hexo: @@ -127,15 +151,14 @@ def from_state(cls, state: State, config: EngineConfig | None = None) -> Hexo: A reconstructed `Hexo` game after replaying all serialized moves. """ game = cls.new(config=config) + push = game.push for raw_turn in state.get("turns", []): if len(raw_turn) not in (1, 2): - raise IllegalTurnError( - "state contains invalid turn size; each turn must have 1 or 2 coordinates" - ) + raise IllegalTurnError(STATE_INVALID_TURN_SIZE) for raw_coord in raw_turn: - game.push((int(raw_coord[0]), int(raw_coord[1]))) + push((int(raw_coord[0]), int(raw_coord[1]))) for raw_coord in state.get("pending", []): - game.push((int(raw_coord[0]), int(raw_coord[1]))) + push((int(raw_coord[0]), int(raw_coord[1]))) return game def to_state(self) -> State: @@ -168,7 +191,7 @@ def moves_left_in_turn(self) -> int: :return: Number of stones left to place in the current turn. """ - return 2 - len(self._pending) + return SUBMOVES_PER_TURN - len(self._pending) def pending_moves(self) -> tuple[Coord, ...]: """ @@ -202,34 +225,28 @@ def is_legal(self, move: Sequence[Coord]) -> tuple[bool, str | None]: Tuple `(is_legal, reason)` where `reason` is `None` when legal. """ if self._winner is not None: - return False, "game is over" + return False, GAME_IS_OVER if self._pending: - return False, "cannot play a full move while a partial turn is in progress" + return False, PARTIAL_TURN_IN_PROGRESS - if len(move) != 2: - return False, "each turn must place exactly 2 stones" + if len(move) != SUBMOVES_PER_TURN: + return False, TURN_MUST_PLACE_TWO a, b = move[0], move[1] if a == b: - return False, "duplicate coordinates in same turn" + return False, DUPLICATE_COORDINATES if a in self._board: - return False, f"occupied cell: {a}" + return False, occupied_cell(a) if b in self._board: - return False, f"occupied cell: {b}" + return False, occupied_cell(b) if a not in self._legal_moves_cache: - return ( - False, - f"placement {a} has no occupied cell within distance <= {self.config.placement_radius}", - ) + return False, placement_out_of_radius(a, self.config.placement_radius) if ( b not in self._legal_moves_cache and hex_distance(a, b) > self.config.placement_radius ): - return ( - False, - f"placement {b} has no occupied cell within distance <= {self.config.placement_radius}", - ) + return False, placement_out_of_radius(b, self.config.placement_radius) return True, None @@ -243,17 +260,14 @@ def is_legal_move(self, coord: Coord) -> tuple[bool, str | None]: Tuple `(is_legal, reason)` where `reason` is `None` when legal. """ if self._winner is not None: - return False, "game is over" - if len(self._pending) >= 2: - return False, "current turn already has two placements" - if coord in self._board: - return False, f"occupied cell: {coord}" + return False, GAME_IS_OVER + if len(self._pending) >= SUBMOVES_PER_TURN: + return False, TURN_ALREADY_HAS_TWO if coord in self._legal_moves_cache: return True, None - return ( - False, - f"placement {coord} has no occupied cell within distance <= {self.config.placement_radius}", - ) + if coord in self._board: + return False, occupied_cell(coord) + return False, placement_out_of_radius(coord, self.config.placement_radius) def play(self, move: Sequence[Coord]) -> TurnRecord: """ @@ -266,7 +280,7 @@ def play(self, move: Sequence[Coord]) -> TurnRecord: """ legal, reason = self.is_legal(move) if not legal: - raise IllegalTurnError(reason or "illegal move") + raise IllegalTurnError(reason or ILLEGAL_MOVE) self.push(move[0]) if self._winner is not None: @@ -288,18 +302,27 @@ def push(self, coord: Coord) -> TurnRecord | None: """ legal, reason = self.is_legal_move(coord) if not legal: - raise IllegalTurnError(reason or "illegal placement") + raise IllegalTurnError(reason or ILLEGAL_PLACEMENT) - prev_to_move = self._to_move + to_move = self._to_move + prev_to_move = to_move prev_winner = self._winner prev_pending = tuple(self._pending) prev_history_len = len(self._turn_history) - removed_from_cache = coord in self._legal_moves_cache - - self._board[coord] = self._to_move - self._stones_by_player[self._to_move].add(coord) - self._legal_moves_cache.discard(coord) - added_legal = self._expand_legal_cache_from(coord) + board = self._board + legal_cache = self._legal_moves_cache + removed_from_cache = coord in legal_cache + + board[coord] = to_move + player_cells = self._stones_by_player[to_move] + player_cells.add(coord) + legal_cache.discard(coord) + added_legal = expand_legal_cache_from( + coord, + self._radius_offsets, + board, + legal_cache, + ) self._pending.append(coord) self._undo_stack.append( ( @@ -309,28 +332,16 @@ def push(self, coord: Coord) -> TurnRecord | None: prev_pending, prev_history_len, removed_from_cache, - tuple(added_legal), + added_legal, ) ) - won = self._has_winning_line(self._to_move, (coord,)) + won = has_winning_line_at(player_cells, coord, self.config.win_length) if won: - record = TurnRecord( - player=self._to_move, placements=tuple(self._pending), won=True - ) - self._winner = self._to_move - self._turn_history.append(record) - self._pending.clear() - return record + return self._finish_turn(won=True) - if len(self._pending) == 2: - record = TurnRecord( - player=self._to_move, placements=tuple(self._pending), won=False - ) - self._turn_history.append(record) - self._pending.clear() - self._to_move = self._to_move.opponent - return record + if len(self._pending) == SUBMOVES_PER_TURN: + return self._finish_turn(won=False) return None @@ -342,7 +353,7 @@ def undo(self) -> TurnRecord: Record of the affected turn before the undo. """ if not self._undo_stack: - raise IllegalTurnError("cannot undo: no placement has been played") + raise IllegalTurnError(NO_PLACEMENT_TO_UNDO) affected = ( self._turn_history[-1] @@ -362,13 +373,14 @@ def undo(self) -> TurnRecord: self._stones_by_player[prev_to_move].discard(coord) self._to_move = prev_to_move self._winner = prev_winner - self._pending = list(prev_pending) + self._pending.clear() + self._pending.extend(prev_pending) if removed_from_cache: self._legal_moves_cache.add(coord) for added in added_legal: self._legal_moves_cache.discard(added) if len(self._turn_history) > prev_history_len: - self._turn_history = self._turn_history[:prev_history_len] + self._turn_history.pop() return affected def at(self, coord: Coord) -> Player | None: @@ -401,85 +413,19 @@ def legal_moves(self) -> Collection[Coord]: """ return self._legal_moves_view - def _expand_legal_cache_from(self, anchor: Coord) -> set[Coord]: + def _finish_turn(self, won: bool) -> TurnRecord: """ - Expand legal move cache from one occupied anchor coordinate. - - :param anchor: - Occupied coordinate used to generate new reachable empty cells. - :return: - Coordinates that were newly introduced in the legal-move cache. - """ - aq, ar = anchor - added: set[Coord] = set() - for dq, dr in self._radius_offsets: - coord = (aq + dq, ar + dr) - if coord in self._board: - continue - if coord not in self._legal_moves_cache: - self._legal_moves_cache.add(coord) - added.add(coord) - return added - - @staticmethod - def _build_radius_offsets(radius: int) -> tuple[Coord, ...]: - """ - Build axial offsets for a hex disc of a given radius. - - :param radius: - Radius in hex distance. - :return: - Tuple of `(dq, dr)` offsets within the radius. - """ - offsets: list[Coord] = [] - for dq in range(-radius, radius + 1): - dr_min = max(-radius, -dq - radius) - dr_max = min(radius, -dq + radius) - for dr in range(dr_min, dr_max + 1): - offsets.append((dq, dr)) - return tuple(offsets) - - def _has_winning_line(self, player: Player, newly_placed: Sequence[Coord]) -> bool: + Finalize the current turn, updating history and turn ownership. """ - Determine whether the latest move creates a winning alignment. - - :param player: - Player whose board alignment should be evaluated. - :param newly_placed: - Two coordinates placed by `player` in the current turn. - :return: - `True` if the move creates a line meeting the win length. - """ - needed = self.config.win_length - player_cells = self._stones_by_player[player] - for center in newly_placed: - for axis in AXES: - run = ( - 1 - + self._count_direction(player_cells, center, axis) - + self._count_direction(player_cells, center, (-axis[0], -axis[1])) - ) - if run >= needed: - return True - return False - - @staticmethod - def _count_direction(cells: set[Coord], origin: Coord, step: Coord) -> int: - """ - Count contiguous stones in one direction from an origin. - - :param cells: - Set of coordinates occupied by one player. - :param origin: - Starting coordinate for the directional scan. - :param step: - Axis step increment used for scanning. - :return: - Number of contiguous cells encountered along the direction. - """ - count = 0 - cur = (origin[0] + step[0], origin[1] + step[1]) - while cur in cells: - count += 1 - cur = (cur[0] + step[0], cur[1] + step[1]) - return count + record = TurnRecord( + player=self._to_move, + placements=tuple(self._pending), + won=won, + ) + self._turn_history.append(record) + self._pending.clear() + if won: + self._winner = self._to_move + else: + self._to_move = self._to_move.opponent + return record diff --git a/src/hexo/errors.py b/src/hexo/errors.py index f9df344..f1969c3 100644 --- a/src/hexo/errors.py +++ b/src/hexo/errors.py @@ -1,3 +1,8 @@ +from __future__ import annotations + +from hexo.types import Coord + + class HexoError(Exception): """ Represent the base exception for all Hexo engine failures. @@ -14,3 +19,30 @@ class IllegalTurnError(HexoError): This exception is raised when validation fails during state reconstruction, move application, or other rule-checked operations. """ + + +STATE_INVALID_TURN_SIZE = ( + "state contains invalid turn size; each turn must have 1 or 2 coordinates" +) +GAME_IS_OVER = "game is over" +PARTIAL_TURN_IN_PROGRESS = "cannot play a full move while a partial turn is in progress" +TURN_MUST_PLACE_TWO = "each turn must place exactly 2 stones" +DUPLICATE_COORDINATES = "duplicate coordinates in same turn" +TURN_ALREADY_HAS_TWO = "current turn already has two placements" +NO_PLACEMENT_TO_UNDO = "cannot undo: no placement has been played" +ILLEGAL_MOVE = "illegal move" +ILLEGAL_PLACEMENT = "illegal placement" + + +def occupied_cell(coord: Coord) -> str: + """ + Return an "occupied cell" validation message for one coordinate. + """ + return f"occupied cell: {coord}" + + +def placement_out_of_radius(coord: Coord, radius: int) -> str: + """ + Return a distance-rule violation message for one coordinate. + """ + return f"placement {coord} has no occupied cell within distance <= {radius}" diff --git a/src/hexo/geometry.py b/src/hexo/geometry.py deleted file mode 100644 index 347284b..0000000 --- a/src/hexo/geometry.py +++ /dev/null @@ -1,17 +0,0 @@ -from hexo.types import Coord - - -def hex_distance(a: Coord, b: Coord) -> int: - """ - Compute hex-grid distance between two axial coordinates. - - :param a: - First coordinate as `(q, r)`. - :param b: - Second coordinate as `(q, r)`. - :return: - Hex distance between `a` and `b`. - """ - dq = a[0] - b[0] - dr = a[1] - b[1] - return (abs(dq) + abs(dr) + abs(dq + dr)) // 2 diff --git a/uv.lock b/uv.lock index c39193f..3b765e3 100644 --- a/uv.lock +++ b/uv.lock @@ -156,7 +156,7 @@ wheels = [ [[package]] name = "hexo" -version = "0.2.0" +version = "0.2.1" source = { editable = "." } dependencies = [ { name = "pedros" },