From 4151810f1bde19f967d208e68dd97a46c912f564 Mon Sep 17 00:00:00 2001 From: Daniel Dolezal Date: Sun, 5 Apr 2026 16:48:12 +0200 Subject: [PATCH] add GameplayDatabase database with dashboard --- server/Server.py | 119 ++++- server/bootstrap.py | 10 + server/database/GameplayDatabase.py | 478 ++++++++++++++++++ server/database/__init__.py | 1 + server/templates/dashboard.html | 756 ++++++++++++++++++++++++++++ tests/test_GameplayDatabase.py | 111 ++++ 6 files changed, 1471 insertions(+), 4 deletions(-) create mode 100644 server/database/GameplayDatabase.py create mode 100644 server/database/__init__.py create mode 100644 server/templates/dashboard.html create mode 100644 tests/test_GameplayDatabase.py diff --git a/server/Server.py b/server/Server.py index b77dc53..277b2aa 100644 --- a/server/Server.py +++ b/server/Server.py @@ -7,13 +7,14 @@ from server.GameBoard import GameBoard from snakes import SnakeBuilder from server.storage import StorageLoader +from server.database import GameplayDatabase from server.metrics import ( MetricsStoreBuilder, MetricsCollector, ) -from quart import Quart, request, jsonify +from quart import Quart, request, jsonify, render_template import logging, json, os, re, time from typing import cast @@ -27,7 +28,7 @@ class Server: 'version': '1.0.0', } - def __init__(self, data_path:str, snake_type:str, storage_type:str, debug:bool=False, check_tls_security:bool=False, game_state_backend:str='memory', game_state_redis_url:str='redis://localhost:6379/0', game_state_ttl_sec:int=900, game_state_local_cache:bool=True, metrics_backend:str='memory', metrics_redis_url:str='redis://localhost:6379/0', metrics_ttl_sec:int|None=None): + def __init__(self, data_path:str, snake_type:str, storage_type:str, debug:bool=False, check_tls_security:bool=False, game_state_backend:str='memory', game_state_redis_url:str='redis://localhost:6379/0', game_state_ttl_sec:int=900, game_state_local_cache:bool=True, metrics_backend:str='memory', metrics_redis_url:str='redis://localhost:6379/0', metrics_ttl_sec:int|None=None, gameplay_db_enabled:bool=True, gameplay_db_path:str|None=None, gameplay_db_busy_timeout_ms:int=5000): self.debug = debug self.snake_type = snake_type self.storage_type = storage_type @@ -71,8 +72,15 @@ class Server: self.logger = build_logger('Battlesnake', debug_env_var='DEBUG_SERVER') self.snake_version = self._get_snake_version() + self.gameplay_database = None + if gameplay_db_enabled: + db_path = gameplay_db_path or os.path.join(data_path, 'data', 'database', 'gameplay.sqlite3') + self.gameplay_database = GameplayDatabase( + db_path=db_path, + busy_timeout_ms=gameplay_db_busy_timeout_ms, + ) - self.app = Quart('Battlesnake') + self.app = Quart('Battlesnake', template_folder=os.path.join(data_path, 'server', 'templates')) # info is called when you create your Battlesnake on play.battlesnake.com # and controls your Battlesnake's appearance @@ -92,6 +100,7 @@ class Server: await self._prune_stale_games() game_state = await request.get_json() await self._create_game_board(game_state) + await self._record_gameplay_start(game_state) await await_log(self.logger.info(f'GAME START: {game_state['game']}')) return 'ok' @@ -104,6 +113,7 @@ class Server: game_board = cast(GameBoard, await self._get_game_board(game_state)) next_move = game_board.snake_neat_make_a_move() await self._persist_game_board(game_state['game']['id'], game_board) + await self._record_gameplay_turn(game_state, next_move, game_board) elapsed_ms = (time.perf_counter() - move_started) * 1000.0 await self.metrics_collector.record_move(next_move, elapsed_ms) @@ -158,6 +168,8 @@ class Server: async def shutdown_state_storage(): await self.game_state_store.close() await self.metrics_collector.close() + if self.gameplay_database is not None: + await self.gameplay_database.close() @self.app.get('/cleanup') async def cleanup(): @@ -178,6 +190,36 @@ class Server: {'Content-Type': 'text/plain; version=0.0.4; charset=utf-8'}, ) + @self.app.get('/dashboard/summary') + async def dashboard_summary(): + summary = await self._get_dashboard_summary() + return jsonify(summary) + + @self.app.get('/dashboard') + async def dashboard_view(): + initial_game_id = request.args.get('game_id', '') + return await render_template( + 'dashboard.html', + initial_game_id=initial_game_id, + ) + + @self.app.get('/dashboard/games') + async def dashboard_games(): + raw_limit = request.args.get('limit', '50') + try: + limit = max(1, min(200, int(raw_limit))) + except ValueError: + limit = 50 + games = await self._get_dashboard_games(limit) + return jsonify(games) + + @self.app.get('/dashboard/game/') + async def dashboard_game_replay(game_id:str): + replay = await self._get_dashboard_game_replay(game_id) + if replay is None: + return jsonify({'error':'game_not_found', 'game_id':game_id}), 404 + return jsonify(replay) + async def run(self, host:str='0.0.0.0', port:int=8000, debug:bool=False): logging.getLogger('werkzeug').setLevel(logging.ERROR) @@ -278,7 +320,7 @@ class Server: async def _get_game_board(self, game_state:dict, end:bool=False) -> GameBoard: game_id = game_state['game']['id'] - game_board: GameBoard + game_board:GameBoard if self.game_state_local_cache and game_id in self.running_games: game_board = self.running_games[game_id] else: @@ -325,3 +367,72 @@ class Server: self.game_move_counts.pop(game_id, None) self.game_last_seen_unix.pop(game_id, None) await self.metrics_collector.record_stuck_removed() + + async def _record_gameplay_start(self, game_state:dict) -> None: + if self.gameplay_database is None: + return + try: + await self.gameplay_database.record_game_start(game_state) + except Exception as error: + await await_log(self.logger.warning(f'Gameplay DB start record failed:{error}')) + + def _extract_latest_snake_thinking(self, game_board:GameBoard) -> dict | None: + try: + history = game_board.snake_class.get_history() + except Exception: + return None + if not isinstance(history, list) or len(history) == 0: + return None + latest = history[-1] + return latest if isinstance(latest, dict) else None + + async def _record_gameplay_turn(self, game_state:dict, my_move:str, game_board:GameBoard) -> None: + if self.gameplay_database is None: + return + try: + thinking = self._extract_latest_snake_thinking(game_board) + await self.gameplay_database.record_turn(game_state, my_move, thinking) + except Exception as error: + await await_log(self.logger.warning(f'Gameplay DB turn record failed:{error}')) + + async def _record_gameplay_end(self, game_state:dict) -> None: + if self.gameplay_database is None: + return + try: + await self.gameplay_database.record_game_end(game_state) + except Exception as error: + await await_log(self.logger.warning(f'Gameplay DB end record failed:{error}')) + + async def _get_dashboard_summary(self) -> dict: + if self.gameplay_database is None: + return {'enabled':False} + try: + summary = await self.gameplay_database.get_summary() + summary['enabled'] = True + return summary + except Exception as error: + await await_log(self.logger.warning(f'Gameplay DB summary failed:{error}')) + return {'enabled':True, 'error':'summary_unavailable'} + + async def _get_dashboard_games(self, limit:int=50) -> dict: + if self.gameplay_database is None: + return {'enabled':False, 'games':[]} + try: + games = await self.gameplay_database.list_games(limit=limit) + return {'enabled':True, 'games':games} + except Exception as error: + await await_log(self.logger.warning(f'Gameplay DB game list failed:{error}')) + return {'enabled':True, 'error':'games_unavailable', 'games':[]} + + async def _get_dashboard_game_replay(self, game_id:str) -> dict | None: + if self.gameplay_database is None: + return {'enabled':False, 'error':'database_disabled', 'game_id':game_id} + try: + replay = await self.gameplay_database.get_game_replay(game_id) + if replay is None: + return None + replay['enabled'] = True + return replay + except Exception as error: + await await_log(self.logger.warning(f'Gameplay DB replay failed:{error}')) + return {'enabled':True, 'error':'replay_unavailable', 'game_id':game_id} diff --git a/server/bootstrap.py b/server/bootstrap.py index 5e6e6b2..63cd3d3 100644 --- a/server/bootstrap.py +++ b/server/bootstrap.py @@ -32,6 +32,13 @@ def build_server_from_env(default_snake_type:str) -> Server: else: metrics_ttl_sec = int(metrics_ttl_sec_raw) + gameplay_db_enabled = env_bool('GAMEPLAY_DB_ENABLED', True) + gameplay_db_path = os.environ.get( + 'GAMEPLAY_DB_PATH', + os.path.join(data_path, 'data', 'database', 'gameplay.sqlite3'), + ) + gameplay_db_busy_timeout_ms = int(os.environ.get('GAMEPLAY_DB_BUSY_TIMEOUT_MS', '5000')) + server = Server( data_path=data_path, snake_type=os.environ.get('SNAKE', default_snake_type), @@ -45,6 +52,9 @@ def build_server_from_env(default_snake_type:str) -> Server: metrics_backend=metrics_backend, metrics_redis_url=metrics_redis_url, metrics_ttl_sec=metrics_ttl_sec, + gameplay_db_enabled=gameplay_db_enabled, + gameplay_db_path=gameplay_db_path, + gameplay_db_busy_timeout_ms=gameplay_db_busy_timeout_ms, ) if env_bool('STORE_GAME_HISTORY'): diff --git a/server/database/GameplayDatabase.py b/server/database/GameplayDatabase.py new file mode 100644 index 0000000..e69d82e --- /dev/null +++ b/server/database/GameplayDatabase.py @@ -0,0 +1,478 @@ +from datetime import datetime, timezone +import asyncio, sqlite3, json +from pathlib import Path + +class GameplayDatabase: + def __init__(self, db_path:str, busy_timeout_ms:int = 5000): + self.db_path = db_path + self.busy_timeout_ms = max(1000, int(busy_timeout_ms)) + self._initialize_database() + + def _connect(self) -> sqlite3.Connection: + connection = sqlite3.connect( + self.db_path, + timeout=max(1, self.busy_timeout_ms // 1000), + isolation_level=None, + ) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys = ON") + connection.execute("PRAGMA journal_mode = WAL") + connection.execute("PRAGMA synchronous = NORMAL") + connection.execute(f"PRAGMA busy_timeout = {self.busy_timeout_ms}") + return connection + + def _initialize_database(self) -> None: + Path(self.db_path).parent.mkdir(parents=True, exist_ok=True) + with self._connect() as connection: + connection.executescript(""" + CREATE TABLE IF NOT EXISTS games ( + game_id TEXT PRIMARY KEY, + started_at TEXT NOT NULL, + ended_at TEXT, + width INTEGER, + height INTEGER, + source TEXT, + map_name TEXT, + ruleset_name TEXT, + ruleset_version TEXT, + your_snake_id TEXT, + your_snake_name TEXT, + winner_names_json TEXT, + winner_you INTEGER NOT NULL DEFAULT 0, + final_turn INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'running' + ); + + CREATE TABLE IF NOT EXISTS turns ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + game_id TEXT NOT NULL, + turn INTEGER NOT NULL, + observed_at TEXT NOT NULL, + my_move TEXT, + my_thinking_json TEXT, + board_state_json TEXT NOT NULL, + snakes_json TEXT NOT NULL, + you_json TEXT NOT NULL, + food_json TEXT NOT NULL, + hazards_json TEXT NOT NULL, + UNIQUE (game_id, turn), + FOREIGN KEY (game_id) REFERENCES games(game_id) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS snake_turns ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + game_id TEXT NOT NULL, + turn INTEGER NOT NULL, + snake_id TEXT NOT NULL, + snake_name TEXT, + health INTEGER, + length INTEGER, + head_x INTEGER, + head_y INTEGER, + body_json TEXT NOT NULL, + is_you INTEGER NOT NULL DEFAULT 0, + inferred_move TEXT, + UNIQUE (game_id, turn, snake_id), + FOREIGN KEY (game_id) REFERENCES games(game_id) ON DELETE CASCADE + ); + + CREATE INDEX IF NOT EXISTS idx_turns_game_turn ON turns(game_id, turn); + CREATE INDEX IF NOT EXISTS idx_games_status ON games(status); + CREATE INDEX IF NOT EXISTS idx_snake_turns_game_turn ON snake_turns(game_id, turn); + """) + self._ensure_column_exists(connection, "turns", "my_thinking_json", "TEXT") + + def _ensure_column_exists(self, connection:sqlite3.Connection, table_name:str, column_name:str, column_type:str) -> None: + existing = connection.execute(f"PRAGMA table_info({table_name})").fetchall() + if any(row["name"] == column_name for row in existing): + return + + connection.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_type}") + + def _utc_now(self) -> str: + return datetime.now(timezone.utc).isoformat() + + def _to_json(self, payload:dict) -> str: + return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + + def _from_json(self, payload:str|None): + if payload is None or payload == "": + return None + try: + return json.loads(payload) + except json.JSONDecodeError: + return None + + def _extract_snakes(self, game_state:dict) -> list[dict]: + return list(game_state.get("board", {}).get("snakes", [])) + + def _extract_you(self, game_state:dict) -> dict: + return dict(game_state.get("you", {})) + + def _infer_direction(self, old_head:tuple[int, int]|None, new_head:tuple[int, int]|None) -> str|None: + if old_head is None or new_head is None: + return None + + delta_x = new_head[0] - old_head[0] + delta_y = new_head[1] - old_head[1] + if delta_x == 1 and delta_y == 0: + return "right" + if delta_x == -1 and delta_y == 0: + return "left" + if delta_x == 0 and delta_y == 1: + return "up" + if delta_x == 0 and delta_y == -1: + return "down" + return None + + def _record_game_start_sync(self, game_state:dict) -> None: + game = game_state.get("game", {}) + board = game_state.get("board", {}) + you = self._extract_you(game_state) + ruleset = game.get("ruleset", {}) + + with self._connect() as connection: + connection.execute(""" + INSERT INTO games ( + game_id, started_at, width, height, source, map_name, + ruleset_name, ruleset_version, your_snake_id, your_snake_name, status + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'running') + ON CONFLICT(game_id) DO UPDATE SET + width = excluded.width, + height = excluded.height, + source = excluded.source, + map_name = excluded.map_name, + ruleset_name = excluded.ruleset_name, + ruleset_version = excluded.ruleset_version, + your_snake_id = excluded.your_snake_id, + your_snake_name = excluded.your_snake_name, + status = 'running' + """, + ( + game.get("id"), + self._utc_now(), + board.get("width"), + board.get("height"), + game.get("source"), + game.get("map"), + ruleset.get("name"), + ruleset.get("version"), + you.get("id"), + you.get("name"), + ) + ) + + def _record_turn_sync(self, game_state:dict, my_move:str|None, my_thinking:dict|None) -> None: + game = game_state.get("game", {}) + board = game_state.get("board", {}) + snakes = self._extract_snakes(game_state) + you = self._extract_you(game_state) + game_id = game.get("id") + turn = int(game_state.get("turn", 0)) + + with self._connect() as connection: + connection.execute(""" + INSERT INTO turns ( + game_id, turn, observed_at, my_move, my_thinking_json, + board_state_json, snakes_json, you_json, food_json, hazards_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(game_id, turn) DO UPDATE SET + observed_at = excluded.observed_at, + my_move = excluded.my_move, + my_thinking_json = excluded.my_thinking_json, + board_state_json = excluded.board_state_json, + snakes_json = excluded.snakes_json, + you_json = excluded.you_json, + food_json = excluded.food_json, + hazards_json = excluded.hazards_json + """, + ( + game_id, + turn, + self._utc_now(), + my_move, + self._to_json(my_thinking) if my_thinking is not None else None, + self._to_json(board), + self._to_json(snakes), + self._to_json(you), + self._to_json(board.get("food", [])), + self._to_json(board.get("hazards", [])), + ), + ) + + previous_positions:dict[str, tuple[int, int]] = {} + if turn > 0: + previous_rows = connection.execute(""" + SELECT snake_id, head_x, head_y + FROM snake_turns + WHERE game_id = ? AND turn = ? + """, + (game_id, turn - 1), + ).fetchall() + + previous_positions = { + row["snake_id"]: (int(row["head_x"]), int(row["head_y"])) + for row in previous_rows + if row["head_x"] is not None and row["head_y"] is not None + } + + you_id = you.get("id") + for snake in snakes: + snake_id = snake.get("id") + head = snake.get("head", {}) + head_x = head.get("x") + head_y = head.get("y") + if snake_id is None: + continue + + new_head = ( + (int(head_x), int(head_y)) + if head_x is not None and head_y is not None + else None + ) + inferred = self._infer_direction( + previous_positions.get(snake_id), new_head + ) + + connection.execute(""" + INSERT INTO snake_turns ( + game_id, turn, snake_id, snake_name, health, length, + head_x, head_y, body_json, is_you, inferred_move + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(game_id, turn, snake_id) DO UPDATE SET + snake_name = excluded.snake_name, + health = excluded.health, + length = excluded.length, + head_x = excluded.head_x, + head_y = excluded.head_y, + body_json = excluded.body_json, + is_you = excluded.is_you, + inferred_move = excluded.inferred_move + """, + ( + game_id, + turn, + snake_id, + snake.get("name"), + snake.get("health"), + snake.get("length"), + head_x, + head_y, + self._to_json(snake.get("body", [])), + 1 if snake_id == you_id else 0, + inferred, + ), + ) + + connection.execute(""" + UPDATE games + SET final_turn = CASE WHEN ? > final_turn THEN ? ELSE final_turn END + WHERE game_id = ? + """, + (turn, turn, game_id), + ) + + def _record_game_end_sync(self, game_state:dict) -> None: + game = game_state.get("game", {}) + game_id = game.get("id") + board = game_state.get("board", {}) + snakes = list(board.get("snakes", [])) + you = self._extract_you(game_state) + winner_names = [snake.get("name") for snake in snakes if snake.get("name")] + you_id = you.get("id") + winner_you = any(snake.get("id") == you_id for snake in snakes) + + with self._connect() as connection: + connection.execute(""" + UPDATE games + SET ended_at = ?, + winner_names_json = ?, + winner_you = ?, + final_turn = CASE WHEN ? > final_turn THEN ? ELSE final_turn END, + status = 'finished' + WHERE game_id = ? + """, + ( + self._utc_now(), + self._to_json(winner_names), + 1 if winner_you else 0, + int(game_state.get("turn", 0)), + int(game_state.get("turn", 0)), + game_id, + ), + ) + + def _get_summary_sync(self, recent_limit:int=15) -> dict: + with self._connect() as connection: + totals = connection.execute(""" + SELECT + COUNT(*) AS total_games, + SUM(CASE WHEN status = 'running' THEN 1 ELSE 0 END) AS running_games, + SUM(CASE WHEN status = 'finished' THEN 1 ELSE 0 END) AS finished_games, + SUM(CASE WHEN status = 'finished' AND winner_you = 1 THEN 1 ELSE 0 END) AS wins, + SUM(CASE WHEN status = 'finished' AND winner_you = 0 THEN 1 ELSE 0 END) AS losses, + AVG(CASE WHEN status = 'finished' THEN final_turn ELSE NULL END) AS avg_turns + FROM games + """ + ).fetchone() + + recent = connection.execute(""" + SELECT game_id, started_at, ended_at, map_name, your_snake_name, winner_you, final_turn, status + FROM games + ORDER BY started_at DESC + LIMIT ? + """, + (max(1, int(recent_limit)),), + ).fetchall() + + return { + "database": self.db_path, + "total_games": int(totals["total_games"] or 0), + "running_games": int(totals["running_games"] or 0), + "finished_games": int(totals["finished_games"] or 0), + "wins": int(totals["wins"] or 0), + "losses": int(totals["losses"] or 0), + "avg_turns_finished": round(float(totals["avg_turns"] or 0.0), 2), + "recent_games": [{ + "game_id": row["game_id"], + "started_at": row["started_at"], + "ended_at": row["ended_at"], + "map": row["map_name"], + "snake": row["your_snake_name"], + "winner_you": bool(row["winner_you"]), + "final_turn": int(row["final_turn"] or 0), + "status": row["status"], + } for row in recent ], + } + + def _list_games_sync(self, limit:int=50) -> list[dict]: + with self._connect() as connection: + rows = connection.execute(""" + SELECT game_id, started_at, ended_at, map_name, source, ruleset_name, + your_snake_name, winner_you, winner_names_json, final_turn, status + FROM games + ORDER BY started_at DESC + LIMIT ? + """, + (max(1, int(limit)),), + ).fetchall() + + return [{ + "game_id": row["game_id"], + "started_at": row["started_at"], + "ended_at": row["ended_at"], + "map": row["map_name"], + "source": row["source"], + "ruleset": row["ruleset_name"], + "snake": row["your_snake_name"], + "winner_you": bool(row["winner_you"]), + "winner_names": self._from_json(row["winner_names_json"]) or [], + "final_turn": int(row["final_turn"] or 0), + "status": row["status"], + } for row in rows] + + def _get_game_replay_sync(self, game_id:str) -> dict | None: + with self._connect() as connection: + game_row = connection.execute(""" + SELECT game_id, started_at, ended_at, width, height, source, map_name, + ruleset_name, ruleset_version, your_snake_id, your_snake_name, + winner_names_json, winner_you, final_turn, status + FROM games + WHERE game_id = ? + """, + (game_id,), + ).fetchone() + + if game_row is None: + return None + + turn_rows = connection.execute(""" + SELECT turn, observed_at, my_move, my_thinking_json, + board_state_json, food_json, hazards_json, you_json + FROM turns + WHERE game_id = ? + ORDER BY turn ASC + """, + (game_id,), + ).fetchall() + + snake_rows = connection.execute(""" + SELECT turn, snake_id, snake_name, health, length, head_x, head_y, + body_json, is_you, inferred_move + FROM snake_turns + WHERE game_id = ? + ORDER BY turn ASC, is_you DESC, snake_name ASC + """, + (game_id,), + ).fetchall() + + snakes_by_turn:dict[int, list[dict]] = {} + for row in snake_rows: + turn = int(row["turn"]) + snakes_by_turn.setdefault(turn, []).append({ + "snake_id": row["snake_id"], + "snake_name": row["snake_name"], + "health": row["health"], + "length": row["length"], + "head": {"x": row["head_x"], "y": row["head_y"]}, + "body": self._from_json(row["body_json"]) or [], + "is_you": bool(row["is_you"]), + "inferred_move": row["inferred_move"], + }) + + replay_turns = [] + for row in turn_rows: + turn_number = int(row["turn"]) + replay_turns.append({ + "turn": turn_number, + "observed_at": row["observed_at"], + "my_move": row["my_move"], + "my_thinking": self._from_json(row["my_thinking_json"]), + "board": self._from_json(row["board_state_json"]), + "food": self._from_json(row["food_json"]) or [], + "hazards": self._from_json(row["hazards_json"]) or [], + "you": self._from_json(row["you_json"]) or {}, + "snakes": snakes_by_turn.get(turn_number, []), + }) + + return { + "game": { + "game_id": game_row["game_id"], + "started_at": game_row["started_at"], + "ended_at": game_row["ended_at"], + "width": game_row["width"], + "height": game_row["height"], + "source": game_row["source"], + "map": game_row["map_name"], + "ruleset_name": game_row["ruleset_name"], + "ruleset_version": game_row["ruleset_version"], + "your_snake_id": game_row["your_snake_id"], + "your_snake_name": game_row["your_snake_name"], + "winner_names": self._from_json(game_row["winner_names_json"]) or [], + "winner_you": bool(game_row["winner_you"]), + "final_turn": int(game_row["final_turn"] or 0), + "status": game_row["status"], + }, + "turns": replay_turns, + } + + async def record_game_start(self, game_state:dict) -> None: + await asyncio.to_thread(self._record_game_start_sync, game_state) + + async def record_turn(self, game_state:dict, my_move:str|None, my_thinking:dict|None=None) -> None: + await asyncio.to_thread(self._record_turn_sync, game_state, my_move, my_thinking) + + async def record_game_end(self, game_state:dict) -> None: + await asyncio.to_thread(self._record_game_end_sync, game_state) + + async def get_summary(self, recent_limit:int=15) -> dict: + return await asyncio.to_thread(self._get_summary_sync, recent_limit) + + async def list_games(self, limit:int=50) -> list[dict]: + return await asyncio.to_thread(self._list_games_sync, limit) + + async def get_game_replay(self, game_id:str) -> dict|None: + return await asyncio.to_thread(self._get_game_replay_sync, game_id) + + async def close(self) -> None: + return None diff --git a/server/database/__init__.py b/server/database/__init__.py new file mode 100644 index 0000000..9dd9a60 --- /dev/null +++ b/server/database/__init__.py @@ -0,0 +1 @@ +from .GameplayDatabase import GameplayDatabase diff --git a/server/templates/dashboard.html b/server/templates/dashboard.html new file mode 100644 index 0000000..8184db0 --- /dev/null +++ b/server/templates/dashboard.html @@ -0,0 +1,756 @@ + + + + + + Snake Dashboard + + + +
+
+
+

