feat: add Prism snake and gameplay database lifecycle
Build and Push Docker Container / build-and-push (push) Successful in 8m3s
Build and Push Docker Container / build-and-push (push) Successful in 8m3s
- Add bitboard-accelerated Prism and versioned Supreme snake implementations. - Add database-backed move benchmarks and focused strategy tests. - Normalize gameplay storage while preserving replay compatibility. - Add deterministic game quality scoring and replay retention tiers. - Add backup-first SQLite cleanup, verification, and replacement tooling. - Add safe compact-plus-delta database merging with conflict detection. - Extend SQLite and PostgreSQL schemas for replay and quality metadata. - Add PostgreSQL development service and pytest import configuration. - Update gameplay documentation and the quart_common submodule revision.
This commit is contained in:
@@ -0,0 +1,137 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Benchmark snake move latency against sampled states from gameplay SQLite."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sqlite3
|
||||
from statistics import mean, median
|
||||
import sys
|
||||
from time import perf_counter
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from server.GameBoard import GameBoard
|
||||
from snakes import SnakeBuilder
|
||||
|
||||
def percentile(values: list[float], quantile: float) -> float:
|
||||
ordered = sorted(values)
|
||||
index = min(len(ordered) - 1, round((len(ordered) - 1) * quantile))
|
||||
return ordered[index]
|
||||
|
||||
def load_states(db_path: str, samples: int, stride: int) -> list[tuple[dict, dict]]:
|
||||
connection = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
|
||||
connection.execute("PRAGMA query_only = ON")
|
||||
max_id = int(connection.execute("SELECT max(id) FROM turns").fetchone()[0] or 0)
|
||||
if max_id == 0:
|
||||
return []
|
||||
|
||||
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,
|
||||
g.game_id, g.source, g.map_name,
|
||||
g.ruleset_name, g.ruleset_version, t.turn
|
||||
FROM turns AS t
|
||||
JOIN games AS g ON g.game_id = t.game_id
|
||||
WHERE t.id >= ?
|
||||
ORDER BY t.id
|
||||
LIMIT 1
|
||||
"""
|
||||
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])
|
||||
if not you:
|
||||
you = next(
|
||||
(snake for snake in board.get("snakes", []) if snake.get("id") == row[2]),
|
||||
{},
|
||||
)
|
||||
metadata = {
|
||||
"game_id": row[3],
|
||||
"source": row[4] or "custom",
|
||||
"map": row[5] or "standard",
|
||||
"ruleset": {
|
||||
"name": row[6] or "standard",
|
||||
"version": row[7] or "v1.0.0",
|
||||
"settings": {},
|
||||
},
|
||||
"turn": int(row[8]),
|
||||
}
|
||||
states.append((board, {"you": you, **metadata}))
|
||||
next_id += stride
|
||||
connection.close()
|
||||
return states
|
||||
|
||||
def benchmark(snake_name: str, states: list[tuple[dict, dict]], repeat: int) -> dict:
|
||||
durations: list[float] = []
|
||||
moves = 0
|
||||
for pass_number in range(repeat):
|
||||
for board_data, metadata in states:
|
||||
snake = SnakeBuilder.build(snake_name)
|
||||
game_id = f"benchmark-{pass_number}-{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,
|
||||
)
|
||||
state = {
|
||||
"game": {
|
||||
"id": game_id,
|
||||
"ruleset": metadata["ruleset"],
|
||||
"source": metadata["source"],
|
||||
"map": metadata["map"],
|
||||
"timeout": 500,
|
||||
},
|
||||
"turn": metadata["turn"],
|
||||
"board": board_data,
|
||||
"you": metadata["you"],
|
||||
}
|
||||
board.read_game_data(state)
|
||||
started = perf_counter()
|
||||
snake.choose_move(board)
|
||||
durations.append((perf_counter() - started) * 1000)
|
||||
moves += 1
|
||||
|
||||
return {
|
||||
"snake": snake_name,
|
||||
"moves": moves,
|
||||
"mean_ms": mean(durations),
|
||||
"median_ms": median(durations),
|
||||
"p95_ms": percentile(durations, 0.95),
|
||||
"max_ms": max(durations),
|
||||
}
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--database", required=True)
|
||||
parser.add_argument("--snake", action="append", default=[])
|
||||
parser.add_argument("--samples", type=int, default=100)
|
||||
parser.add_argument("--stride", type=int, default=997)
|
||||
parser.add_argument("--repeat", type=int, default=1)
|
||||
args = parser.parse_args()
|
||||
|
||||
states = load_states(args.database, max(1, args.samples), max(1, args.stride))
|
||||
if not states:
|
||||
raise SystemExit("No gameplay states found")
|
||||
|
||||
snake_names = args.snake or ["ApexBattleSnake", "PrismBattleSnake_GPT_5_6_Sol"]
|
||||
print(f"Loaded {len(states)} states from {args.database}")
|
||||
for snake_name in snake_names:
|
||||
result = benchmark(snake_name, states, max(1, args.repeat))
|
||||
print(
|
||||
f"{result['snake']}: {result['moves']} moves, "
|
||||
f"mean={result['mean_ms']:.2f} ms, median={result['median_ms']:.2f} ms, "
|
||||
f"p95={result['p95_ms']:.2f} ms, max={result['max_ms']:.2f} ms"
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user