daniel156161/snake-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(snake): modularize engine and add tournament tools

Browse Source 
- Split active strategies, reusable engine code, core classes, and legacy snakes.
- Replace implicit snake imports with explicit module registrations.
- Extract Prism duel, spatial, and survival behavior into focused mixins.
- Improve duel scoring with food races, pressure, caches, and depth metrics.
- Add deterministic arena scenarios and paired seeded engine tournaments.
- Expand benchmark telemetry and bump Prism to version 1.3.0.
- Update documentation and tests for the new package layout and tooling.
This commit is contained in:
Daniel Dolezal
2026-08-01 20:25:07 +02:00
parent cb6c8d4dc8
commit 3a9af3f54d
39 changed files with 1447 additions and 768 deletions
Download Patch File Download Diff File Expand all files Collapse all files
snakes/engine/__init__.py
+18
View File
@@ -0,0 +1,18 @@
"""Reusable high-performance board and search engines for competitive snakes."""
from snakes.engine.bitboard import BitBoard
from snakes.engine.duel import BitboardDuelMixin
from snakes.engine.duel_search import BitboardDuelSearch, DuelState
from snakes.engine.spatial import BitboardSpatialMixin
from snakes.engine.survival import BitboardSurvivalMixin
from snakes.engine.survival_search import CompactSurvivalSearch
__all__ = (
"BitBoard",
"BitboardDuelMixin",
"BitboardDuelSearch",
"BitboardSpatialMixin",
"BitboardSurvivalMixin",
"CompactSurvivalSearch",
"DuelState",
)
snakes/engine/bitboard.py
+364
View File
@@ -0,0 +1,364 @@
"""Bitboard engine for Battlesnake grid spatial operations.
Cell index = y * width + x. Bit *i* of a Python int represents cell *i*.
All heavy BFS / flood-fill / territory ops run on plain integer arithmetic —
no sets, deques, or per-cell Python objects.
Typical 11×11 board → 121-bit integers. Python big-int ops on these are
extremely fast (single C-level limb operations under the hood).
"""
from __future__ import annotations
class BitBoard:
"""Pre-computed masks and fast spatial primitives for a fixed grid size."""
__slots__ = (
"width", "height", "size", "board_mask",
"_not_rightcol", "_not_leftcol",
"_neighbor_masks",
)
def __init__(self, width: int, height: int) -> None:
self.width = width
self.height = height
self.size = width * height
self.board_mask = (1 << self.size) - 1
# Column masks — prevent bit-shift wrap-around at row boundaries
rightcol = 0
leftcol = 0
for y in range(height):
rightcol |= 1 << (y * width + width - 1)
leftcol |= 1 << (y * width)
self._not_rightcol = self.board_mask & ~rightcol
self._not_leftcol = self.board_mask & ~leftcol
# Per-cell neighbour bitmask (4-connected)
nb = [0] * self.size
for idx in range(self.size):
x, y = idx % width, idx // width
mask = 0
if x > 0:
mask |= 1 << (idx - 1)
if x < width - 1:
mask |= 1 << (idx + 1)
if y > 0:
mask |= 1 << (idx - width)
if y < height - 1:
mask |= 1 << (idx + width)
nb[idx] = mask
self._neighbor_masks = nb
# ── Coordinate helpers ────────────────────────────────────────────────────
def idx(self, x: int, y: int) -> int:
"""(x, y) → flat index."""
return y * self.width + x
def coord(self, flat: int) -> tuple[int, int]:
"""Flat index → (x, y)."""
return flat % self.width, flat // self.width
def pt_bit(self, x: int, y: int) -> int:
"""Single-cell bitmask for (x, y)."""
return 1 << (y * self.width + x)
def set_to_bits(self, points: set[tuple[int, int]]) -> int:
"""Convert a set of (x, y) tuples to a bitmask."""
w = self.width
bits = 0
for x, y in points:
bits |= 1 << (y * w + x)
return bits
def in_bounds(self, x: int, y: int) -> bool:
return 0 <= x < self.width and 0 <= y < self.height
# ── Core spatial primitives ───────────────────────────────────────────────
def flood_fill(self, start_idx: int, blocked_bits: int) -> int:
"""Return bitmask of all cells reachable from *start_idx* (inclusive)."""
free = self.board_mask & ~blocked_bits
start_bit = 1 << start_idx
# If start is blocked, return just itself
if not (start_bit & free):
return start_bit
reachable = start_bit
frontier = start_bit
w = self.width
nrc = self._not_rightcol
nlc = self._not_leftcol
while frontier:
expanded = (
((frontier & nrc) << 1)
| ((frontier & nlc) >> 1)
| (frontier << w)
| (frontier >> w)
) & free & ~reachable
if not expanded:
break
reachable |= expanded
frontier = expanded
return reachable
def flood_count(self, start_idx: int, blocked_bits: int) -> int:
"""Count of cells reachable from *start_idx*."""
return self.flood_fill(start_idx, blocked_bits).bit_count()
def open_neighbor_count(self, cell_idx: int, blocked_bits: int) -> int:
"""Number of free neighbours of *cell_idx*."""
return (self._neighbor_masks[cell_idx] & ~blocked_bits & self.board_mask).bit_count()
def neighbors_of(self, cell_idx: int) -> int:
"""Bitmask of 4-connected neighbours (may include blocked cells)."""
return self._neighbor_masks[cell_idx]
# ── Territory (dual-BFS expansion) ────────────────────────────────────────
def territory(
self,
my_idx: int,
enemy_indices: list[int],
blocked_bits: int,
) -> int:
"""Simultaneous BFS from *my_idx* and all enemies.
Returns Apex-compatible territory over cells reachable from ``my_idx``:
+1 when we arrive first, -1 when an enemy arrives first, and 0 for ties.
Enemy-only disconnected regions are not counted.
"""
if not enemy_indices:
return 0
free = self.board_mask & ~blocked_bits
w = self.width
nrc = self._not_rightcol
nlc = self._not_leftcol
my_front = 1 << my_idx
my_seen = my_front
en_front = 0
for ei in enemy_indices:
en_front |= 1 << ei
en_seen = en_front
# Each side must expand independently. A cell reached at the same depth is
# unclaimed, but it is not a wall: both sides may route through it later.
# Match Apex semantics by scoring only cells reachable from our head:
# ours when we arrive first, theirs when an enemy arrives first, and zero
# on ties. Enemy-only disconnected regions are intentionally ignored.
score = (my_front & ~en_front).bit_count()
enemy_before = 0
while my_front:
my_exp = (
((my_front & nrc) << 1)
| ((my_front & nlc) >> 1)
| (my_front << w)
| (my_front >> w)
) & free & ~my_seen
en_exp = (
((en_front & nrc) << 1)
| ((en_front & nlc) >> 1)
| (en_front << w)
| (en_front >> w)
) & free & ~en_seen
enemy_before |= en_front
score += (my_exp & ~enemy_before & ~en_exp).bit_count()
score -= (my_exp & enemy_before).bit_count()
my_seen |= my_exp
en_seen |= en_exp
my_front = my_exp
en_front = en_exp
return score
# ── Partition sizes (for articulation-point detection) ────────────────────
def partition_sizes(self, cut_idx: int, blocked_bits: int) -> list[int]:
"""Remove *cut_idx* from the free space and return sizes of each
resulting connected component among its neighbours.
Returns an empty list when the point is not a cut vertex (single component
or ≤1 free neighbour).
"""
test_blocked = blocked_bits | (1 << cut_idx)
free_nb = self._neighbor_masks[cut_idx] & ~test_blocked & self.board_mask
if free_nb.bit_count() <= 1:
return []
seen_all = 0
sizes: list[int] = []
temp = free_nb
while temp:
bit = temp & (-temp) # lowest set bit
temp ^= bit
if bit & seen_all:
continue
component = self.flood_fill(bit.bit_length() - 1, test_blocked)
seen_all |= component
sizes.append(component.bit_count())
return sizes if len(sizes) > 1 else []
# ── BFS distance map (indexed by cell idx) ────────────────────────────────
def distance_map(self, start_idx: int, blocked_bits: int) -> dict[int, int]:
"""BFS distance from *start_idx* to every reachable cell.
Returns ``{cell_idx: distance}`` — same semantics as the original
``_distance_map`` but using bitboard expansion internally.
"""
free = self.board_mask & ~blocked_bits
start_bit = 1 << start_idx
distances: dict[int, int] = {start_idx: 0}
frontier = start_bit
seen = frontier
dist = 0
w = self.width
nrc = self._not_rightcol
nlc = self._not_leftcol
while frontier:
dist += 1
expanded = (
((frontier & nrc) << 1)
| ((frontier & nlc) >> 1)
| (frontier << w)
| (frontier >> w)
) & free & ~seen
if not expanded:
break
seen |= expanded
# Extract individual bits
temp = expanded
while temp:
bit = temp & (-temp)
idx = bit.bit_length() - 1
distances[idx] = dist
temp ^= bit
frontier = expanded
return distances
# ── Path distance (BFS to single target) ──────────────────────────────────
def path_distance(
self,
start_idx: int,
goal_idx: int,
blocked_bits: int,
) -> int | None:
"""Shortest path length from *start_idx* to *goal_idx*, or ``None``."""
# Unblock the goal cell so BFS can reach it
free = (self.board_mask & ~blocked_bits) | (1 << goal_idx)
start_bit = 1 << start_idx
goal_bit = 1 << goal_idx
if start_idx == goal_idx:
return 0
frontier = start_bit
seen = frontier
dist = 0
w = self.width
nrc = self._not_rightcol
nlc = self._not_leftcol
while frontier:
dist += 1
expanded = (
((frontier & nrc) << 1)
| ((frontier & nlc) >> 1)
| (frontier << w)
| (frontier >> w)
) & free & ~seen
if not expanded:
break
if expanded & goal_bit:
return dist
seen |= expanded
frontier = expanded
return None
# ── Nearest-food BFS ──────────────────────────────────────────────────────
def nearest_food(
self,
start_idx: int,
food_bits: int,
blocked_bits: int,
) -> tuple[int | None, int | None]:
"""BFS from *start_idx* to nearest food cell.
Food cells are passable even if in *blocked_bits* (matching original
``_nearest_food_info`` semantics).
Returns ``(distance, cell_idx)`` or ``(None, None)``.
"""
if not food_bits:
return None, None
# Food tiles are always steppable
free = (self.board_mask & ~blocked_bits) | food_bits
start_bit = 1 << start_idx
# Check start
if start_bit & food_bits:
return 0, start_idx
# Preserve Apex's deterministic up/down/left/right BFS tie-breaking. A
# pure bit frontier finds the right distance but selects the lowest flat
# index when several foods are equally close, which can change contested-
# food scoring and therefore the selected move.
queue = [start_idx]
seen = start_bit
cursor = 0
layer_end = 1
dist = 0
w = self.width
size = self.size
while cursor < len(queue):
cell = queue[cursor]
cursor += 1
x = cell % w
candidates = (
cell + w,
cell - w,
cell - 1,
cell + 1,
)
for direction, neighbor in enumerate(candidates):
if neighbor < 0 or neighbor >= size:
continue
if direction == 2 and x == 0:
continue
if direction == 3 and x == w - 1:
continue
bit = 1 << neighbor
if bit & seen or not bit & free:
continue
if bit & food_bits:
return dist + 1, neighbor
seen |= bit
queue.append(neighbor)
if cursor == layer_end:
dist += 1
layer_end = len(queue)
return None, None
snakes/engine/duel.py
+89
View File
@@ -0,0 +1,89 @@
"""Reusable compact duel-search integration for Apex-style snakes."""
from __future__ import annotations
from snakes.engine.duel_search import BitboardDuelSearch
class BitboardDuelMixin:
def _new_duel_search(
self, food_set: set, hazard_set: set, hazard_count: dict,
hazard_damage: int, width: int, height: int, deadline: float | None,
) -> BitboardDuelSearch:
if self._duel_search_context is None:
self._duel_search_context = BitboardDuelSearch(
board=self._get_bb(width, height),
food=food_set,
hazards=hazard_set,
hazard_count=hazard_count,
hazard_damage=hazard_damage,
deadline=deadline,
)
return self._duel_search_context
def _minimax_candidate_id(
self, my_body: list, enemy_body: list, my_target: tuple[int, int],
food_set: set, hazard_set: set,
my_health: int, enemy_health: int, hazard_damage: int, hazard_count: dict,
width: int, height: int, max_depth: int, alpha: float, beta: float,
deadline: float | None, previous_hazard_set: set | None = None,
) -> tuple[float, int]:
"""Resolve our selected move and every enemy reply simultaneously."""
search = self._new_duel_search(
food_set, hazard_set, hazard_count, hazard_damage,
width, height, deadline,
)
adaptive_depth = max_depth
remaining = self._remaining_ms(deadline)
if remaining > 250:
adaptive_depth = min(7, max_depth + 1)
elif remaining < 120:
adaptive_depth = min(max_depth, 2)
return search.search_candidate(
my_body=my_body,
enemy_body=enemy_body,
my_target=my_target,
my_health=my_health,
enemy_health=enemy_health,
max_depth=adaptive_depth,
previous_hazards=previous_hazard_set if previous_hazard_set is not None else hazard_set,
)
def _minimax_sim_id(
self, my_body: list, enemy_body: list, food_set: set, hazard_set: set,
my_health: int, enemy_health: int, hazard_damage: int, hazard_count: dict,
width: int, height: int, max_depth: int, alpha: float, beta: float,
deadline: float | None, previous_hazard_set: set | None = None,
) -> tuple[float, int]:
"""Run iterative deepening with one reusable compact search context."""
search = self._new_duel_search(
food_set, hazard_set, hazard_count, hazard_damage,
width, height, deadline,
)
return search.search(
my_body=my_body,
enemy_body=enemy_body,
my_health=my_health,
enemy_health=enemy_health,
max_depth=max_depth,
previous_hazards=previous_hazard_set if previous_hazard_set is not None else hazard_set,
)
def _minimax_sim(
self, my_body: list, enemy_body: list, food_set: set, hazard_set: set,
my_health: int, enemy_health: int, hazard_damage: int, hazard_count: dict,
width: int, height: int, depth: int, alpha: float, beta: float,
deadline: float | None, previous_hazard_set: set | None = None,
) -> float:
"""Compatibility entry point for tests and callers requesting one depth."""
search = self._new_duel_search(
food_set, hazard_set, hazard_count, hazard_damage,
width, height, deadline,
)
return search.search_depth(
my_body=my_body,
enemy_body=enemy_body,
my_health=my_health,
enemy_health=enemy_health,
depth=depth,
previous_hazards=previous_hazard_set if previous_hazard_set is not None else hazard_set,
)
snakes/engine/duel_search.py
+453
View File
@@ -0,0 +1,453 @@
"""Deadline-aware simultaneous duel search using compact tuple bodies and bitboards."""
from __future__ import annotations
from collections.abc import Iterable
from dataclasses import dataclass
from time import perf_counter
from snakes.engine.bitboard import BitBoard
Body = tuple[int, ...]
@dataclass(frozen=True, slots=True)
class DuelState:
my_body: Body
enemy_body: Body
food_bits: int
my_health: int
enemy_health: int
previous_hazard_bits: int
class BitboardDuelSearch:
"""Iterative-deepening paranoid minimax for a two-snake game.
The public API still accepts Battlesnake body dictionaries. Search nodes use
flat cell indices, immutable tuples, and integer masks to avoid allocation of
coordinate dictionaries and sets in the hot path.
"""
WIN = 100_000.0
LOSS = -100_000.0
def __init__(
self,
board: BitBoard,
food: Iterable[tuple[int, int]],
hazards: Iterable[tuple[int, int]],
hazard_count: dict[tuple[int, int], int],
hazard_damage: int,
deadline: float | None,
) -> None:
self.board = board
self.deadline = deadline
self.hazard_damage = hazard_damage
self.food_bits = board.set_to_bits(set(food))
self.hazard_bits = board.set_to_bits(set(hazards))
self.hazard_stacks = {
board.idx(x, y): count for (x, y), count in hazard_count.items()
}
self.transposition: dict[tuple[DuelState, int], tuple[float, str, int | None]] = {}
self.killer_moves: dict[int, int] = {}
self.history: dict[int, int] = {}
self._body_bits_cache: dict[Body, int] = {}
self._evaluation_cache: dict[DuelState, float] = {}
self.nodes = 0
self.cache_hits = 0
self.evaluation_cache_hits = 0
self.completed_depth = 0
self.deadline_exits = 0
def body_from_dicts(self, body: list[dict]) -> Body:
return tuple(self.board.idx(seg["x"], seg["y"]) for seg in body)
def search(
self,
my_body: list[dict],
enemy_body: list[dict],
my_health: int,
enemy_health: int,
max_depth: int,
previous_hazards: Iterable[tuple[int, int]],
) -> tuple[float, int]:
state = DuelState(
my_body=self.body_from_dicts(my_body),
enemy_body=self.body_from_dicts(enemy_body),
food_bits=self.food_bits,
my_health=my_health,
enemy_health=enemy_health,
previous_hazard_bits=self.board.set_to_bits(set(previous_hazards)),
)
result = self._evaluate(state)
completed_depth = 0
for depth in range(1, max_depth + 1):
if self._out_of_time(5.0):
break
window = 80.0 if completed_depth else float("inf")
alpha, beta = result - window, result + window
value, completed = self._search(state, depth, alpha, beta)
if completed and window != float("inf") and (value <= alpha or value >= beta):
value, completed = self._search(state, depth, -float("inf"), float("inf"))
if not completed:
break
result = value
completed_depth = depth
self.completed_depth = max(self.completed_depth, completed_depth)
return result, completed_depth
def search_candidate(
self,
my_body: list[dict],
enemy_body: list[dict],
my_target: tuple[int, int],
my_health: int,
enemy_health: int,
max_depth: int,
previous_hazards: Iterable[tuple[int, int]],
) -> tuple[float, int]:
"""Evaluate one selected move against every simultaneous enemy reply.
``max_depth`` counts the selected root turn, so a completed depth of one
means all opponent replies to that move were resolved.
"""
state = DuelState(
my_body=self.body_from_dicts(my_body),
enemy_body=self.body_from_dicts(enemy_body),
food_bits=self.food_bits,
my_health=my_health,
enemy_health=enemy_health,
previous_hazard_bits=self.board.set_to_bits(set(previous_hazards)),
)
target_idx = self.board.idx(my_target[0], my_target[1])
if not self.board.neighbors_of(state.my_body[0]) & (1 << target_idx):
return self.LOSS, 0
result = self._evaluate(state)
completed_depth = 0
for depth in range(1, max_depth + 1):
if self._out_of_time(5.0):
break
window = 80.0 if completed_depth else float("inf")
alpha, beta = result - window, result + window
value, completed = self._search_selected_move(state, target_idx, depth, alpha, beta)
if completed and window != float("inf") and (value <= alpha or value >= beta):
value, completed = self._search_selected_move(
state, target_idx, depth, -float("inf"), float("inf")
)
if not completed:
break
result = value
completed_depth = depth
self.completed_depth = max(self.completed_depth, completed_depth)
return result, completed_depth
def search_depth(
self,
my_body: list[dict],
enemy_body: list[dict],
my_health: int,
enemy_health: int,
depth: int,
previous_hazards: Iterable[tuple[int, int]],
) -> float:
state = DuelState(
my_body=self.body_from_dicts(my_body),
enemy_body=self.body_from_dicts(enemy_body),
food_bits=self.food_bits,
my_health=my_health,
enemy_health=enemy_health,
previous_hazard_bits=self.board.set_to_bits(set(previous_hazards)),
)
value, completed = self._search(state, depth, -float("inf"), float("inf"))
if completed:
self.completed_depth = max(self.completed_depth, depth)
return value
def _search_selected_move(
self,
state: DuelState,
my_target: int,
depth: int,
alpha: float,
beta: float,
) -> tuple[float, bool]:
"""Resolve the selected root move with the opponent on the same turn."""
self.nodes += 1
if self._out_of_time():
return self._evaluate(state), False
enemy_moves = self._candidate_targets(state.enemy_body)
if not enemy_moves:
return self.WIN + depth, True
enemy_moves = self._ordered_moves(enemy_moves, state, depth, False)
worst = float("inf")
for enemy_target in enemy_moves:
if self._out_of_time():
return (worst if worst != float("inf") else self._evaluate(state)), False
child, terminal = self._advance(state, my_target, enemy_target)
if terminal is not None:
value = terminal
completed = True
elif depth <= 1:
value = self._evaluate(child)
completed = True
else:
value, completed = self._search(child, depth - 1, alpha, beta)
if not completed:
return (worst if worst != float("inf") else value), False
worst = min(worst, value)
beta = min(beta, worst)
if beta <= alpha:
break
return worst, True
def _search(self, state: DuelState, depth: int, alpha: float, beta: float) -> tuple[float, bool]:
self.nodes += 1
if self._out_of_time():
return self._evaluate(state), False
if depth <= 0:
return self._evaluate(state), True
cache_key = (state, depth)
original_alpha, original_beta = alpha, beta
cached = self.transposition.get(cache_key)
if cached is not None:
self.cache_hits += 1
cached_value, bound, preferred_move = cached
if bound == "exact":
return cached_value, True
if bound == "lower":
alpha = max(alpha, cached_value)
else:
beta = min(beta, cached_value)
if alpha >= beta:
return cached_value, True
my_moves = self._candidate_targets(state.my_body)
enemy_moves = self._candidate_targets(state.enemy_body)
if not my_moves:
return self.LOSS - depth, True
if not enemy_moves:
return self.WIN + depth, True
my_moves = self._ordered_moves(my_moves, state, depth, True, preferred_move if cached is not None else None)
enemy_moves = self._ordered_moves(enemy_moves, state, depth, False)
best = -float("inf")
best_move: int | None = None
for my_target in my_moves:
worst = float("inf")
for enemy_target in enemy_moves:
if self._out_of_time():
return (best if best != -float("inf") else self._evaluate(state)), False
child, terminal = self._advance(state, my_target, enemy_target)
if terminal is not None:
value = terminal
completed = True
else:
value, completed = self._search(child, depth - 1, alpha, beta)
if not completed:
return (best if best != -float("inf") else value), False
worst = min(worst, value)
if worst <= alpha:
self.killer_moves[depth] = my_target
self.history[my_target] = self.history.get(my_target, 0) + depth * depth
break
if worst > best:
best = worst
best_move = my_target
alpha = max(alpha, best)
if alpha >= beta:
break
if best <= original_alpha:
bound = "upper"
elif best >= original_beta:
bound = "lower"
else:
bound = "exact"
self.transposition[cache_key] = (best, bound, best_move)
return best, True
def _advance(self, state: DuelState, my_target: int, enemy_target: int) -> tuple[DuelState, float | None]:
my_ate = bool((1 << my_target) & state.food_bits)
enemy_ate = bool((1 << enemy_target) & state.food_bits)
my_body = self._advance_body(state.my_body, my_target, my_ate)
enemy_body = self._advance_body(state.enemy_body, enemy_target, enemy_ate)
my_dead = my_target in my_body[1:] or my_target in enemy_body[1:]
enemy_dead = enemy_target in enemy_body[1:] or enemy_target in my_body[1:]
if my_target == enemy_target:
if len(my_body) <= len(enemy_body):
my_dead = True
if len(enemy_body) <= len(my_body):
enemy_dead = True
my_health = 100 if my_ate else state.my_health - 1
enemy_health = 100 if enemy_ate else state.enemy_health - 1
if not my_ate:
my_health -= self._hazard_cost(my_target, state.previous_hazard_bits)
if not enemy_ate:
enemy_health -= self._hazard_cost(enemy_target, state.previous_hazard_bits)
my_dead = my_dead or my_health <= 0
enemy_dead = enemy_dead or enemy_health <= 0
if my_dead and enemy_dead:
return state, -500.0
if my_dead:
return state, self.LOSS
if enemy_dead:
return state, self.WIN
eaten_bits = 0
if my_ate:
eaten_bits |= 1 << my_target
if enemy_ate:
eaten_bits |= 1 << enemy_target
child = DuelState(
my_body=my_body,
enemy_body=enemy_body,
food_bits=state.food_bits & ~eaten_bits,
my_health=my_health,
enemy_health=enemy_health,
previous_hazard_bits=self.hazard_bits,
)
return child, None
def _candidate_targets(self, body: Body) -> list[int]:
"""Return in-bounds targets; `_advance` resolves simultaneous collisions.
Delaying occupancy checks until both targets and food growth are known is
essential: whether either tail vacates depends on that snake eating.
"""
return list(self._iter_bits(self.board.neighbors_of(body[0])))
def _ordered_moves(
self, moves: list[int], state: DuelState, depth: int, mine: bool,
preferred: int | None = None,
) -> list[int]:
body = state.my_body if mine else state.enemy_body
other = state.enemy_body if mine else state.my_body
killer = self.killer_moves.get(depth)
center_x = (self.board.width - 1) / 2.0
center_y = (self.board.height - 1) / 2.0
def score(target: int) -> tuple[float, int]:
x, y = self.board.coord(target)
food_bonus = 200.0 if (1 << target) & state.food_bits else 0.0
space = self.board.flood_count(target, (self._body_bits(body[1:]) | self._body_bits(other[1:])) & ~(1 << target))
center = -(abs(x - center_x) + abs(y - center_y))
preferred_bonus = 20_000.0 if target == preferred else 0.0
killer_bonus = 10_000.0 if target == killer else 0.0
return preferred_bonus + killer_bonus + self.history.get(target, 0) + food_bonus + space * 2.0 + center, -target
# Our strongest-looking moves first; enemy ordering uses the same quality
# estimate because dangerous enemy replies tend to gain space and food.
return sorted(moves, key=score, reverse=True)
def _evaluate(self, state: DuelState) -> float:
cached = self._evaluation_cache.get(state)
if cached is not None:
self.evaluation_cache_hits += 1
return cached
my_blocked = self._body_bits(state.my_body[1:]) | self._body_bits(state.enemy_body[1:])
my_head, enemy_head = state.my_body[0], state.enemy_body[0]
my_space = self.board.flood_count(my_head, my_blocked)
enemy_space = self.board.flood_count(enemy_head, my_blocked)
my_liberties = self.board.open_neighbor_count(my_head, my_blocked)
enemy_liberties = self.board.open_neighbor_count(enemy_head, my_blocked)
territory = self.board.territory(my_head, [enemy_head], my_blocked)
my_tail_path = self.board.path_distance(my_head, state.my_body[-1], my_blocked)
enemy_tail_path = self.board.path_distance(enemy_head, state.enemy_body[-1], my_blocked)
tail_score = (12.0 if my_tail_path is not None else -24.0) - (12.0 if enemy_tail_path is not None else -24.0)
my_hazard = self._hazard_cost(my_head, state.previous_hazard_bits)
enemy_hazard = self._hazard_cost(enemy_head, state.previous_hazard_bits)
length_delta = len(state.my_body) - len(state.enemy_body)
length_score = length_delta * 20.0
health_score = (state.my_health - state.enemy_health) * 0.18
forced_score = (my_liberties > 1) * 10.0 - (enemy_liberties > 1) * 10.0
# Food races matter most when health is low or eating changes head-to-head
# priority. Compare actual path lengths rather than Manhattan distance so a
# food tile behind a body wall is not treated as reachable.
food_score = 0.0
if state.food_bits:
my_food = self.board.nearest_food(my_head, state.food_bits, my_blocked)
enemy_food = self.board.nearest_food(enemy_head, state.food_bits, my_blocked)
my_distance = my_food[0] if my_food[0] is not None else 200
enemy_distance = enemy_food[0] if enemy_food[0] is not None else 200
my_urgency = max(0.0, (55.0 - state.my_health) / 55.0)
enemy_urgency = max(0.0, (55.0 - state.enemy_health) / 55.0)
food_score += (enemy_distance - my_distance) * 2.5
food_score -= my_distance * my_urgency * 5.0
food_score += enemy_distance * enemy_urgency * 3.0
if length_delta == 0 and my_distance < enemy_distance:
food_score += 14.0
elif length_delta < 0 and my_distance <= enemy_distance:
food_score += 20.0
# Reward maintaining safe pressure around the opposing head. This captures
# two-turn head traps that raw territory and flood counts often score as a
# neutral position.
head_distance = self.board.path_distance(my_head, enemy_head, my_blocked)
pressure_score = 0.0
if head_distance is not None and head_distance <= 3:
pressure = (4 - head_distance) * 6.0
pressure_score = pressure if length_delta > 0 else -pressure if length_delta < 0 else 0.0
value = (
(my_space - enemy_space) * 1.5 + territory * 1.2
+ (my_liberties - enemy_liberties) * 14.0 + length_score + health_score
+ tail_score + forced_score + food_score + pressure_score
+ (enemy_hazard - my_hazard) * 0.8
)
if len(self._evaluation_cache) < 32_768:
self._evaluation_cache[state] = value
return value
def _hazard_cost(self, target: int, previous_hazard_bits: int) -> int:
bit = 1 << target
if not (bit & self.hazard_bits & previous_hazard_bits):
return 0
return self.hazard_damage * self.hazard_stacks.get(target, 1)
@staticmethod
def _advance_body(body: Body, target: int, ate: bool) -> Body:
return (target,) + body if ate else (target,) + body[:-1]
@staticmethod
def _tail_stacked(body: Body) -> bool:
return len(body) >= 2 and body[-1] == body[-2]
def _body_bits(self, body: Body) -> int:
cached = self._body_bits_cache.get(body)
if cached is not None:
return cached
bits = 0
for cell in body:
bits |= 1 << cell
if len(self._body_bits_cache) < 16_384:
self._body_bits_cache[body] = bits
return bits
@staticmethod
def _iter_bits(bits: int):
while bits:
bit = bits & -bits
yield bit.bit_length() - 1
bits ^= bit
def _out_of_time(self, reserve_ms: float = 0.0) -> bool:
if self.deadline is None:
return False
expired = perf_counter() + reserve_ms / 1000.0 >= self.deadline
if expired:
self.deadline_exits += 1
return expired
snakes/engine/spatial.py
+216
View File
@@ -0,0 +1,216 @@
"""Bitboard-backed spatial primitives shared by competitive snakes."""
from __future__ import annotations
from snakes.engine.bitboard import BitBoard
class BitboardSpatialMixin:
def _get_bb(self, width: int, height: int) -> BitBoard:
"""Return (possibly cached) BitBoard for the current dimensions."""
if self._bb is None or width != self._bb_w or height != self._bb_h:
self._bb = BitBoard(width, height)
self._bb_w = width
self._bb_h = height
return self._bb
def _blocked_to_bits(self, blocked: set[tuple[int, int]], width: int, height: int) -> int:
"""Convert blocked cells to bits without stale identity-based caching."""
return self._get_bb(width, height).set_to_bits(blocked)
def _flood_fill_count(self, start: tuple, blocked: set, width: int, height: int) -> int:
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
start_idx = bb.idx(start[0], start[1])
# A7/E2: per-turn transposition cache (kept from Apex)
cache_key = (start_idx, blocked_bits, width, height)
cached = self._bfs_cache.get(cache_key)
if cached is not None:
return cached
result = bb.flood_count(start_idx, blocked_bits)
if len(self._bfs_cache) < self._bfs_cache_max:
self._bfs_cache[cache_key] = result
return result
def _territory_fast(
self, my_pos: tuple, blocked: set, width: int, height: int,
deadline: float | None = None,
) -> int:
if not self._enemy_heads:
return 0
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
my_idx = bb.idx(my_pos[0], my_pos[1])
enemy_idxs = [bb.idx(eh[0], eh[1]) for eh in self._enemy_heads]
return bb.territory(my_idx, enemy_idxs, blocked_bits)
def _articulation_penalty(
self, point: tuple, blocked: set, width: int, height: int, required_space: int,
) -> float:
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
point_idx = bb.idx(point[0], point[1])
sizes = bb.partition_sizes(point_idx, blocked_bits)
if not sizes:
return 0.0
min_size = min(sizes)
if min_size < required_space:
return 1500.0
elif min_size < required_space * 2:
return 400.0
else:
return 85.0
def _bounded_bfs(self, start: tuple, blocked: set, width: int, height: int, limit: int) -> set:
"""Bitboard-accelerated bounded BFS. Returns a set for API compatibility."""
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
start_idx = bb.idx(start[0], start[1])
reachable_bits = bb.flood_fill(start_idx, blocked_bits)
result: set[tuple[int, int]] = set()
temp = reachable_bits
w = bb.width
while temp:
bit = temp & (-temp)
idx = bit.bit_length() - 1
result.add((idx % w, idx // w))
temp ^= bit
if len(result) >= limit:
break
return result
def _distance_map(self, start: tuple, blocked: set, width: int, height: int) -> dict:
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
start_idx = bb.idx(start[0], start[1])
idx_dmap = bb.distance_map(start_idx, blocked_bits)
w = bb.width
return {(idx % w, idx // w): d for idx, d in idx_dmap.items()}
def _path_distance(
self, start: tuple, goal: tuple, blocked: set, width: int, height: int,
) -> int | None:
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
return bb.path_distance(
bb.idx(start[0], start[1]),
bb.idx(goal[0], goal[1]),
blocked_bits,
)
def _nearest_food_info(
self, start: tuple, food_set: set, blocked: set, width: int, height: int,
) -> tuple[int | None, tuple | None]:
if not food_set:
return None, None
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
food_bits = bb.set_to_bits(food_set)
start_idx = bb.idx(start[0], start[1])
dist, cell_idx = bb.nearest_food(start_idx, food_bits, blocked_bits)
if dist is None or cell_idx is None:
return None, None
return dist, bb.coord(cell_idx)
def _open_neighbor_count(self, start: tuple, blocked: set, width: int, height: int) -> int:
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
return bb.open_neighbor_count(bb.idx(start[0], start[1]), blocked_bits)
def _next_turn_options(self, head: dict, blocked: set, width: int, height: int) -> int:
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
return bb.open_neighbor_count(bb.idx(head["x"], head["y"]), blocked_bits)
def _legal_moves(
self, my_head, my_body: list, other_snakes: list,
food_set: set, is_constrictor: bool, width: int, height: int,
enemy_can_grow: dict | None = None,
):
"""S10: Bitboard-accelerated legal move generation."""
bb = self._get_bb(width, height)
w = bb.width
# Build occupied bitboard
occupied = 0
for seg in my_body:
occupied |= 1 << (seg["y"] * w + seg["x"])
for snake in other_snakes:
for seg in snake["body"]:
occupied |= 1 << (seg["y"] * w + seg["x"])
hx, hy = my_head["x"], my_head["y"]
head_idx = hy * w + hx
# Own tail can be stepped on
passable = 0
if not is_constrictor and len(my_body) >= 2:
t, t2 = my_body[-1], my_body[-2]
if not (t["x"] == t2["x"] and t["y"] == t2["y"]):
passable |= 1 << (t["y"] * w + t["x"])
# Enemy tails that will vacate
if not is_constrictor:
for snake in other_snakes:
sbody = snake["body"]
if len(sbody) < 2:
continue
st, st2 = sbody[-1], sbody[-2]
if st["x"] == st2["x"] and st["y"] == st2["y"]:
continue # stacked
sid = snake.get("id")
can_grow = None
if enemy_can_grow is not None and sid is not None:
can_grow = enemy_can_grow.get(sid)
if can_grow is None:
can_grow = self._enemy_can_grow_this_turn(snake, food_set)
if not can_grow:
passable |= 1 << (st["y"] * w + st["x"])
legal = bb._neighbor_masks[head_idx] & ((~occupied & bb.board_mask) | passable)
safe: dict[str, dict[str, int]] = {}
for name, (dx, dy) in self.DIRECTIONS.items():
nx, ny = hx + dx, hy + dy
if 0 <= nx < w and 0 <= ny < bb.height:
if (1 << (ny * w + nx)) & legal:
safe[name] = {"x": nx, "y": ny}
return safe
def _enemy_confinement_metrics(
self, enemy_head: tuple, blocked: set, width: int, height: int,
) -> tuple[int, int]:
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
eh_idx = bb.idx(enemy_head[0], enemy_head[1])
eb_bits = blocked_bits & ~(1 << eh_idx)
space = bb.flood_count(eh_idx, eb_bits)
options = bb.open_neighbor_count(eh_idx, eb_bits)
return space, options
def _enemy_constrictor_projection(
self, other_snakes: list, blocked: set, width: int, height: int,
) -> tuple[int, int]:
bb = self._get_bb(width, height)
blocked_bits = self._blocked_to_bits(blocked, width, height)
best_space = 0
total_opts = 0
for enemy in other_snakes:
eh = (enemy["head"]["x"], enemy["head"]["y"])
eh_idx = bb.idx(eh[0], eh[1])
nb = bb.neighbors_of(eh_idx) & ~blocked_bits & bb.board_mask
temp = nb
while temp:
total_opts += 1
bit = temp & (-temp)
n_idx = bit.bit_length() - 1
sp = bb.flood_count(n_idx, blocked_bits | bit)
if sp > best_space:
best_space = sp
temp ^= bit
return best_space, total_opts
snakes/engine/survival.py
+214
View File
@@ -0,0 +1,214 @@
"""Reusable bitboard and adversarial survival rollouts."""
from __future__ import annotations
from time import perf_counter
from snakes.engine.survival_search import CompactSurvivalSearch
class BitboardSurvivalMixin:
def _future_rollout_bonus(
self, move: str, safe_moves: dict, my_body: list, other_snakes: list,
food_set: set, is_constrictor: bool, width: int, height: int,
enemy_can_grow: dict, deadline: float | None,
) -> float:
pos = safe_moves.get(move)
if pos is None:
return -250.0
# Duel minimax already advances the opponent exactly. Keep the much faster
# bitboard-native solo rollout here instead of paying for the same response
# model twice. Constrictor and multiplayer still use adversarial rollouts.
if len(other_snakes) == 1 and not is_constrictor:
return super()._future_rollout_bonus(
move, safe_moves, my_body, other_snakes, food_set, is_constrictor,
width, height, enemy_can_grow, deadline,
)
if self._survival_search_context is None:
remaining = self._remaining_ms(deadline)
enemy_branch = 2 if len(other_snakes) <= 2 and remaining > 100 else 1
self._survival_search_context = CompactSurvivalSearch(
board=self._get_bb(width, height),
food=food_set,
is_constrictor=is_constrictor,
deadline=deadline,
branch=self._planning_branch,
enemy_branch=enemy_branch,
response_cap=8 if remaining > 150 else 4,
)
remaining = self._remaining_ms(deadline)
depth = min(self._planning_depth, 2 if len(other_snakes) > 1 else 3)
if remaining < 90:
depth = min(depth, 2)
elif remaining > 250 and len(other_snakes) <= 2:
depth = min(4, depth + 1)
raw = self._survival_search_context.search_selected(
my_body=my_body,
enemies=other_snakes,
target=(pos["x"], pos["y"]),
depth=depth,
)
return raw * 0.15
def _future_position_score(
self, my_body: list, other_snakes: list, food_set: set, is_constrictor: bool,
width: int, height: int, enemy_can_grow: dict, deadline: float | None,
) -> float:
"""S9: Bitboard-native position scoring for the survival tree.
Builds blocked bitboard directly from body lists (no intermediate set).
Uses precomputed enemy bits instead of rebuilding attack map per node.
"""
if deadline is not None and perf_counter() >= deadline:
return 0.0
bb = self._bb # already initialised in choose_move
w = bb.width
head = my_body[0]
hx, hy = head["x"], head["y"]
head_idx = hy * w + hx
head_bit = 1 << head_idx
body_len = len(my_body)
# ── Build blocked bitboard directly (no set) ──────────────────────
my_bits = 0
for seg in my_body:
my_bits |= 1 << (seg["y"] * w + seg["x"])
# Own tail vacates unless stacked or constrictor
if not is_constrictor and body_len >= 2:
t, t2 = my_body[-1], my_body[-2]
if not (t["x"] == t2["x"] and t["y"] == t2["y"]):
my_bits &= ~(1 << (t["y"] * w + t["x"]))
# Enemy body (precomputed) minus vacating tails
en_bits = self._enemy_body_bits & ~self._enemy_tail_bits
blocked_bits = (my_bits | en_bits) & ~head_bit
# ── Reachable space ───────────────────────────────────────────────
reachable = bb.flood_count(head_idx, blocked_bits)
required = body_len + max(3, body_len // 6) if is_constrictor else body_len
if reachable < required:
return -5000.0
# ── Open neighbours (liberties) ───────────────────────────────────
nb_free = bb._neighbor_masks[head_idx] & ~blocked_bits & bb.board_mask
liberties = nb_free.bit_count()
if liberties == 0:
return -5000.0
# ── Safe next options (enemy-attack aware) ────────────────────────
# Rebuild danger for the simulated length. The root-turn danger mask is
# stale after eating and includes enemy moves blocked in this future body.
danger_here = 0
for enemy in other_snakes:
enemy_len = enemy.get("length", len(enemy["body"]))
if enemy_len < body_len:
continue
enemy_head = enemy["head"]
enemy_idx = enemy_head["y"] * w + enemy_head["x"]
danger_here |= bb._neighbor_masks[enemy_idx]
danger_here &= ~blocked_bits
safe_nb = nb_free & ~danger_here
en_safe = safe_nb.bit_count()
if en_safe == 0:
return -4000.0
next_opts = liberties
sc = reachable * 1.9 + liberties * 14.0 + next_opts * 11.0 + en_safe * 26.0
if en_safe == 1:
sc -= 420.0
return sc
def _future_survival_tree(
self, my_body: list, other_snakes: list, food_set: set, is_constrictor: bool,
width: int, height: int, enemy_can_grow: dict,
depth: int, branch: int, deadline: float | None,
) -> float:
"""S9/S11: Bitboard-accelerated survival tree.
Inlines legal-move check with bitboard ops instead of per-direction
Python loops. Uses the bitboard-native _future_position_score.
"""
if depth <= 0 or (deadline is not None and perf_counter() >= deadline):
return 0.0
bb = self._bb
w = bb.width
head = my_body[0]
hx, hy = head["x"], head["y"]
head_idx = hy * w + hx
body_len = len(my_body)
# ── Build occupied bitboard for legal-move check ──────────────────
occupied_bits = 0
for seg in my_body:
occupied_bits |= 1 << (seg["y"] * w + seg["x"])
occupied_bits |= self._enemy_body_bits
# Own tail can be stepped on if not stacked/constrictor
passable = 0
if not is_constrictor and body_len >= 2:
t, t2 = my_body[-1], my_body[-2]
if not (t["x"] == t2["x"] and t["y"] == t2["y"]):
passable |= 1 << (t["y"] * w + t["x"])
# Enemy vacating tails are also steppable
passable |= self._enemy_tail_bits
# Legal moves: free neighbours OR passable tiles
legal_bits = bb._neighbor_masks[head_idx] & ((~occupied_bits & bb.board_mask) | passable)
if not legal_bits:
return -5000.0
# ── Precompute food bitboard once ─────────────────────────────────
food_bits_local = 0
for fx, fy in food_set:
food_bits_local |= 1 << (fy * w + fx)
# ── Score each legal move ─────────────────────────────────────────
scored: list[tuple[float, list]] = []
temp = legal_bits
while temp:
if deadline is not None and perf_counter() >= deadline:
break
bit = temp & (-temp)
temp ^= bit
idx = bit.bit_length() - 1
nx, ny = idx % w, idx // w
pos = {"x": nx, "y": ny}
ate = bool(bit & food_bits_local)
fb = self._future_body(my_body, pos, ate, is_constrictor)
sc = self._future_position_score(
fb, other_snakes, food_set, is_constrictor,
width, height, enemy_can_grow, deadline,
)
scored.append((sc, fb))
if not scored:
return -5000.0
DEATH = self._TREE_DEATH_THRESHOLD
viable = [(sc, fb) for sc, fb in scored if sc > DEATH]
if not viable:
return max(sc for sc, _ in scored)
viable.sort(key=lambda x: x[0], reverse=True)
if depth == 1:
return viable[0][0]
best = viable[0][0]
for sc, fb in viable[:branch]:
if deadline is not None and perf_counter() >= deadline:
break
cont = self._future_survival_tree(
fb, other_snakes, food_set, is_constrictor,
width, height, enemy_can_grow, depth - 1, branch, deadline,
)
total = sc + cont * 0.72
if total > best:
best = total
return best
snakes/engine/survival_search.py
+276
View File
@@ -0,0 +1,276 @@
"""Compact adversarial rollout for multiplayer Battlesnake positions."""
from __future__ import annotations
from itertools import product
from time import perf_counter
from snakes.engine.bitboard import BitBoard
Body = tuple[int, ...]
EnemyBodies = tuple[Body, ...]
StateKey = tuple[Body, EnemyBodies, int, int]
class CompactSurvivalSearch:
"""Small paranoid beam search with simultaneous enemy responses.
It is deliberately narrower than full multiplayer minimax: each enemy keeps
only its most dangerous replies and the combined response beam is capped.
This models moving opponents without exhausting the request deadline.
"""
DEATH = -5000.0
def __init__(
self,
board: BitBoard,
food: set[tuple[int, int]],
is_constrictor: bool,
deadline: float | None,
branch: int,
enemy_branch: int = 2,
response_cap: int = 8,
) -> None:
self.board = board
self.food_bits = board.set_to_bits(food)
self.is_constrictor = is_constrictor
self.deadline = deadline
self.branch = max(1, branch)
self.enemy_branch = max(1, enemy_branch)
self.response_cap = max(1, response_cap)
self.cache: dict[StateKey, float] = {}
self.body_bits_cache: dict[Body, int] = {}
self.nodes = 0
self.cache_hits = 0
self.completed_depth = 0
self.deadline_exits = 0
def body_from_dicts(self, body: list[dict]) -> Body:
return tuple(self.board.idx(segment["x"], segment["y"]) for segment in body)
def search_selected(
self,
my_body: list[dict],
enemies: list[dict],
target: tuple[int, int],
depth: int,
) -> float:
mine = self.body_from_dicts(my_body)
enemy_bodies = tuple(self.body_from_dicts(enemy["body"]) for enemy in enemies)
target_idx = self.board.idx(*target)
if not self.board.neighbors_of(mine[0]) & (1 << target_idx):
return self.DEATH
value, completed = self._selected_root(
mine, enemy_bodies, self.food_bits, target_idx, depth,
)
if completed:
self.completed_depth = max(self.completed_depth, depth)
return value
def _selected_root(
self, mine: Body, enemies: EnemyBodies, food_bits: int, target: int, depth: int,
) -> tuple[float, bool]:
replies = self._enemy_responses(enemies, mine, target, food_bits)
if not replies:
replies = [()]
worst = float("inf")
completed = True
for response in replies:
if self._out_of_time():
completed = False
break
child = self._advance(mine, enemies, target, response, food_bits)
if child is None:
value = self.DEATH
else:
next_mine, next_enemies, next_food = child
value = self._evaluate(next_mine, next_enemies)
if depth > 1 and value > self.DEATH:
value += self._search(next_mine, next_enemies, next_food, depth - 1) * 0.72
worst = min(worst, value)
value = self._evaluate(mine, enemies) if worst == float("inf") else worst
return value, completed
def _search(self, mine: Body, enemies: EnemyBodies, food_bits: int, depth: int) -> float:
self.nodes += 1
if self._out_of_time() or depth <= 0:
return 0.0
key = (mine, enemies, food_bits, depth)
cached = self.cache.get(key)
if cached is not None:
self.cache_hits += 1
return cached
occupied = self._occupied(mine, enemies)
targets = list(self._iter_bits(self.board.neighbors_of(mine[0])))
ranked: list[tuple[float, int]] = []
for target in targets:
# Collision legality is finalized simultaneously because eating controls
# whether tails vacate.
ate = bool((1 << target) & food_bits)
own_tail_blocked = self.is_constrictor or ate
body_blocked = self._body_bits(mine if own_tail_blocked else mine[:-1])
enemy_blocked = 0
for enemy in enemies:
enemy_blocked |= self._body_bits(enemy[:-1] if not self.is_constrictor else enemy)
if (1 << target) & (body_blocked | enemy_blocked):
continue
free_space = self.board.flood_count(target, occupied & ~(1 << target))
ranked.append((free_space + (20 if ate else 0), target))
ranked.sort(reverse=True)
if not ranked:
return self.DEATH
best = self.DEATH
for _, target in ranked[:self.branch]:
replies = self._enemy_responses(enemies, mine, target, food_bits) or [()]
worst = float("inf")
for response in replies:
if self._out_of_time():
break
child = self._advance(mine, enemies, target, response, food_bits)
if child is None:
value = self.DEATH
else:
next_mine, next_enemies, next_food = child
value = self._evaluate(next_mine, next_enemies)
if depth > 1 and value > self.DEATH:
value += self._search(next_mine, next_enemies, next_food, depth - 1) * 0.72
worst = min(worst, value)
if worst != float("inf"):
best = max(best, worst)
if not self._out_of_time() and len(self.cache) < 16_384:
self.cache[key] = best
return best
def _enemy_responses(
self, enemies: EnemyBodies, mine: Body, my_target: int, food_bits: int,
) -> list[tuple[int, ...]]:
if not enemies:
return []
choices: list[list[int]] = []
my_length_after = len(mine) + int(bool((1 << my_target) & food_bits))
for enemy in enemies:
ranked: list[tuple[float, int]] = []
for target in self._iter_bits(self.board.neighbors_of(enemy[0])):
ate = bool((1 << target) & food_bits)
enemy_length_after = len(enemy) + int(ate)
score = 0.0
if target == my_target:
score += 1000.0 if enemy_length_after >= my_length_after else -1000.0
tx, ty = self.board.coord(target)
mx, my = self.board.coord(my_target)
score -= abs(tx - mx) + abs(ty - my)
score += self.board.open_neighbor_count(target, self._occupied(mine, enemies)) * 3.0
score += 20.0 if ate else 0.0
ranked.append((score, target))
ranked.sort(reverse=True)
choices.append([target for _, target in ranked[:self.enemy_branch]])
responses: list[tuple[int, ...]] = []
for response in product(*choices):
responses.append(response)
if len(responses) >= self.response_cap:
break
return responses
def _advance(
self,
mine: Body,
enemies: EnemyBodies,
my_target: int,
enemy_targets: tuple[int, ...],
food_bits: int,
) -> tuple[Body, EnemyBodies, int] | None:
my_ate = bool((1 << my_target) & food_bits)
next_mine = self._advance_body(mine, my_target, my_ate)
next_enemies = tuple(
self._advance_body(body, target, bool((1 << target) & food_bits))
for body, target in zip(enemies, enemy_targets)
)
# Body and self collisions after all tails have moved.
if my_target in next_mine[1:]:
return None
if any(my_target in enemy[1:] for enemy in next_enemies):
return None
surviving: list[Body] = []
for index, enemy in enumerate(next_enemies):
target = enemy[0]
dead = target in enemy[1:] or target in next_mine[1:]
dead = dead or any(
target in other[1:] for other_index, other in enumerate(next_enemies)
if other_index != index
)
if target == my_target:
if len(enemy) >= len(next_mine):
return None
dead = True
if not dead:
# Enemy/enemy head collisions remove equal-length snakes and the shorter.
for other_index, other in enumerate(next_enemies):
if other_index != index and target == other[0] and len(enemy) <= len(other):
dead = True
break
if not dead:
surviving.append(enemy)
eaten = (1 << my_target) if my_ate else 0
for body, target in zip(enemies, enemy_targets):
if (1 << target) & food_bits:
eaten |= 1 << target
return next_mine, tuple(surviving), food_bits & ~eaten
def _evaluate(self, mine: Body, enemies: EnemyBodies) -> float:
blocked = self._occupied(mine, enemies) & ~(1 << mine[0])
space = self.board.flood_count(mine[0], blocked)
liberties = self.board.open_neighbor_count(mine[0], blocked)
if liberties == 0 or space < len(mine):
return self.DEATH
enemy_pressure = 0.0
for enemy in enemies:
enemy_blocked = blocked & ~(1 << enemy[0])
enemy_space = self.board.flood_count(enemy[0], enemy_blocked)
enemy_liberties = self.board.open_neighbor_count(enemy[0], enemy_blocked)
enemy_pressure += max(0, 3 - enemy_liberties) * 18.0
if len(mine) > len(enemy):
enemy_pressure += max(0, 8 - enemy_space) * 8.0
return space * 1.9 + liberties * 32.0 + enemy_pressure - len(enemies) * 4.0
def _occupied(self, mine: Body, enemies: EnemyBodies) -> int:
occupied = self._body_bits(mine)
for enemy in enemies:
occupied |= self._body_bits(enemy)
return occupied
def _body_bits(self, body: Body) -> int:
cached = self.body_bits_cache.get(body)
if cached is not None:
return cached
bits = 0
for cell in body:
bits |= 1 << cell
if len(self.body_bits_cache) < 16_384:
self.body_bits_cache[body] = bits
return bits
def _advance_body(self, body: Body, target: int, ate: bool) -> Body:
if self.is_constrictor or ate:
return (target,) + body
return (target,) + body[:-1]
@staticmethod
def _iter_bits(bits: int):
while bits:
bit = bits & -bits
yield bit.bit_length() - 1
bits ^= bit
def _out_of_time(self) -> bool:
expired = self.deadline is not None and perf_counter() >= self.deadline
if expired:
self.deadline_exits += 1
return expired