Battlesnake Replay Dashboard

+

Full-screen replay with turn-by-turn move reasoning and snake state.

+
+
+
+ +
+
+
+

Games

+

Pick a game to inspect the match, move timeline, and why your snake chose each turn.

+
+
+ + + + + + + + + + +
GameStatusW/LTurns
+
+
+ +
+
+ + + + + + Turn - +
+ +
+
+
+ You + Enemy + Food + Hazard +
+
+
+ +
+
+
+
+
+ + + + diff --git a/tests/test_GameplayDatabase.py b/tests/test_GameplayDatabase.py new file mode 100644 index 0000000..1314c03 --- /dev/null +++ b/tests/test_GameplayDatabase.py @@ -0,0 +1,111 @@ +import unittest + +from pathlib import Path +import tempfile, sqlite3 + +from server.database import GameplayDatabase + +class TestGameplayDatabase(unittest.IsolatedAsyncioTestCase): + def _build_state(self, turn:int, me_head:tuple[int, int], enemy_head:tuple[int, int], include_enemy:bool=True) -> dict: + snakes = [ + { + "id": "me", + "name": "Me", + "health": 90, + "length": 3, + "head": {"x": me_head[0], "y": me_head[1]}, + "body": [ + {"x": me_head[0], "y": me_head[1]}, + {"x": me_head[0] - 1, "y": me_head[1]}, + {"x": me_head[0] - 2, "y": me_head[1]}, + ], + } + ] + + if include_enemy: + snakes.append({ + "id": "enemy", + "name": "Enemy", + "health": 90, + "length": 3, + "head": {"x": enemy_head[0], "y": enemy_head[1]}, + "body": [ + {"x": enemy_head[0], "y": enemy_head[1]}, + {"x": enemy_head[0], "y": enemy_head[1] + 1}, + {"x": enemy_head[0], "y": enemy_head[1] + 2}, + ], + }) + + return { + "turn": turn, + "game": { + "id": "game-abc", + "source": "league", + "map": "standard", + "ruleset": {"name": "standard", "version": "v1.0.0"}, + }, + "board": { + "width": 11, + "height": 11, + "food": [{"x": 2, "y": 2}], + "hazards": [], + "snakes": snakes, + }, + "you": snakes[0], + } + + async def test_records_gameplay_with_wal_and_inferred_moves(self): + with tempfile.TemporaryDirectory() as temp_dir: + db_path = Path(temp_dir) / "gameplay.sqlite3" + database = GameplayDatabase(str(db_path), busy_timeout_ms=4000) + + await database.record_game_start(self._build_state(turn=0, me_head=(1, 1), enemy_head=(5, 5))) + await database.record_turn( + self._build_state(turn=1, me_head=(2, 1), enemy_head=(5, 4)), + my_move="right", + my_thinking={ + "turn": 1, + "reason": "safe_space", + "scores": {"right": 1.0}, + }, + ) + await database.record_turn( + self._build_state(turn=2, me_head=(2, 2), enemy_head=(4, 4)), + my_move="up", + my_thinking={"turn": 2, "reason": "food", "scores": {"up": 1.4}}, + ) + await database.record_game_end(self._build_state(turn=2, me_head=(2, 2), enemy_head=(4, 4), include_enemy=False)) + + connection = sqlite3.connect(str(db_path)) + journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0] + self.assertEqual(str(journal_mode).lower(), "wal") + + games = connection.execute("SELECT status, winner_you, final_turn FROM games WHERE game_id = ?", ("game-abc",)).fetchone() + self.assertEqual(games[0], "finished") + self.assertEqual(games[1], 1) + self.assertEqual(games[2], 2) + + turns_count = connection.execute("SELECT COUNT(*) FROM turns WHERE game_id = ?", ("game-abc",)).fetchone()[0] + self.assertEqual(turns_count, 2) + + me_inferred = connection.execute("SELECT inferred_move FROM snake_turns WHERE game_id = ? AND turn = ? AND snake_id = ?", ("game-abc", 2, "me")).fetchone()[0] + enemy_inferred = connection.execute("SELECT inferred_move FROM snake_turns WHERE game_id = ? AND turn = ? AND snake_id = ?", ("game-abc", 2, "enemy")).fetchone()[0] + self.assertEqual(me_inferred, "up") + self.assertEqual(enemy_inferred, "left") + + summary = await database.get_summary() + self.assertEqual(summary["finished_games"], 1) + self.assertEqual(summary["wins"], 1) + + replay = await database.get_game_replay("game-abc") + self.assertIsNotNone(replay) + replay = replay or {} + self.assertEqual(replay["game"]["final_turn"], 2) + self.assertEqual(len(replay["turns"]), 2) + self.assertEqual(replay["turns"][1]["my_move"], "up") + self.assertEqual(replay["turns"][1]["my_thinking"]["reason"], "food") + + connection.close() + +if __name__ == "__main__": + unittest.main()