Compare commits

...

5 Commits

Author SHA1 Message Date
daniel156161 9b99b526e4 perf(snake): cache survival rollout state
Build and Push Docker Container / build-and-push (push) Successful in 6m1s
- Cache occupancy bitboards for repeated multiplayer rollout positions.
- Memoize position evaluations to avoid duplicate flood-fill calculations.
- Reuse shared values while ranking simultaneous enemy responses.
- Include evaluation hits in Prism rollout telemetry.
- Document the optimization and bump Prism to version 1.4.0.
2026-08-01 20:39:56 +02:00
daniel156161 3a9af3f54d feat(snake): modularize engine and add tournament tools
- 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.
2026-08-01 20:25:07 +02:00
daniel156161 cb6c8d4dc8 feat(snake): add adaptive adversarial search
- Share duel search contexts and transpositions across candidate moves.
- Add aspiration windows, principal variation ordering, and body caches.
- Model simultaneous multiplayer responses with a compact beam rollout.
- Adapt search depth and response breadth to the remaining deadline.
- Add a deterministic arena benchmark with optional JSON reporting.
- Expose search metrics, document benchmarking, and bump Prism to 1.2.0.
2026-08-01 19:26:19 +02:00
daniel156161 6643eb35af fix: resolve duel roots and recover legacy snake data
- Resolve selected moves and enemy replies on the same simulated turn.
- Add an Apex candidate hook and bump the Prism snake to version 1.1.0.
- Rebuild benchmark states from normalized turn data when snapshots are empty.
- Synthesize missing game snake identities during legacy database migration.
- Add regression coverage for duel timing and partial legacy schemas.
2026-08-01 19:11:32 +02:00
daniel156161 c646392b84 fix: preserve gameplay data and correct duel evaluation
- Preserve snake customizations across database migrations and merges.
- Lazily load optional storage backends for SQLite maintenance scripts.
- Match Apex territory and nearest-food tie-breaking semantics.
- Resolve duel occupancy after simultaneous movement and food growth.
- Recompute simulated head-to-head danger after body growth.
- Add regression coverage and declare the aiofiles dependency.
2026-08-01 18:16:04 +02:00
49 changed files with 2787 additions and 1022 deletions
+43 -14
View File
@@ -1,17 +1,14 @@
# Battlesnake Python Starter Project
An official Battlesnake template written in Python. Get started at [play.battlesnake.com](https://play.battlesnake.com).
![Battlesnake Logo](https://media.battlesnake.com/social/StarterSnakeGitHubRepos_Python.png)
This project is a great starting point for anyone wanting to program their first Battlesnake in Python. It can be run locally or easily deployed to a cloud provider of your choosing. See the [Battlesnake API Docs](https://docs.battlesnake.com/api) for more detail.
This project is a great starting point for anyone wanting to program their first Battlesnake in Python. It can be run locally or easily deployed to a cloud provider of your choosing. See the [Battlesnake API Docs](https://docs.battlesnake.com/api) for more detail.
## Technologies Used
This project uses [Python 3](https://www.python.org/) and [Flask](https://flask.palletsprojects.com/). It also comes with an optional [Dockerfile](https://docs.docker.com/engine/reference/builder/) to help with deployment.
## Run Your Battlesnake
Install dependencies using pip
```sh
@@ -19,7 +16,6 @@ pip install -r requirements.txt
```
Start your Battlesnake
```sh
python main.py
```
@@ -39,47 +35,56 @@ Open [localhost:8000](http://localhost:8000) in your browser and you should see
```
## Play a Game Locally
Install the [Battlesnake CLI](https://github.com/BattlesnakeOfficial/rules/tree/main/cli)
* You can [download compiled binaries here](https://github.com/BattlesnakeOfficial/rules/releases)
* or [install as a go package](https://github.com/BattlesnakeOfficial/rules/tree/main/cli#installation) (requires Go 1.18 or higher)
Command to run a local game
```sh
battlesnake play -W 11 -H 11 --name 'Python Starter Project' --url http://localhost:8000 -g solo --browser
```
## Next Steps
Continue with the [Battlesnake Quickstart Guide](https://docs.battlesnake.com/quickstart) to customize and improve your Battlesnake's behavior.
## Included Competitive Snake
This repo now includes `snakes/BestBattleSnake.py`, a stronger default snake that combines:
This repo retains `snakes/legacy/BestBattleSnake.py`, a stronger historical snake that combines:
- collision and head-to-head risk checks
- flood-fill space evaluation to avoid traps
- food routing that gets more aggressive as health drops
- tail access checks for better long-term survival
Run it explicitly with:
```sh
SNAKE=BestBattleSnake python main.py
```
Optional duel tuning (when only 2 snakes are alive):
```sh
BATTLE_SNAKE_DUEL_STYLE=balanced python main.py
```
Allowed values: `safe`, `balanced`, `aggressive`.
## Snake package layout
The snake code is split by responsibility:
- `snakes/strategies/` — actively maintained Apex and Prism entry points
- `snakes/engine/` — reusable bitboards, spatial mixins, duel search, and survival search
- `snakes/core/` — shared base classes
- `snakes/legacy/` — historical snakes retained for compatibility and benchmarks
Snake selection still uses the existing registry names, so deployment values such
as `SNAKE=PrismBattleSnake_GPT_5_6_Sol` remain unchanged.
## PrismBattleSnake_GPT_5_6_Sol
`PrismBattleSnake_GPT_5_6_Sol` is a separate snake that keeps Apex's strategy while
accelerating hot spatial operations with a Python-integer bitboard engine. Its
filename, class, and registry key include the model name, while its public
Battlesnake API name remains `PrismBattleSnake`.
accelerating hot spatial operations with a Python-integer bitboard engine. It
also shares duel transpositions across candidate moves, uses principal-variation
ordering, aspiration windows, path-aware food races, and a deeper tactical
horizon, and runs a compact adversarial multiplayer rollout with simultaneous
enemy responses and cached occupancy/evaluation states. Its filename, class, and registry
key include the model name, while its public Battlesnake API name remains
`PrismBattleSnake`.
Run it with:
```sh
@@ -96,6 +101,30 @@ python scripts/benchmark_snakes_from_db.py \
The benchmark opens SQLite read-only and reports mean, median, p95, and maximum
move latency. Increase `--samples` for a broader but slower comparison.
Run the deterministic CI-friendly arena benchmark without a gameplay database:
```sh
just bench-snake-arena positions=100
```
It rotates through duel, hazard, multiplayer, constrictor, and cramped-endgame
positions. It reports latency, completed duel/rollout depth, searched nodes,
cache hits, deadline exits, and move disagreements between Apex and Prism. Use
`--scenario hazard` (repeatable) when invoking the Python script to isolate a
scenario. Add `output=data/arena-report.json` to save a machine-readable report.
For representative strategy evaluation, provide recorded positions to
`scripts/benchmark_snake_arena.py --database /path/to/gameplay.sqlite3`.
Run paired seeded games through the official local Battlesnake rules engine:
```sh
just bench-snake-tournament games=20 gametype=standard map=standard
```
Each seed is played twice with Apex and Prism swapping initial engine slots. The
report includes wins, draws, win rates, and average game length. Save all
per-game results with `output=data/tournament-report.json`. The tournament starts
both snake servers with gameplay persistence disabled, adds the engine identity
header required by the API, and shuts them down when finished.
## Compact gameplay database
New gameplay turns use normalized storage: the turn row stores food, hazards,
move, and thinking data once; snake identity is stored once per game in
+16
View File
@@ -62,6 +62,22 @@ bench-best-snake iterations="1000":
PYTHONPATH="{{justfile_directory()}}" python "{{justfile_directory()}}/tests/bench_best_battle_snake.py" --iterations "{{iterations}}"
bench-snake-arena positions="100" output="":
#!/usr/bin/env bash
set -euo pipefail
args=(--positions "{{positions}}")
if [ -n "{{output}}" ]; then args+=(--json-output "{{output}}"); fi
PYTHONPATH="{{justfile_directory()}}" python "{{justfile_directory()}}/scripts/benchmark_snake_arena.py" "${args[@]}"
bench-snake-tournament games="20" gametype="standard" map="standard" output="":
#!/usr/bin/env bash
set -euo pipefail
args=(--games "{{games}}" --gametype "{{gametype}}" --map "{{map}}")
if [ -n "{{output}}" ]; then args+=(--json-output "{{output}}"); fi
PYTHONPATH="{{justfile_directory()}}" python "{{justfile_directory()}}/scripts/run_seeded_snake_tournament.py" "${args[@]}"
build-battlesnake-cli:
#!/usr/bin/env bash
set -euo pipefail
+1
View File
@@ -9,6 +9,7 @@ description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"aiofiles>=25.1.0",
"aiologger>=0.7.0",
"dotenv>=0.9.9",
"httpx>=0.28.0",
+142
View File
@@ -0,0 +1,142 @@
#!/usr/bin/env python3
"""Compare snake decisions and latency on deterministic synthetic positions.
For outcome/win-rate tournaments use the local Battlesnake CLI. This harness is
fast enough for CI and detects move disagreements, crashes, and latency changes.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from statistics import mean, median
from time import perf_counter
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from scripts.benchmark_snakes_from_db import load_states, percentile
from scripts.snake_arena_scenarios import SCENARIOS, synthetic_states
from server.GameBoard import GameBoard
from snakes import SnakeBuilder
def evaluate(name: str, states: list[tuple[dict, dict]]) -> tuple[list[str], dict]:
moves: list[str] = []
durations: list[float] = []
duel_depths: list[int] = []
rollout_depths: list[int] = []
duel_nodes = 0
rollout_nodes = 0
cache_hits = 0
deadline_exits = 0
scenario_durations: dict[str, list[float]] = {}
for index, (board_data, metadata) in enumerate(states):
snake = SnakeBuilder.build(name)
game_id = f"arena-{name}-{index}-{metadata['game_id']}"
board = GameBoard(
game_id=game_id, width=board_data["width"], height=board_data["height"],
ruleset=metadata["ruleset"], source=metadata["source"],
map=metadata["map"], snake_class=snake,
)
board.read_game_data({
"game": {
"id": game_id, "ruleset": metadata["ruleset"],
"source": metadata["source"], "map": metadata["map"], "timeout": 500,
},
"turn": metadata["turn"], "board": board_data, "you": metadata["you"],
})
started = perf_counter()
moves.append(snake.choose_move(board))
duration = (perf_counter() - started) * 1000.0
durations.append(duration)
scenario = metadata.get("scenario", "recorded")
scenario_durations.setdefault(scenario, []).append(duration)
history = snake.get_history() if hasattr(snake, "get_history") else []
if history:
thinking = history[-1]
duel_depths.append(int(thinking.get("prism_duel_depth", thinking.get("minimax_depth_reached", 0))))
rollout_depths.append(int(thinking.get("prism_rollout_depth", 0)))
duel_nodes += int(thinking.get("prism_duel_nodes", 0))
rollout_nodes += int(thinking.get("prism_rollout_nodes", 0))
cache_hits += int(thinking.get("prism_duel_cache_hits", 0))
cache_hits += int(thinking.get("prism_rollout_cache_hits", 0))
deadline_exits += int(thinking.get("prism_duel_deadline_exits", 0))
deadline_exits += int(thinking.get("prism_rollout_deadline_exits", 0))
return moves, {
"snake": name, "positions": len(states),
"mean_ms": mean(durations), "median_ms": median(durations),
"p95_ms": percentile(durations, 0.95), "max_ms": max(durations),
"mean_duel_depth": mean(duel_depths) if duel_depths else 0.0,
"mean_rollout_depth": mean(rollout_depths) if rollout_depths else 0.0,
"duel_nodes": duel_nodes, "rollout_nodes": rollout_nodes,
"cache_hits": cache_hits, "deadline_exits": deadline_exits,
"scenario_mean_ms": {
scenario: mean(values) for scenario, values in scenario_durations.items()
},
}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--snake", action="append", default=[])
parser.add_argument("--database")
parser.add_argument("--positions", type=int, default=100)
parser.add_argument("--stride", type=int, default=997)
parser.add_argument("--scenario", action="append", choices=sorted(SCENARIOS))
parser.add_argument("--json-output")
args = parser.parse_args()
states = (
load_states(args.database, max(1, args.positions), max(1, args.stride))
if args.database else synthetic_states(max(1, args.positions), args.scenario)
)
if not states:
raise SystemExit("No benchmark positions found")
names = args.snake or ["ApexBattleSnake", "PrismBattleSnake_GPT_5_6_Sol"]
move_sets: dict[str, list[str]] = {}
reports: list[dict] = []
for name in names:
moves, report = evaluate(name, states)
move_sets[name] = moves
reports.append(report)
print(
f"{name}: mean={report['mean_ms']:.3f} ms, "
f"p95={report['p95_ms']:.3f} ms, max={report['max_ms']:.3f} ms, "
f"duel-depth={report['mean_duel_depth']:.2f}, "
f"rollout-depth={report['mean_rollout_depth']:.2f}, "
f"nodes={report['duel_nodes'] + report['rollout_nodes']}, "
f"cache-hits={report['cache_hits']}, deadline-exits={report['deadline_exits']}"
)
baseline = names[0]
disagreements = {
name: sum(a != b for a, b in zip(move_sets[baseline], move_sets[name]))
for name in names[1:]
}
if disagreements:
print(f"Move disagreements versus {baseline}: {disagreements}")
scenario_disagreements = {}
for name in names[1:]:
counts: dict[str, int] = {}
for index, (baseline_move, candidate_move) in enumerate(
zip(move_sets[baseline], move_sets[name])
):
if baseline_move != candidate_move:
scenario = states[index][1].get("scenario", "recorded")
counts[scenario] = counts.get(scenario, 0) + 1
scenario_disagreements[name] = counts
if any(scenario_disagreements.values()):
print(f"Disagreements by scenario: {scenario_disagreements}")
payload = {
"reports": reports,
"baseline": baseline,
"disagreements": disagreements,
"scenario_disagreements": scenario_disagreements,
}
if args.json_output:
Path(args.json_output).write_text(json.dumps(payload, indent=2) + "\n")
if __name__ == "__main__":
main()
+47 -11
View File
@@ -31,7 +31,8 @@ def load_states(db_path: str, samples: int, stride: int) -> list[tuple[dict, dic
states: list[tuple[dict, dict]] = []
next_id = max(1, max_id - (samples - 1) * stride)
query = """
SELECT t.board_state_json, t.you_json, g.your_snake_id,
SELECT t.id, t.board_state_json, t.you_json, t.food_json, t.hazards_json,
g.your_snake_id, g.your_snake_name, g.width, g.height,
g.game_id, g.source, g.map_name,
g.ruleset_name, g.ruleset_version, t.turn
FROM turns AS t
@@ -40,30 +41,65 @@ def load_states(db_path: str, samples: int, stride: int) -> list[tuple[dict, dic
ORDER BY t.id
LIMIT 1
"""
snake_query = """
SELECT st.snake_id, COALESCE(gs.snake_name, st.snake_name),
st.health, st.length, st.head_x, st.head_y, st.body_json,
COALESCE(gs.customizations_json, '{}')
FROM snake_turns AS st
LEFT JOIN game_snakes AS gs
ON gs.game_id = st.game_id AND gs.snake_id = st.snake_id
WHERE st.game_id = ? AND st.turn = ?
ORDER BY st.id
"""
while len(states) < samples and next_id <= max_id:
row = connection.execute(query, (next_id,)).fetchone()
if row is None:
break
board = json.loads(row[0])
you = json.loads(row[1])
board = json.loads(row[1])
you = json.loads(row[2])
if not board.get("snakes"):
snakes = []
for snake_row in connection.execute(snake_query, (row[9], row[14])):
snake_id = snake_row[0]
snake_name = snake_row[1] or (row[6] if snake_id == row[5] else snake_id)
body = json.loads(snake_row[6])
snakes.append({
"id": snake_id,
"name": snake_name,
"health": snake_row[2],
"length": snake_row[3],
"head": {"x": snake_row[4], "y": snake_row[5]},
"body": body,
"customizations": json.loads(snake_row[7]),
})
board = {
"width": row[7],
"height": row[8],
"food": json.loads(row[3]),
"hazards": json.loads(row[4]),
"snakes": snakes,
}
if not you:
you = next(
(snake for snake in board.get("snakes", []) if snake.get("id") == row[2]),
(snake for snake in board.get("snakes", []) if snake.get("id") == row[5]),
{},
)
if not you or not board.get("snakes"):
next_id = int(row[0]) + stride
continue
metadata = {
"game_id": row[3],
"source": row[4] or "custom",
"map": row[5] or "standard",
"game_id": row[9],
"source": row[10] or "custom",
"map": row[11] or "standard",
"ruleset": {
"name": row[6] or "standard",
"version": row[7] or "v1.0.0",
"name": row[12] or "standard",
"version": row[13] or "v1.0.0",
"settings": {},
},
"turn": int(row[8]),
"turn": int(row[14]),
}
states.append((board, {"you": you, **metadata}))
next_id += stride
next_id = int(row[0]) + stride
connection.close()
return states
+15 -4
View File
@@ -117,15 +117,26 @@ def copy_game_snakes(source:sqlite3.Connection, destination:sqlite3.Connection,
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='game_snakes'"
).fetchone()
if has_table:
cursor = source.execute(
"SELECT game_id, snake_id, snake_name, is_you FROM game_snakes ORDER BY game_id, snake_id"
columns = object_columns(source, "game_snakes")
customizations = (
"COALESCE(customizations_json, '{}') AS customizations_json"
if "customizations_json" in columns else "'{}' AS customizations_json"
)
cursor = source.execute(f"""
SELECT game_id, snake_id, snake_name, is_you, {customizations}
FROM game_snakes ORDER BY game_id, snake_id
""")
else:
cursor = source.execute("""
SELECT game_id, snake_id, MAX(snake_name), MAX(is_you)
SELECT game_id, snake_id, MAX(snake_name), MAX(is_you),
'{}' AS customizations_json
FROM snake_turns GROUP BY game_id, snake_id ORDER BY game_id, snake_id
""")
sql = "INSERT INTO game_snakes (game_id,snake_id,snake_name,is_you) VALUES (?,?,?,?)"
sql = """
INSERT INTO game_snakes (
game_id,snake_id,snake_name,is_you,customizations_json
) VALUES (?,?,?,?,?)
"""
count = 0
while rows := cursor.fetchmany(batch_size):
values = [tuple(row) for row in rows if row[0] in allowed]
+30 -15
View File
@@ -143,27 +143,42 @@ def copy_game_snakes(source:sqlite3.Connection, destination:sqlite3.Connection,
has_game_snakes = source.execute("""
SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'game_snakes'
""").fetchone() is not None
if has_game_snakes:
cursor = source.execute("""
SELECT game_id, snake_id, snake_name, is_you
FROM game_snakes ORDER BY game_id, snake_id
""")
else:
cursor = source.execute("""
SELECT game_id, snake_id, MAX(snake_name), MAX(is_you)
FROM snake_turns
GROUP BY game_id, snake_id
ORDER BY game_id, snake_id
""")
sql = """
INSERT INTO game_snakes (game_id, snake_id, snake_name, is_you)
VALUES (?, ?, ?, ?)
INSERT OR IGNORE INTO game_snakes (
game_id, snake_id, snake_name, is_you, customizations_json
) VALUES (?, ?, ?, ?, ?)
"""
count = 0
if has_game_snakes:
columns = object_columns(source, "game_snakes")
customizations = (
"COALESCE(customizations_json, '{}') AS customizations_json"
if "customizations_json" in columns else "'{}' AS customizations_json"
)
cursor = source.execute(f"""
SELECT game_id, snake_id, snake_name, is_you, {customizations}
FROM game_snakes ORDER BY game_id, snake_id
""")
while rows := cursor.fetchmany(batch_size):
retained_rows = [tuple(row) for row in rows if row[0] in retained_ids]
before = destination.total_changes
destination.executemany(sql, retained_rows)
count += destination.total_changes - before
# Older databases can contain an empty or only partially populated
# game_snakes table. Always synthesize missing identities from snake_turns.
cursor = source.execute("""
SELECT game_id, snake_id, MAX(snake_name), MAX(is_you),
'{}' AS customizations_json
FROM snake_turns
GROUP BY game_id, snake_id
ORDER BY game_id, snake_id
""")
while rows := cursor.fetchmany(batch_size):
retained_rows = [tuple(row) for row in rows if row[0] in retained_ids]
before = destination.total_changes
destination.executemany(sql, retained_rows)
count += len(retained_rows)
count += destination.total_changes - before
return count
def decode_json(value:str|None, fallback):
+238
View File
@@ -0,0 +1,238 @@
#!/usr/bin/env python3
"""Run paired seeded games through the official local Battlesnake rules engine."""
from __future__ import annotations
import argparse
import json
import os
import re
import signal
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from collections import Counter
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from pathlib import Path
from statistics import mean
ROOT = Path(__file__).resolve().parents[1]
ENGINE_USER_AGENT = "BattlesnakeEngine/local-tournament"
RESULT_RE = re.compile(
r"Game completed after (\d+) turns\.(?: (.+?) was the winner\.| It was a draw\.)"
)
def wait_for_server(url: str, process: subprocess.Popen, timeout: float = 15.0) -> None:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if process.poll() is not None:
raise RuntimeError(f"Snake server exited with status {process.returncode}")
try:
with urllib.request.urlopen(url, timeout=0.5) as response:
if response.status == 200:
return
except OSError:
time.sleep(0.1)
raise TimeoutError(f"Snake server did not become ready at {url}")
class _EngineHeaderProxy(BaseHTTPRequestHandler):
target: str
def do_GET(self) -> None:
self._forward()
def do_POST(self) -> None:
self._forward()
def _forward(self) -> None:
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length) if length else None
request = urllib.request.Request(
f"{self.target}{self.path}", data=body, method=self.command,
headers={
"Content-Type": self.headers.get("Content-Type", "application/json"),
"User-Agent": ENGINE_USER_AGENT,
},
)
try:
with urllib.request.urlopen(request, timeout=2.0) as response:
payload = response.read()
self.send_response(response.status)
self.send_header("Content-Type", response.headers.get("Content-Type", "application/json"))
except urllib.error.HTTPError as error:
payload = error.read()
self.send_response(error.code)
self.send_header("Content-Type", error.headers.get("Content-Type", "text/plain"))
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def log_message(self, format: str, *args) -> None:
pass
def start_proxy(port: int, target_port: int) -> tuple[ThreadingHTTPServer, Thread]:
handler = type(
f"EngineHeaderProxy{port}",
(_EngineHeaderProxy,),
{"target": f"http://127.0.0.1:{target_port}"},
)
server = ThreadingHTTPServer(("127.0.0.1", port), handler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
return server, thread
def start_server(snake: str, port: int) -> subprocess.Popen:
env = os.environ.copy()
env.update({
"HOST": "127.0.0.1",
"PORT": str(port),
"SNAKE": snake,
"DEBUG": "false",
"DEBUG_SERVER": "false",
"STORE_GAME_HISTORY": "false",
"GAMEPLAY_DB_ENABLED": "false",
"METRICS_CLEAR_WORKERS_ON_STARTUP": "false",
})
process = subprocess.Popen(
[sys.executable, str(ROOT / "main.py")],
cwd=ROOT,
env=env,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
wait_for_server(f"http://127.0.0.1:{port}", process)
return process
def stop_server(process: subprocess.Popen) -> None:
if process.poll() is not None:
return
os.killpg(process.pid, signal.SIGTERM)
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
os.killpg(process.pid, signal.SIGKILL)
process.wait(timeout=5)
def run_game(
cli: Path,
seed: int,
game_type: str,
map_name: str,
players: list[tuple[str, str]],
width: int,
height: int,
timeout_ms: int,
) -> dict:
with tempfile.NamedTemporaryFile(prefix="snake-arena-", suffix=".jsonl") as output:
command = [
str(cli), "play", "-W", str(width), "-H", str(height),
"-g", game_type, "--map", map_name, "--seed", str(seed),
"--timeout", str(timeout_ms), "--output", output.name,
]
for name, url in players:
command.extend(("--name", name, "--url", url))
completed = subprocess.run(
command, cwd=ROOT, text=True, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, timeout=180, check=False,
)
if completed.returncode != 0:
raise RuntimeError(
f"Rules engine failed for seed {seed}:\n{completed.stdout[-2000:]}"
)
output.seek(0)
lines = [json.loads(line) for line in output if line.strip()]
terminal = lines[-1] if lines else {}
match = RESULT_RE.search(completed.stdout)
turns = int(match.group(1)) if match else max(0, len(lines) - 2)
winner = terminal.get("winnerName") or None
draw = bool(terminal.get("isDraw", winner is None))
return {"seed": seed, "winner": winner, "draw": draw, "turns": turns}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--games", type=int, default=20, help="Number of unique seeds")
parser.add_argument("--seed-start", type=int, default=1)
parser.add_argument("--gametype", default="standard")
parser.add_argument("--map", default="standard")
parser.add_argument("--width", type=int, default=11)
parser.add_argument("--height", type=int, default=11)
parser.add_argument("--timeout", type=int, default=500)
parser.add_argument("--base-port", type=int, default=9301)
parser.add_argument("--cli", default=str(ROOT / ".testing/tools/battlesnake-cli/battlesnake"))
parser.add_argument("--json-output")
args = parser.parse_args()
cli = Path(args.cli)
if not cli.is_file():
raise SystemExit(f"Battlesnake CLI not found: {cli}. Run: just build-battlesnake-cli")
apex_port, prism_port = args.base_port, args.base_port + 1
apex_proxy_port, prism_proxy_port = args.base_port + 2, args.base_port + 3
servers: list[subprocess.Popen] = []
proxies: list[tuple[ThreadingHTTPServer, Thread]] = []
results: list[dict] = []
started = time.perf_counter()
try:
servers = [
start_server("ApexBattleSnake", apex_port),
start_server("PrismBattleSnake_GPT_5_6_Sol", prism_port),
]
proxies = [
start_proxy(apex_proxy_port, apex_port),
start_proxy(prism_proxy_port, prism_port),
]
urls = {
"Apex": f"http://127.0.0.1:{apex_proxy_port}",
"Prism": f"http://127.0.0.1:{prism_proxy_port}",
}
for offset in range(max(1, args.games)):
seed = args.seed_start + offset
# Swap engine slots for every seed. This controls for deterministic map
# spawn positions and gives each strategy both initial placements.
for order in (("Apex", "Prism"), ("Prism", "Apex")):
result = run_game(
cli=cli, seed=seed, game_type=args.gametype, map_name=args.map,
players=[(name, urls[name]) for name in order],
width=args.width, height=args.height, timeout_ms=args.timeout,
)
result["order"] = list(order)
results.append(result)
print(
f"seed={seed} order={'/'.join(order)} winner={result['winner'] or 'draw'} "
f"turns={result['turns']}"
)
finally:
for proxy, thread in reversed(proxies):
proxy.shutdown()
proxy.server_close()
thread.join(timeout=2)
for server in reversed(servers):
stop_server(server)
wins = Counter(result["winner"] or "draw" for result in results)
summary = {
"games": len(results),
"unique_seeds": max(1, args.games),
"gametype": args.gametype,
"map": args.map,
"wins": dict(wins),
"win_rates": {
key: value / len(results) for key, value in wins.items()
},
"mean_turns": mean(result["turns"] for result in results),
"elapsed_seconds": time.perf_counter() - started,
"results": results,
}
print(json.dumps({key: value for key, value in summary.items() if key != "results"}, indent=2))
if args.json_output:
Path(args.json_output).write_text(json.dumps(summary, indent=2) + "\n")
if __name__ == "__main__":
main()
+108
View File
@@ -0,0 +1,108 @@
"""Deterministic scenario corpus for the local snake arena."""
from __future__ import annotations
from copy import deepcopy
from tests.bench_best_battle_snake import build_game_state
def _state(payload: dict, scenario: str, index: int) -> tuple[dict, dict]:
payload["game"]["id"] = f"arena-{scenario}-{index}"
return payload["board"], {
"game_id": payload["game"]["id"],
"source": "custom",
"map": payload["game"].get("map", "standard"),
"ruleset": payload["game"]["ruleset"],
"turn": payload["turn"],
"you": payload["you"],
"scenario": scenario,
}
def _standard_duel(index: int) -> dict:
payload = build_game_state()
payload["turn"] = 20 + index
payload["board"]["food"] = [
{"x": 1 + index % 3, "y": 9},
{"x": 9, "y": 1 + (index // 3) % 3},
]
return payload
def _hazard_duel(index: int) -> dict:
payload = _standard_duel(index)
payload["you"]["health"] = 38 + index % 12
payload["board"]["snakes"][0]["health"] = payload["you"]["health"]
hazard_x = 5 + index % 2
payload["board"]["hazards"] = [
{"x": hazard_x, "y": y} for y in range(1, 10) if y != 5
]
return payload
def _multiplayer(index: int) -> dict:
payload = _standard_duel(index)
third = {
"id": "enemy-2",
"name": "enemy-2",
"health": 65,
"length": 5,
"head": {"x": 2, "y": 8},
"body": [
{"x": 2, "y": 8}, {"x": 2, "y": 9}, {"x": 2, "y": 10},
{"x": 1, "y": 10}, {"x": 0, "y": 10},
],
}
payload["board"]["snakes"].append(third)
return payload
def _constrictor(index: int) -> dict:
payload = _multiplayer(index)
payload["game"]["ruleset"] = deepcopy(payload["game"]["ruleset"])
payload["game"]["ruleset"]["name"] = "constrictor"
payload["board"]["food"] = []
return payload
def _cramped_duel(index: int) -> dict:
payload = _standard_duel(index)
payload["board"]["width"] = 7
payload["board"]["height"] = 7
mine = {
"id": "me", "name": "me", "health": 72, "length": 7,
"head": {"x": 2, "y": 3},
"body": [
{"x": 2, "y": 3}, {"x": 2, "y": 2}, {"x": 2, "y": 1},
{"x": 1, "y": 1}, {"x": 1, "y": 2}, {"x": 1, "y": 3},
{"x": 1, "y": 4},
],
}
enemy = {
"id": "enemy", "name": "enemy", "health": 72, "length": 7,
"head": {"x": 4, "y": 3},
"body": [
{"x": 4, "y": 3}, {"x": 4, "y": 2}, {"x": 4, "y": 1},
{"x": 5, "y": 1}, {"x": 5, "y": 2}, {"x": 5, "y": 3},
{"x": 5, "y": 4},
],
}
payload["you"] = mine
payload["board"]["snakes"] = [mine, enemy]
payload["board"]["food"] = [{"x": 3, "y": 5 + index % 2}]
payload["board"]["hazards"] = []
return payload
SCENARIOS = {
"duel": _standard_duel,
"hazard": _hazard_duel,
"multi": _multiplayer,
"constrictor": _constrictor,
"cramped": _cramped_duel,
}
def synthetic_states(count: int, scenarios: list[str] | None = None) -> list[tuple[dict, dict]]:
selected = scenarios or list(SCENARIOS)
unknown = set(selected) - set(SCENARIOS)
if unknown:
raise ValueError(f"Unknown arena scenarios: {', '.join(sorted(unknown))}")
states: list[tuple[dict, dict]] = []
for index in range(count):
scenario = selected[index % len(selected)]
states.append(_state(SCENARIOS[scenario](index), scenario, index))
return states
+1 -1
View File
@@ -1,4 +1,4 @@
from snakes.TemplateSnake import TemplateSnake
from snakes.core.template import TemplateSnake
from datetime import datetime
class GameBoard:
+32 -4
View File
@@ -1,12 +1,40 @@
from typing import TYPE_CHECKING, Any
from .GameplayDatabase import GameplayDatabase
from .backend import GameplayBackendBuilder
from .LocalStorage import LocalStorage
from .EdgeDB import EdgeDB
if TYPE_CHECKING:
from .EdgeDB import EdgeDB
from .LocalStorage import LocalStorage
__all__ = (
"EdgeDB",
"GameplayBackendBuilder",
"GameplayDatabase",
"LocalStorage",
"StorageLoader",
)
def __getattr__(name:str):
"""Load optional storage backends only when explicitly requested.
Database maintenance scripts import SQLite backend modules through this
package. Eagerly importing LocalStorage used to make those scripts require
unrelated web-storage dependencies such as aiofiles.
"""
if name == "LocalStorage":
from .LocalStorage import LocalStorage
return LocalStorage
if name == "EdgeDB":
from .EdgeDB import EdgeDB
return EdgeDB
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
class StorageLoader:
@classmethod
def build(self, selected_storage:str) -> LocalStorage|EdgeDB:
storage_module = __import__(f"server.database.{selected_storage}", fromlist=[selected_storage])
def build(cls, selected_storage:str) -> Any:
storage_module = __import__(
f"server.database.{selected_storage}", fromlist=[selected_storage],
)
storage_class = getattr(storage_module, selected_storage)
return storage_class
-555
View File
@@ -1,555 +0,0 @@
"""PrismBattleSnake_GPT_5_6_Sol v1.0.0
Built on ApexBattleSnake v1.0.0. All strategic logic is inherited.
Performance improvement: all spatial primitives (flood fill, territory,
articulation detection, distance maps, pathfinding) replaced by a
bitboard engine that uses integer arithmetic instead of Python sets/deques.
Key speedups:
S1: Bitboard flood fill replaces BFS deque+set with integer bit-expansion.
~60× faster per call, eliminates _neighbors() generator overhead.
S2: Bitboard territory dual-BFS expansion on ints replaces per-cell
distance-map comparison loop.
S3: Bitboard articulation partition sizes via bit-flood instead of
_bounded_bfs with sets.
S4: Bitboard distance map BFS via bit-expansion + bit-extract.
S5: Bitboard path distance early-exit BFS on ints.
S6: Bitboard nearest food BFS food search on ints.
S7: Per-turn BitBoard instance cached for board dimensions.
S8: Blocked-set bitboard conversion cached within a turn to avoid
redundant O(n) conversions for the same frozen set.
S9: Survival-tree uses bitboards natively enemy body/attack bits
precomputed once at tree root, no per-node set/dict rebuilds.
S10: _legal_moves override uses bitboard neighbour mask instead of
per-direction Python loop + _in_bounds calls.
S11: _future_survival_tree inlines legal-move check with bitboard ops.
S12: Duel minimax uses tuple bodies and bitboard move generation.
S13: Iterative deepening reuses a transposition table and move-order hints.
"""
from __future__ import annotations
from typing import Any
from time import perf_counter
from snakes.ApexBattleSnake import ApexBattleSnake
from snakes.bitboard import BitBoard
from snakes.bitboard_duel_search import BitboardDuelSearch
from server.GameBoard import GameBoard
# Direction offsets for coord-dict → tuple conversion
_DIR_DELTAS = ((0, 1), (0, -1), (-1, 0), (1, 0))
_DIR_NAMES = ("up", "down", "left", "right")
class PrismBattleSnake_GPT_5_6_Sol(ApexBattleSnake):
VERSION = "1.0.0"
def __init__(self) -> None:
super().__init__()
self.name = "PrismBattleSnake"
self.version = self.VERSION
# S7: cached BitBoard instance (reused while board dimensions stay the same)
self._bb: BitBoard | None = None
self._bb_w: int = 0
self._bb_h: int = 0
# S9: precomputed enemy state for survival tree (set per turn in choose_move)
self._enemy_body_bits: int = 0 # all enemy body cells as bitboard
self._enemy_tail_bits: int = 0 # enemy tails that will vacate
self._enemy_attack_danger: int = 0 # tiles where enemy len >= our len
self._enemy_attack_opportunity: int = 0 # tiles where enemy len < our len
# ── BitBoard accessor ────────────────────────────────────────────────────
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)
# ── choose_move override: precompute enemy bits ──────────────────────────
def choose_move(self, game_data: GameBoard) -> str:
bb = self._get_bb(game_data.get_width(), game_data.get_height())
# S9: precompute enemy body / tail / attack bitboards for survival tree
other_snakes = game_data.get_other_snakes()
my_snake = game_data.get_my_snake()
my_len = my_snake.get("length", len(my_snake["body"]))
food_set = {(f["x"], f["y"]) for f in game_data.get_food()}
game_type = game_data.get_type()
is_constrictor = game_type == "constrictor"
w = bb.width
enemy_body_bits = 0
enemy_tail_bits = 0
enemy_attack_danger = 0
enemy_attack_opportunity = 0
for snake in other_snakes:
for seg in snake["body"]:
enemy_body_bits |= 1 << (seg["y"] * w + seg["x"])
body = snake["body"]
# Check if tail will vacate
if not is_constrictor and len(body) >= 2:
tail_stacked = (body[-1]["x"] == body[-2]["x"] and body[-1]["y"] == body[-2]["y"])
if not tail_stacked:
can_grow = self._enemy_can_grow_this_turn(snake, food_set)
if not can_grow:
enemy_tail_bits |= 1 << (body[-1]["y"] * w + body[-1]["x"])
# Attack map: tiles enemy head can reach in 1 move
eh = snake["head"]
e_len = snake.get("length", len(body))
ehx, ehy = eh["x"], eh["y"]
for dx, dy in _DIR_DELTAS:
nx, ny = ehx + dx, ehy + dy
if 0 <= nx < w and 0 <= ny < bb.height:
bit = 1 << (ny * w + nx)
if e_len >= my_len:
enemy_attack_danger |= bit
else:
enemy_attack_opportunity |= bit
self._enemy_body_bits = enemy_body_bits
self._enemy_tail_bits = enemy_tail_bits
self._enemy_attack_danger = enemy_attack_danger
self._enemy_attack_opportunity = enemy_attack_opportunity
return super().choose_move(game_data)
# ── S1: Bitboard flood fill ──────────────────────────────────────────────
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
# ── S2: Bitboard territory ──────────────────────────────────────────────
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)
# ── S3: Bitboard articulation penalty ────────────────────────────────────
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
# ── S4: Bitboard distance map ───────────────────────────────────────────
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()}
# ── S5: Bitboard path distance ──────────────────────────────────────────
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,
)
# ── S6: Bitboard nearest food ───────────────────────────────────────────
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)
# ── Bitboard open-neighbour helpers ──────────────────────────────────────
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)
# ── S12/S13: compact bitboard duel search ───────────────────────────────
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:
return BitboardDuelSearch(
board=self._get_bb(width, height),
food=food_set,
hazards=hazard_set,
hazard_count=hazard_count,
hazard_damage=hazard_damage,
deadline=deadline,
)
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,
)
# ── S9: Optimised survival tree (bitboard-native) ────────────────────────
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) ────────────────────────
# Remove tiles where an enemy of >= our length could head-to-head.
# The danger bitboard was precomputed; filter out tiles blocked by
# current body (enemy can't step there either).
danger_here = self._enemy_attack_danger & ~blocked_bits
safe_nb = nb_free & ~danger_here
en_safe = safe_nb.bit_count()
if en_safe == 0:
return -4000.0
sc = reachable * 1.9 + liberties * 14.0 + liberties * 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
h = bb.height
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
# ── S10: Bitboard legal moves ────────────────────────────────────────────
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
# ── Enemy confinement (uses bitboard flood) ──────────────────────────────
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
+46 -25
View File
@@ -1,40 +1,61 @@
import importlib
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class SnakeRegistration:
module: str
version: str
SNAKE_REGISTRATIONS = {
"TemplateSnake": SnakeRegistration("snakes.core.template", "1.0.0"),
"ApexBattleSnake": SnakeRegistration("snakes.strategies.apex", "1.0.0"),
"PrismBattleSnake_GPT_5_6_Sol": SnakeRegistration(
"snakes.strategies.prism", "1.4.0"
),
"DummSnake": SnakeRegistration("snakes.legacy.DummSnake", "1.0.0"),
"LogicSnake": SnakeRegistration("snakes.legacy.LogicSnake", "1.1.0"),
"MasterSnake": SnakeRegistration("snakes.legacy.MasterSnake", "1.2.0"),
"BetterMasterSnake": SnakeRegistration("snakes.legacy.BetterMasterSnake", "1.3.0"),
"BestBattleSnake": SnakeRegistration("snakes.legacy.BestBattleSnake", "2.6.0"),
"TrainedBattleSnake": SnakeRegistration(
"snakes.legacy.TrainedBattleSnake", "0.1.0"
),
"UltimateBattleSnake": SnakeRegistration(
"snakes.legacy.UltimateBattleSnake", "4.5.0"
),
"SupremeBattleSnake_ClaudeOpus4_6": SnakeRegistration(
"snakes.legacy.SupremeBattleSnake_ClaudeOpus4_6",
"1.0.0",
),
}
# Backward-compatible public version map.
SNAKE_REGISTRY = {
"TemplateSnake": "1.0.0",
"DummSnake": "1.0.0",
"LogicSnake": "1.1.0",
"MasterSnake": "1.2.0",
"BetterMasterSnake": "1.3.0",
"BestBattleSnake": "2.6.0",
"TrainedBattleSnake": "0.1.0",
"UltimateBattleSnake": "4.5.0",
"ApexBattleSnake": "1.0.0",
"SupremeBattleSnake_ClaudeOpus4_6": "1.0.0",
"PrismBattleSnake_GPT_5_6_Sol": "1.0.0",
name: registration.version for name, registration in SNAKE_REGISTRATIONS.items()
}
DEFAULT_SNAKE_CONFIG = {
'apiversion': '1',
'author': '',
'color': '#888888',
'head': 'default',
'tail': 'default',
"apiversion": "1",
"author": "",
"color": "#888888",
"head": "default",
"tail": "default",
}
def build_snake(selected_snake:str):
if selected_snake not in SNAKE_REGISTRY:
def build_snake(selected_snake: str):
registration = SNAKE_REGISTRATIONS.get(selected_snake)
if registration is None:
raise ValueError(f"Unknown snake: {selected_snake}")
snake_module = importlib.import_module(f"snakes.{selected_snake}")
snake_module = importlib.import_module(registration.module)
snake_class = getattr(snake_module, selected_snake)
return snake_class()
def get_snake_version(selected_snake:str) -> str|None:
version = SNAKE_REGISTRY.get(selected_snake)
if version is None:
return None
return str(version)
def get_snake_version(selected_snake: str) -> str | None:
registration = SNAKE_REGISTRATIONS.get(selected_snake)
return registration.version if registration is not None else None
class SnakeBuilder:
@classmethod
@@ -42,5 +63,5 @@ class SnakeBuilder:
return build_snake(selected_snake)
@classmethod
def get_version(self, selected_snake:str) -> str|None:
def get_version(self, selected_snake: str) -> str | None:
return get_snake_version(selected_snake)
-291
View File
@@ -1,291 +0,0 @@
"""Deadline-aware simultaneous duel search using compact tuple bodies and bitboards."""
from __future__ import annotations
from dataclasses import dataclass
from time import perf_counter
from typing import Iterable
from snakes.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]] = {}
self.killer_moves: dict[int, int] = {}
self.history: dict[int, int] = {}
self.nodes = 0
self.cache_hits = 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
value, completed = self._search(state, depth, -float("inf"), float("inf"))
if not completed:
break
result = value
completed_depth = 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, _ = self._search(state, depth, -float("inf"), float("inf"))
return value
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 = 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._legal_targets(state.my_body, state.enemy_body)
enemy_moves = self._legal_targets(state.enemy_body, state.my_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)
enemy_moves = self._ordered_moves(enemy_moves, state, depth, False)
best = -float("inf")
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
best = max(best, worst)
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)
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 _legal_targets(self, body: Body, other_body: Body) -> list[int]:
occupied = self._body_bits(body) | self._body_bits(other_body)
if not self._tail_stacked(body):
occupied &= ~(1 << body[-1])
if not self._tail_stacked(other_body):
occupied &= ~(1 << other_body[-1])
legal = self.board.neighbors_of(body[0]) & ~occupied & self.board.board_mask
return list(self._iter_bits(legal))
def _ordered_moves(self, moves: list[int], state: DuelState, depth: int, mine: bool) -> 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))
killer_bonus = 10_000.0 if target == killer else 0.0
return 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:
my_blocked = self._body_bits(state.my_body[1:]) | self._body_bits(state.enemy_body[1:])
my_space = self.board.flood_count(state.my_body[0], my_blocked)
enemy_space = self.board.flood_count(state.enemy_body[0], my_blocked)
my_liberties = self.board.open_neighbor_count(state.my_body[0], my_blocked)
enemy_liberties = self.board.open_neighbor_count(state.enemy_body[0], my_blocked)
length_score = (len(state.my_body) - len(state.enemy_body)) * 18.0
health_score = (state.my_health - state.enemy_health) * 0.15
return (my_space - enemy_space) * 2.0 + (my_liberties - enemy_liberties) * 12.0 + length_score + health_score
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]
@staticmethod
def _body_bits(body: Body) -> int:
bits = 0
for cell in body:
bits |= 1 << cell
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
return perf_counter() + reserve_ms / 1000.0 >= self.deadline
+5
View File
@@ -0,0 +1,5 @@
"""Shared snake base classes."""
from snakes.core.template import TemplateSnake
__all__ = ("TemplateSnake",)
+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",
)
@@ -127,8 +127,9 @@ class BitBoard:
) -> int:
"""Simultaneous BFS from *my_idx* and all enemies.
Returns (my_cells enemy_cells). Cells equidistant from both sides are
counted for neither (contested).
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
@@ -139,48 +140,44 @@ class BitBoard:
nlc = self._not_leftcol
my_front = 1 << my_idx
my_terr = my_front
my_seen = my_front
en_front = 0
for ei in enemy_indices:
en_front |= 1 << ei
en_terr = en_front
en_seen = en_front
remaining = free & ~my_terr & ~en_terr
# 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
while (my_front or en_front) and remaining:
# Expand both sides simultaneously (same BFS depth → ties go to neither)
my_exp = 0
if my_front:
my_exp = (
((my_front & nrc) << 1)
| ((my_front & nlc) >> 1)
| (my_front << w)
| (my_front >> w)
) & remaining
en_exp = 0
if en_front:
en_exp = (
((en_front & nrc) << 1)
| ((en_front & nlc) >> 1)
| (en_front << w)
| (en_front >> w)
) & remaining
# Contested cells (reached by both at the same depth) → neither claims
contested = my_exp & en_exp
my_exp &= ~contested
en_exp &= ~contested
my_terr |= my_exp
en_terr |= en_exp
remaining &= ~(my_exp | en_exp | contested)
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 my_terr.bit_count() - en_terr.bit_count()
return score
# ── Partition sizes (for articulation-point detection) ────────────────────
@@ -324,32 +321,44 @@ class BitBoard:
if start_bit & food_bits:
return 0, start_idx
frontier = start_bit
seen = frontier
# 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
nrc = self._not_rightcol
nlc = self._not_leftcol
size = self.size
while frontier:
dist += 1
expanded = (
((frontier & nrc) << 1)
| ((frontier & nlc) >> 1)
| (frontier << w)
| (frontier >> w)
) & free & ~seen
if not expanded:
break
hit = expanded & food_bits
if hit:
# Return the first (lowest-index) food cell found
first_bit = hit & (-hit)
return dist, first_bit.bit_length() - 1
seen |= expanded
frontier = expanded
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
+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,
)
+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
+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
+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
+295
View File
@@ -0,0 +1,295 @@
"""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]
EvaluationKey = tuple[Body, EnemyBodies]
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.evaluation_cache: dict[EvaluationKey, float] = {}
self.occupied_cache: dict[EvaluationKey, int] = {}
self.body_bits_cache: dict[Body, int] = {}
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(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)
ranked: list[tuple[float, int]] = []
for target in self._iter_bits(self.board.neighbors_of(mine[0])):
# 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))
occupied = self._occupied(mine, enemies)
mx, my = self.board.coord(my_target)
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)
score -= abs(tx - mx) + abs(ty - my)
score += self.board.open_neighbor_count(target, occupied) * 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:
key = (mine, enemies)
cached = self.evaluation_cache.get(key)
if cached is not None:
self.evaluation_cache_hits += 1
return cached
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
value = space * 1.9 + liberties * 32.0 + enemy_pressure - len(enemies) * 4.0
if len(self.evaluation_cache) < 32_768:
self.evaluation_cache[key] = value
return value
def _occupied(self, mine: Body, enemies: EnemyBodies) -> int:
key = (mine, enemies)
cached = self.occupied_cache.get(key)
if cached is not None:
return cached
occupied = self._body_bits(mine)
for enemy in enemies:
occupied |= self._body_bits(enemy)
if len(self.occupied_cache) < 32_768:
self.occupied_cache[key] = occupied
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
@@ -7,7 +7,7 @@ import os
from quart_common.web.env import env_int
from server.dataset.RLBootstrapDataset import RLBootstrapDataset
from snakes.TemplateSnake import TemplateSnake
from snakes.core.template import TemplateSnake
from server.GameBoard import GameBoard
class BestBattleSnake(TemplateSnake):
@@ -1,4 +1,4 @@
from snakes.TemplateSnake import TemplateSnake
from snakes.core.template import TemplateSnake
from server.GameBoard import GameBoard
from collections import deque
@@ -1,4 +1,4 @@
from snakes.TemplateSnake import TemplateSnake
from snakes.core.template import TemplateSnake
import random
@@ -1,4 +1,4 @@
from snakes.TemplateSnake import TemplateSnake
from snakes.core.template import TemplateSnake
import random
from scipy import spatial
@@ -1,4 +1,4 @@
from snakes.TemplateSnake import TemplateSnake
from snakes.core.template import TemplateSnake
class MasterSnake(TemplateSnake):
VERSION = "1.2.0"
@@ -29,8 +29,8 @@ from __future__ import annotations
from typing import Any
from time import perf_counter
from snakes.ApexBattleSnake import ApexBattleSnake
from snakes.bitboard import BitBoard
from snakes.strategies.apex import ApexBattleSnake
from snakes.engine.bitboard import BitBoard
from server.GameBoard import GameBoard
# Direction offsets for coord-dict → tuple conversion
@@ -3,7 +3,7 @@ from typing import Any
import random, json, os
from server.TrainBattleSnakeAI import MOVES, extract_feature_values
from snakes.TemplateSnake import TemplateSnake
from snakes.core.template import TemplateSnake
class TrainedBattleSnake(TemplateSnake):
VERSION = "0.1.0"
@@ -6,7 +6,7 @@ import heapq, os
from quart_common.web.env import env_int
from snakes.TemplateSnake import TemplateSnake
from snakes.core.template import TemplateSnake
from server.GameBoard import GameBoard
from server.dataset.RLBootstrapDataset import RLBootstrapDataset
+1
View File
@@ -0,0 +1 @@
"""Historical snake strategies retained for replay and comparison."""
+6
View File
@@ -0,0 +1,6 @@
"""Actively maintained competitive snake strategies."""
from snakes.strategies.apex import ApexBattleSnake
from snakes.strategies.prism import PrismBattleSnake_GPT_5_6_Sol
__all__ = ("ApexBattleSnake", "PrismBattleSnake_GPT_5_6_Sol")
@@ -7,7 +7,7 @@ import heapq, os
from quart_common.web.env import env_int
from server.dataset.RLBootstrapDataset import RLBootstrapDataset
from snakes.TemplateSnake import TemplateSnake
from snakes.core.template import TemplateSnake
from server.GameBoard import GameBoard
class ApexBattleSnake(TemplateSnake):
@@ -18,20 +18,20 @@ class ApexBattleSnake(TemplateSnake):
New improvements:
A1: Iterative deepening minimax tries depth 1,2,...,N within time budget; keeps deepest
fully-completed result instead of a fixed depth=2 call.
fully-completed result instead of a fixed depth=2 call.
A2: Hazard-aware starvation check Dijkstra with per-tile hazard cost replaces BFS food
distance when hazards are present and health < 55. Correctly models health depletion
through hazard corridors when choosing whether to seek food.
distance when hazards are present and health < 55. Correctly models health depletion
through hazard corridors when choosing whether to seek food.
A3: Phase-adaptive scoring weights board occupancy drives a game_phase scalar [0,1].
Territory weight scales up late-game; food bias scales down. Stored as self._game_phase.
Territory weight scales up late-game; food bias scales down. Stored as self._game_phase.
A4: Rich GameplayDatabase thinking data add_to_history records game_phase, food_count,
enemy lengths/healths, minimax_depth_reached, score_gap, safe_moves_count per turn.
enemy lengths/healths, minimax_depth_reached, score_gap, safe_moves_count per turn.
A5: Dynamic duel aggression auto-adjusts head_pressure/distance_safety multipliers based
on (my_len - enemy_len) delta on top of the configured duel style preset.
on (my_len - enemy_len) delta on top of the configured duel style preset.
A6: Constrictor endgame encirclement when enemy is sealed in a region <= our body length,
apply a strong encirclement bonus to close out the win efficiently.
apply a strong encirclement bonus to close out the win efficiently.
A7: Bounded BFS transposition cache caps per-turn cache at 4096 entries to prevent
memory growth in long games with many unique blocked-set combinations.
memory growth in long games with many unique blocked-set combinations.
"""
VERSION = "1.0.0"
@@ -468,15 +468,11 @@ class ApexBattleSnake(TemplateSnake):
if self._time_exceeded(deadline):
break
pos = safe_moves[m]
ate = (pos["x"], pos["y"]) in food_set
fb = self._future_body(my_body, pos, ate, False)
nmy_h = 100 if ate else my_health - 1
if (pos["x"], pos["y"]) in hazard_set and not ate:
nmy_h -= hazard_damage * hazard_count.get((pos["x"], pos["y"]), 1)
mm_val, depth_done = self._minimax_sim_id(
my_body=fb, enemy_body=enemy["body"],
mm_val, depth_done = self._minimax_candidate_id(
my_body=my_body, enemy_body=enemy["body"],
my_target=(pos["x"], pos["y"]),
food_set=food_set, hazard_set=hazard_set,
my_health=nmy_h, enemy_health=enemy_health,
my_health=my_health, enemy_health=enemy_health,
hazard_damage=hazard_damage, hazard_count=hazard_count,
width=width, height=height,
max_depth=self._planning_depth,
@@ -893,6 +889,55 @@ class ApexBattleSnake(TemplateSnake):
# ── A1: Iterative deepening minimax ──────────────────────────────────────────
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]:
"""Evaluate a selected move before continuing the legacy duel search.
Optimized subclasses can override this hook to resolve our selected move
and the opponent's reply simultaneously at the search root.
"""
pos = {"x": my_target[0], "y": my_target[1]}
ate = my_target in food_set
future_body = self._future_body(my_body, pos, ate, False)
future_health = 100 if ate else my_health - 1
effective_previous = previous_hazard_set if previous_hazard_set is not None else hazard_set
if my_target in hazard_set and my_target in effective_previous and not ate:
future_health -= hazard_damage * hazard_count.get(my_target, 1)
return self._minimax_sim_id(
my_body=future_body,
enemy_body=enemy_body,
food_set=food_set,
hazard_set=hazard_set,
my_health=future_health,
enemy_health=enemy_health,
hazard_damage=hazard_damage,
hazard_count=hazard_count,
width=width,
height=height,
max_depth=max_depth,
alpha=alpha,
beta=beta,
deadline=deadline,
previous_hazard_set=previous_hazard_set,
)
def _minimax_sim_id(
self,
my_body: list,
+171
View File
@@ -0,0 +1,171 @@
"""PrismBattleSnake_GPT_5_6_Sol v1.4.0
Built on ApexBattleSnake v1.0.0. All strategic logic is inherited.
Performance improvement: all spatial primitives (flood fill, territory,
articulation detection, distance maps, pathfinding) replaced by a
bitboard engine that uses integer arithmetic instead of Python sets/deques.
Key speedups:
S1: Bitboard flood fill replaces BFS deque+set with integer bit-expansion.
~60× faster per call, eliminates _neighbors() generator overhead.
S2: Bitboard territory dual-BFS expansion on ints replaces per-cell
distance-map comparison loop.
S3: Bitboard articulation partition sizes via bit-flood instead of
_bounded_bfs with sets.
S4: Bitboard distance map BFS via bit-expansion + bit-extract.
S5: Bitboard path distance early-exit BFS on ints.
S6: Bitboard nearest food BFS food search on ints.
S7: Per-turn BitBoard instance cached for board dimensions.
S8: Blocked-set bitboard conversion cached within a turn to avoid
redundant O(n) conversions for the same frozen set.
S9: Survival-tree uses bitboards natively enemy body/attack bits
precomputed once at tree root, no per-node set/dict rebuilds.
S10: _legal_moves override uses bitboard neighbour mask instead of
per-direction Python loop + _in_bounds calls.
S11: _future_survival_tree inlines legal-move check with bitboard ops.
S12: Duel minimax uses tuple bodies and bitboard move generation.
S13: Iterative deepening reuses a transposition table and move-order hints.
S14: Candidate duel moves and enemy replies resolve on the same root turn.
S15: Candidate moves share one duel transposition/search context per turn.
S16: Compact adversarial multiplayer rollout advances plausible enemy replies.
S17: Rollout memoization and adaptive depth spend time on ambiguous positions.
S18: Prism uses a deeper tactical horizon while retaining Apex's timeout reserve.
S19: Rollout occupancy and evaluation caches avoid repeated flood-fill work.
"""
from __future__ import annotations
from server.GameBoard import GameBoard
from snakes.engine.bitboard import BitBoard
from snakes.engine.duel import BitboardDuelMixin
from snakes.engine.duel_search import BitboardDuelSearch
from snakes.engine.spatial import BitboardSpatialMixin
from snakes.engine.survival import BitboardSurvivalMixin
from snakes.engine.survival_search import CompactSurvivalSearch
from snakes.strategies.apex import ApexBattleSnake
# Direction offsets for coord-dict → tuple conversion
_DIR_DELTAS = ((0, 1), (0, -1), (-1, 0), (1, 0))
_DIR_NAMES = ("up", "down", "left", "right")
class PrismBattleSnake_GPT_5_6_Sol(
BitboardDuelMixin,
BitboardSurvivalMixin,
BitboardSpatialMixin,
ApexBattleSnake,
):
VERSION = "1.4.0"
def __init__(self) -> None:
super().__init__()
self.name = "PrismBattleSnake"
self.version = self.VERSION
# Prism's compact state search is fast enough to inspect one additional
# turn. The existing deadline checks and Apex timeout reserve still cap the
# work on difficult positions.
self._planning_depth = max(self._planning_depth, 4)
# S7: cached BitBoard instance (reused while board dimensions stay the same)
self._bb: BitBoard | None = None
self._bb_w: int = 0
self._bb_h: int = 0
# S9: precomputed enemy state for survival tree (set per turn in choose_move)
self._enemy_body_bits: int = 0 # all enemy body cells as bitboard
self._enemy_tail_bits: int = 0 # enemy tails that will vacate
self._enemy_attack_danger: int = 0 # tiles where enemy len >= our len
self._enemy_attack_opportunity: int = 0 # tiles where enemy len < our len
# Shared per-turn search contexts. Candidate moves overlap heavily, so
# rebuilding their transposition tables wastes most iterative-deepening work.
self._duel_search_context: BitboardDuelSearch | None = None
self._survival_search_context: CompactSurvivalSearch | None = None
# ── choose_move override: precompute enemy bits ──────────────────────────
def choose_move(self, game_data: GameBoard) -> str:
bb = self._get_bb(game_data.get_width(), game_data.get_height())
self._duel_search_context = None
self._survival_search_context = None
# S9: precompute enemy body / tail / attack bitboards for survival tree
other_snakes = game_data.get_other_snakes()
my_snake = game_data.get_my_snake()
my_len = my_snake.get("length", len(my_snake["body"]))
food_set = {(f["x"], f["y"]) for f in game_data.get_food()}
all_occupied = {
(seg["x"], seg["y"])
for snake in [my_snake, *other_snakes]
for seg in snake["body"]
}
game_type = game_data.get_type()
is_constrictor = game_type == "constrictor"
w = bb.width
enemy_body_bits = 0
enemy_tail_bits = 0
enemy_attack_danger = 0
enemy_attack_opportunity = 0
for snake in other_snakes:
for seg in snake["body"]:
enemy_body_bits |= 1 << (seg["y"] * w + seg["x"])
body = snake["body"]
# Check if tail will vacate
if not is_constrictor and len(body) >= 2:
tail_stacked = (
body[-1]["x"] == body[-2]["x"] and body[-1]["y"] == body[-2]["y"]
)
if not tail_stacked:
can_grow = self._enemy_can_grow_this_turn(
snake, food_set, all_occupied
)
if not can_grow:
enemy_tail_bits |= 1 << (body[-1]["y"] * w + body[-1]["x"])
# Attack map: tiles enemy head can reach in 1 move
eh = snake["head"]
e_len = snake.get("length", len(body))
ehx, ehy = eh["x"], eh["y"]
for dx, dy in _DIR_DELTAS:
nx, ny = ehx + dx, ehy + dy
if 0 <= nx < w and 0 <= ny < bb.height:
bit = 1 << (ny * w + nx)
if e_len >= my_len:
enemy_attack_danger |= bit
else:
enemy_attack_opportunity |= bit
self._enemy_body_bits = enemy_body_bits
self._enemy_tail_bits = enemy_tail_bits
self._enemy_attack_danger = enemy_attack_danger
self._enemy_attack_opportunity = enemy_attack_opportunity
move = super().choose_move(game_data)
history = self.get_history()
if history:
thinking = history[-1]
if self._duel_search_context is not None:
thinking["prism_duel_depth"] = self._duel_search_context.completed_depth
thinking["prism_duel_nodes"] = self._duel_search_context.nodes
thinking["prism_duel_cache_hits"] = (
self._duel_search_context.cache_hits
+ self._duel_search_context.evaluation_cache_hits
)
thinking["prism_duel_deadline_exits"] = (
self._duel_search_context.deadline_exits
)
if self._survival_search_context is not None:
thinking["prism_rollout_depth"] = (
self._survival_search_context.completed_depth
)
thinking["prism_rollout_nodes"] = self._survival_search_context.nodes
thinking["prism_rollout_cache_hits"] = (
self._survival_search_context.cache_hits
+ self._survival_search_context.evaluation_cache_hits
)
thinking["prism_rollout_deadline_exits"] = (
self._survival_search_context.deadline_exits
)
return move
+1 -1
View File
@@ -3,7 +3,7 @@ import argparse
import time
from server.GameBoard import GameBoard
from snakes.BestBattleSnake import BestBattleSnake
from snakes.legacy.BestBattleSnake import BestBattleSnake
def build_game_state() -> dict:
return {
@@ -0,0 +1,54 @@
import io
import unittest
from unittest.mock import patch
from scripts.run_seeded_snake_tournament import ENGINE_USER_AGENT, run_game
class _Completed:
returncode = 0
stdout = "INFO Game completed after 42 turns. Prism was the winner.\n"
class _OutputFile:
def __init__(self, *args, **kwargs):
self.file = io.BytesIO(
b'{"turn":0}\n'
b'{"winnerId":"snake-id","winnerName":"Prism","isDraw":false}\n'
)
self.name = "arena-output.jsonl"
def __enter__(self):
return self
def __exit__(self, *args):
self.file.close()
def seek(self, offset):
return self.file.seek(offset)
def __iter__(self):
return iter(self.file)
class TestSeededSnakeTournament(unittest.TestCase):
def test_proxy_identifies_requests_as_battlesnake_engine(self):
self.assertIn("BattlesnakeEngine", ENGINE_USER_AGENT)
@patch("scripts.run_seeded_snake_tournament.subprocess.run", return_value=_Completed())
@patch("scripts.run_seeded_snake_tournament.tempfile.NamedTemporaryFile", _OutputFile)
def test_run_game_reads_official_engine_result(self, run):
result = run_game(
cli="battlesnake", seed=7, game_type="standard", map_name="standard",
players=[("Apex", "http://host:9001"), ("Prism", "http://host:9002")],
width=11, height=11, timeout_ms=500,
)
self.assertEqual(result, {
"seed": 7, "winner": "Prism", "draw": False, "turns": 42,
})
command = run.call_args.args[0]
self.assertIn("--seed", command)
self.assertIn("--output", command)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,31 @@
import unittest
from scripts.snake_arena_scenarios import SCENARIOS, synthetic_states
class TestSnakeArenaScenarios(unittest.TestCase):
def test_default_corpus_rotates_through_every_scenario(self):
states = synthetic_states(len(SCENARIOS))
self.assertEqual(
{metadata["scenario"] for _, metadata in states},
set(SCENARIOS),
)
def test_scenario_filter_is_deterministic(self):
first = synthetic_states(3, ["hazard"])
second = synthetic_states(3, ["hazard"])
self.assertEqual(first, second)
self.assertTrue(all(metadata["scenario"] == "hazard" for _, metadata in first))
def test_generated_you_is_present_on_board(self):
for board, metadata in synthetic_states(20):
with self.subTest(scenario=metadata["scenario"]):
ids = {snake["id"] for snake in board["snakes"]}
self.assertIn(metadata["you"]["id"], ids)
self.assertGreater(board["width"], 0)
self.assertGreater(board["height"], 0)
if __name__ == "__main__":
unittest.main()
@@ -2,10 +2,11 @@ import unittest
from time import perf_counter
from snakes import SnakeBuilder, get_snake_version
from snakes.ApexBattleSnake import ApexBattleSnake
from snakes.PrismBattleSnake_GPT_5_6_Sol import PrismBattleSnake_GPT_5_6_Sol
from snakes.bitboard import BitBoard
from snakes.bitboard_duel_search import BitboardDuelSearch
from snakes.engine.bitboard import BitBoard
from snakes.engine.duel_search import BitboardDuelSearch
from snakes.engine.survival_search import CompactSurvivalSearch
from snakes.strategies.apex import ApexBattleSnake
from snakes.strategies.prism import PrismBattleSnake_GPT_5_6_Sol
class TestBitBoard(unittest.TestCase):
@@ -21,20 +22,39 @@ class TestBitBoard(unittest.TestCase):
self.assertEqual(board.territory(board.idx(0, 0), [board.idx(4, 0)], 0), 0)
def test_territory_propagates_through_contested_cells(self):
board = BitBoard(5, 3)
blocked = board.set_to_bits({(0, 1), (1, 1), (3, 1), (4, 1)})
self.assertEqual(board.territory(board.idx(0, 0), [board.idx(4, 0)], blocked), 0)
def test_territory_ignores_enemy_only_disconnected_space_like_apex(self):
board = BitBoard(5, 1)
blocked = board.set_to_bits({(2, 0)})
self.assertEqual(board.territory(board.idx(0, 0), [board.idx(4, 0)], blocked), 2)
def test_nearest_food_returns_shortest_distance(self):
board = BitBoard(5, 5)
food = board.set_to_bits({(4, 4), (2, 1)})
self.assertEqual(board.nearest_food(board.idx(0, 0), food, 0), (3, board.idx(2, 1)))
def test_nearest_food_uses_apex_direction_order_for_ties(self):
board = BitBoard(3, 3)
food = board.set_to_bits({(1, 2), (0, 1), (2, 1), (1, 0)})
self.assertEqual(board.nearest_food(board.idx(1, 1), food, 0), (1, board.idx(1, 2)))
class TestPrismBattleSnake_GPT_5_6_Sol(unittest.TestCase):
def test_api_name_and_version_are_exposed(self):
snake = PrismBattleSnake_GPT_5_6_Sol()
self.assertEqual(snake.name, "PrismBattleSnake")
self.assertEqual(snake.version, "1.0.0")
self.assertEqual(get_snake_version("PrismBattleSnake_GPT_5_6_Sol"), "1.0.0")
self.assertEqual(snake.version, "1.4.0")
self.assertEqual(get_snake_version("PrismBattleSnake_GPT_5_6_Sol"), "1.4.0")
self.assertGreaterEqual(snake._planning_depth, 4)
self.assertIsInstance(SnakeBuilder.build("PrismBattleSnake_GPT_5_6_Sol"), PrismBattleSnake_GPT_5_6_Sol)
def test_bitboard_primitives_match_apex(self):
@@ -82,6 +102,128 @@ class TestPrismBattleSnake_GPT_5_6_Sol(unittest.TestCase):
self.assertGreater(value, 0)
def test_candidate_search_resolves_enemy_reply_on_the_same_turn(self):
board = BitBoard(3, 3)
search = BitboardDuelSearch(
board=board, food=set(), hazards=set(), hazard_count={},
hazard_damage=15, deadline=None,
)
my_body = [{"x": 0, "y": 1}, {"x": 0, "y": 0}]
enemy_body = [{"x": 2, "y": 1}, {"x": 2, "y": 0}]
value, depth = search.search_candidate(
my_body=my_body,
enemy_body=enemy_body,
my_target=(1, 1),
my_health=100,
enemy_health=100,
max_depth=1,
previous_hazards=set(),
)
self.assertEqual(value, -500.0)
self.assertEqual(depth, 1)
def test_candidate_search_does_not_advance_our_snake_twice_at_root(self):
board = BitBoard(4, 1)
search = BitboardDuelSearch(
board=board, food=set(), hazards=set(), hazard_count={},
hazard_damage=15, deadline=None,
)
my_body = [{"x": 0, "y": 0}]
enemy_body = [{"x": 3, "y": 0}]
value, depth = search.search_candidate(
my_body=my_body,
enemy_body=enemy_body,
my_target=(1, 0),
my_health=100,
enemy_health=100,
max_depth=1,
previous_hazards=set(),
)
self.assertEqual(value, 0.0)
self.assertEqual(depth, 1)
def test_duel_search_keeps_tail_blocked_when_its_snake_eats(self):
board = BitBoard(3, 3)
search = BitboardDuelSearch(
board=board, food={(0, 1)}, hazards=set(), hazard_count={},
hazard_damage=15, deadline=None,
)
body = (board.idx(0, 0), board.idx(1, 0), board.idx(1, 1))
advanced = search._advance_body(body, board.idx(0, 1), ate=True)
self.assertIn(board.idx(1, 1), advanced)
def test_candidate_duel_searches_share_the_same_context(self):
snake = PrismBattleSnake_GPT_5_6_Sol()
kwargs = {
"my_body": [{"x": 0, "y": 1}, {"x": 0, "y": 0}],
"enemy_body": [{"x": 3, "y": 1}, {"x": 3, "y": 0}],
"food_set": set(), "hazard_set": set(),
"my_health": 100, "enemy_health": 100,
"hazard_damage": 15, "hazard_count": {},
"width": 4, "height": 3, "max_depth": 2,
"alpha": -1e9, "beta": 1e9, "deadline": perf_counter() + 1.0,
}
snake._minimax_candidate_id(my_target=(1, 1), **kwargs)
first_context = snake._duel_search_context
snake._minimax_candidate_id(my_target=(0, 2), **kwargs)
self.assertIs(snake._duel_search_context, first_context)
self.assertGreater(first_context.nodes, 0)
self.assertGreater(first_context.completed_depth, 0)
def test_duel_evaluation_prioritizes_reachable_food_when_starving(self):
board = BitBoard(5, 3)
search = BitboardDuelSearch(
board=board, food={(2, 1)}, hazards=set(), hazard_count={},
hazard_damage=15, deadline=None,
)
my_body = [{"x": 0, "y": 1}, {"x": 0, "y": 0}]
enemy_body = [{"x": 4, "y": 1}, {"x": 4, "y": 0}]
hungry = search.search_depth(my_body, enemy_body, 15, 100, 0, set())
healthy = search.search_depth(my_body, enemy_body, 100, 100, 0, set())
self.assertLess(hungry, healthy)
def test_compact_rollout_models_lethal_enemy_head_response(self):
board = BitBoard(3, 3)
search = CompactSurvivalSearch(
board=board, food=set(), is_constrictor=False,
deadline=perf_counter() + 1.0, branch=2,
)
mine = [{"x": 0, "y": 1}, {"x": 0, "y": 0}]
enemies = [{
"body": [{"x": 2, "y": 1}, {"x": 2, "y": 0}, {"x": 1, "y": 0}],
}]
value = search.search_selected(mine, enemies, (1, 1), depth=1)
self.assertEqual(value, search.DEATH)
def test_compact_rollout_reuses_transpositions(self):
board = BitBoard(5, 5)
search = CompactSurvivalSearch(
board=board, food=set(), is_constrictor=False,
deadline=perf_counter() + 1.0, branch=2,
)
mine = [{"x": 1, "y": 1}, {"x": 1, "y": 0}]
enemies = [{"body": [{"x": 3, "y": 3}, {"x": 3, "y": 4}]}]
search.search_selected(mine, enemies, (2, 1), depth=3)
hits_before = search.cache_hits
search.search_selected(mine, enemies, (2, 1), depth=3)
self.assertGreater(search.cache_hits, hits_before)
self.assertGreater(search.evaluation_cache_hits, 0)
self.assertGreaterEqual(search.completed_depth, 3)
def test_bitboard_duel_search_reuses_transpositions(self):
board = BitBoard(5, 5)
search = BitboardDuelSearch(
+35
View File
@@ -0,0 +1,35 @@
import unittest
from snakes import SNAKE_REGISTRATIONS, SnakeBuilder
from snakes.engine import (
BitBoard,
BitboardDuelMixin,
BitboardSpatialMixin,
BitboardSurvivalMixin,
)
from snakes.strategies.prism import PrismBattleSnake_GPT_5_6_Sol
class TestSnakePackageLayout(unittest.TestCase):
def test_registry_uses_explicit_package_modules(self):
for name, registration in SNAKE_REGISTRATIONS.items():
with self.subTest(name=name):
self.assertTrue(registration.module.startswith("snakes."))
self.assertNotEqual(registration.module, f"snakes.{name}")
def test_registry_builds_active_strategies_after_package_move(self):
for name in ("ApexBattleSnake", "PrismBattleSnake_GPT_5_6_Sol"):
with self.subTest(name=name):
snake = SnakeBuilder.build(name)
self.assertEqual(snake.__class__.__name__, name)
def test_prism_composes_reusable_engine_mixins(self):
snake = PrismBattleSnake_GPT_5_6_Sol()
self.assertIsInstance(snake, BitboardDuelMixin)
self.assertIsInstance(snake, BitboardSpatialMixin)
self.assertIsInstance(snake, BitboardSurvivalMixin)
self.assertIsInstance(snake._get_bb(11, 11), BitBoard)
if __name__ == "__main__":
unittest.main()
+3 -3
View File
@@ -5,8 +5,8 @@ and that the bitboard engine itself is sound.
"""
import unittest
from snakes.SupremeBattleSnake_ClaudeOpus4_6 import SupremeBattleSnake_ClaudeOpus4_6 as SupremeBattleSnake
from snakes.bitboard import BitBoard
from snakes.legacy.SupremeBattleSnake_ClaudeOpus4_6 import SupremeBattleSnake_ClaudeOpus4_6 as SupremeBattleSnake
from snakes.engine.bitboard import BitBoard
from server.GameBoard import GameBoard
# ── Helpers ───────────────────────────────────────────────────────────────────
@@ -348,7 +348,7 @@ class TestParityWithApex(unittest.TestCase):
def test_trapped_corner(self):
"""Both snakes should survive a forced single-exit scenario."""
from snakes.ApexBattleSnake import ApexBattleSnake
from snakes.strategies.apex import ApexBattleSnake
state = gs(my_body=[(1, 1), (1, 2), (2, 2), (2, 1)],
other_bodies=[], foods=[(5, 5)], width=7, height=7)
+1 -1
View File
@@ -1,6 +1,6 @@
import unittest
from snakes.UltimateBattleSnake import UltimateBattleSnake
from snakes.legacy.UltimateBattleSnake import UltimateBattleSnake
from server.GameBoard import GameBoard
# ── Helpers ───────────────────────────────────────────────────────────────────
+1 -1
View File
@@ -1,6 +1,6 @@
import unittest
from snakes.BestBattleSnake import BestBattleSnake
from snakes.legacy.BestBattleSnake import BestBattleSnake
from server.GameBoard import GameBoard
def make_board(game_state):
+32
View File
@@ -0,0 +1,32 @@
import subprocess
import sys
import unittest
from pathlib import Path
class TestDatabasePackageImports(unittest.TestCase):
def test_sqlite_backend_import_does_not_require_aiofiles(self):
project_root = Path(__file__).resolve().parents[1]
script = """
import builtins
original_import = builtins.__import__
def reject_aiofiles(name, *args, **kwargs):
if name == 'aiofiles' or name.startswith('aiofiles.'):
raise ModuleNotFoundError("aiofiles intentionally unavailable")
return original_import(name, *args, **kwargs)
builtins.__import__ = reject_aiofiles
from server.database.backend.SqliteGameplayBackend import SqliteGameplayBackend
assert SqliteGameplayBackend.__name__ == 'SqliteGameplayBackend'
"""
result = subprocess.run(
[sys.executable, "-c", script],
cwd=project_root,
text=True,
capture_output=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
if __name__ == "__main__":
unittest.main()
+1 -1
View File
@@ -1,5 +1,5 @@
import unittest
from snakes.MasterSnake import MasterSnake
from snakes.legacy.MasterSnake import MasterSnake
class TestMasterSnake(unittest.TestCase):
def setUp(self):
+15 -4
View File
@@ -27,10 +27,14 @@ class TestMergeGameplayDatabases(unittest.TestCase):
90 if cleaned else None, "high" if cleaned else None,
'["already_scored"]' if cleaned else None,
))
connection.execute(
"INSERT INTO game_snakes (game_id,snake_id,snake_name,is_you) VALUES (?,?,?,?)",
(game_id, "me", "PrismBattleSnake", 1),
)
connection.execute("""
INSERT INTO game_snakes (
game_id,snake_id,snake_name,is_you,customizations_json
) VALUES (?,?,?,?,?)
""", (
game_id, "me", "PrismBattleSnake", 1,
'{"color":"#663399","head":"ferret","tail":"swirl"}',
))
connection.execute("""
INSERT INTO turns (
game_id,turn,observed_at,my_move,my_thinking_json,
@@ -69,6 +73,13 @@ class TestMergeGameplayDatabases(unittest.TestCase):
])
self.assertEqual(connection.execute("SELECT COUNT(*) FROM turns").fetchone()[0], 2)
self.assertEqual(connection.execute("SELECT COUNT(*) FROM snake_turns").fetchone()[0], 2)
customizations = connection.execute("""
SELECT game_id, customizations_json FROM game_snakes ORDER BY game_id
""").fetchall()
self.assertEqual(customizations, [
("base-game", '{"color":"#663399","head":"ferret","tail":"swirl"}'),
("delta-game", '{"color":"#663399","head":"ferret","tail":"swirl"}'),
])
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
def test_conflicting_duplicate_aborts_without_destination(self):
+137
View File
@@ -0,0 +1,137 @@
import sqlite3
import tempfile
import unittest
from pathlib import Path
from scripts.migrate_gameplay_database import copy_game_snakes
from server.database.backend.SqliteGameplayBackend import SqliteGameplayBackend
class TestMigrateGameplayDatabase(unittest.TestCase):
def test_copy_game_snakes_preserves_customizations(self):
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
source_path = root / "source.sqlite3"
destination_path = root / "destination.sqlite3"
SqliteGameplayBackend(str(source_path))
SqliteGameplayBackend(str(destination_path))
with sqlite3.connect(source_path) as source:
source.execute("PRAGMA foreign_keys = OFF")
source.execute("""
INSERT INTO game_snakes (
game_id, snake_id, snake_name, is_you, customizations_json
) VALUES (?, ?, ?, ?, ?)
""", (
"game-1", "snake-1", "PrismBattleSnake", 1,
'{"color":"#663399","head":"ferret","tail":"swirl"}',
))
source = sqlite3.connect(source_path)
destination = sqlite3.connect(destination_path)
try:
copied = copy_game_snakes(
source, destination, batch_size=10, retained_ids={"game-1"},
)
destination.commit()
row = destination.execute("""
SELECT snake_name, is_you, customizations_json
FROM game_snakes WHERE game_id = ? AND snake_id = ?
""", ("game-1", "snake-1")).fetchone()
finally:
source.close()
destination.close()
self.assertEqual(copied, 1)
self.assertEqual(row, (
"PrismBattleSnake", 1,
'{"color":"#663399","head":"ferret","tail":"swirl"}',
))
def test_copy_game_snakes_synthesizes_rows_when_table_is_empty(self):
source = sqlite3.connect(":memory:")
destination = sqlite3.connect(":memory:")
try:
source.execute("""
CREATE TABLE game_snakes (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER,
customizations_json TEXT NOT NULL DEFAULT '{}'
)
""")
source.execute("""
CREATE TABLE snake_turns (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER
)
""")
source.executemany(
"INSERT INTO snake_turns VALUES (?, ?, ?, ?)",
[
("game-1", "snake-1", "PrismBattleSnake", 1),
("game-1", "snake-1", "PrismBattleSnake", 1),
("game-1", "snake-2", "Enemy", 0),
],
)
destination.execute("""
CREATE TABLE game_snakes (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER,
customizations_json TEXT NOT NULL DEFAULT '{}',
PRIMARY KEY (game_id, snake_id)
)
""")
copied = copy_game_snakes(
source, destination, batch_size=10, retained_ids={"game-1"},
)
rows = destination.execute("""
SELECT snake_id, snake_name, is_you, customizations_json
FROM game_snakes ORDER BY snake_id
""").fetchall()
finally:
source.close()
destination.close()
self.assertEqual(copied, 2)
self.assertEqual(rows, [
("snake-1", "PrismBattleSnake", 1, "{}"),
("snake-2", "Enemy", 0, "{}"),
])
def test_copy_game_snakes_defaults_legacy_schema_to_empty_customizations(self):
source = sqlite3.connect(":memory:")
destination = sqlite3.connect(":memory:")
try:
source.execute("""
CREATE TABLE game_snakes (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER
)
""")
source.execute(
"INSERT INTO game_snakes VALUES (?, ?, ?, ?)",
("game-1", "snake-1", "LegacySnake", 0),
)
source.execute("""
CREATE TABLE snake_turns (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER
)
""")
destination.execute("""
CREATE TABLE game_snakes (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER,
customizations_json TEXT NOT NULL DEFAULT '{}'
)
""")
copied = copy_game_snakes(
source, destination, batch_size=10, retained_ids={"game-1"},
)
row = destination.execute(
"SELECT customizations_json FROM game_snakes"
).fetchone()
finally:
source.close()
destination.close()
self.assertEqual(copied, 1)
self.assertEqual(row, ("{}",))
if __name__ == "__main__":
unittest.main()
+1 -1
View File
@@ -12,7 +12,7 @@ in the folder where this file exists:
"""
import unittest
from snakes.LogicSnake import avoid_my_neck
from snakes.legacy.LogicSnake import avoid_my_neck
class AvoidNeckTest(unittest.TestCase):
Generated
+2
View File
@@ -345,6 +345,7 @@ name = "snake-python"
version = "0.1.0"
source = { virtual = "." }
dependencies = [
{ name = "aiofiles" },
{ name = "aiologger" },
{ name = "asyncpg" },
{ name = "dotenv" },
@@ -356,6 +357,7 @@ dependencies = [
[package.metadata]
requires-dist = [
{ name = "aiofiles", specifier = ">=25.1.0" },
{ name = "aiologger", specifier = ">=0.7.0" },
{ name = "asyncpg", specifier = ">=0.31.0" },
{ name = "dotenv", specifier = ">=0.9.9" },