Compare commits

...
5 Commits
24 changed files with 883 additions and 724 deletions
+2 -1
View File
@@ -1,7 +1,8 @@
from snakes.TemplateSnake import TemplateSnake
from datetime import datetime
class GameBoard:
def __init__(self, game_id:str, width:int, height:int, ruleset:dict, source:str, map:str, snake_class):
def __init__(self, game_id:str, width:int, height:int, ruleset:dict, source:str, map:str, snake_class:TemplateSnake):
self.id = game_id
self.width = width
self.height = height
+69 -638
View File
@@ -1,8 +1,8 @@
from quart_common.web.logger import build_logger, await_log
from quart_common.web.env import env_bool, env_int
from server.Files import read_file
from server.game_state_store import GameStateStoreBuilder
from server.GameBoard import GameBoard
from snakes import SnakeBuilder
@@ -14,16 +14,23 @@ from server.metrics import (
MetricsCollector,
)
from quart import (
Quart,
request,
jsonify,
render_template,
send_from_directory,
websocket,
)
import asyncio, signal, inspect, logging, json, os, re, time
import asyncio, signal, logging, json, os, re, time
from typing import cast
from quart import Quart
from server.blueprints import (
create_battlesnake_blueprint,
create_metrics_blueprint,
create_dashboard_blueprint,
)
from server.services import (
DashboardEventsService,
DashboardWebSocketHub,
GameRuntimeService,
GameplayTrackingService,
DashboardQueryService,
)
class Server:
default_snake_config = {
@@ -57,21 +64,16 @@ class Server:
self.metrics_redis_url = metrics_redis_url
self.stale_game_timeout_sec = self._get_stale_game_timeout_sec()
self.running_games:dict[str, GameBoard] = {}
self.game_move_counts:dict[str, int] = {}
self.game_last_seen_unix:dict[str, int] = {}
self.dashboard_game_subscribers:set[asyncio.Queue[str]] = set()
self.dashboard_game_subscribers_lock = asyncio.Lock()
self.dashboard_ws_tasks:set[asyncio.Task] = set()
self.dashboard_ws_tasks_lock = asyncio.Lock()
self.dashboard_ws_shutdown_event = asyncio.Event()
self.dashboard_ws_shutdown_message = json.dumps({'type': 'dashboard_ws_shutdown'})
self.dashboard_event_origin = f'worker-{os.getpid()}-{int(time.time() * 1000)}'
self.dashboard_events_channel = os.getenv('DASHBOARD_EVENTS_CHANNEL', 'snake:dashboard:events')
self.dashboard_events_enabled = (self.metrics_backend_normalized == 'redis' and self._env_bool('DASHBOARD_EVENTS_ENABLED', True))
self.dashboard_events_listener_task:asyncio.Task | None = None
self.dashboard_events_redis = None
self.dashboard_events_pubsub = None
self.game_runtime = GameRuntimeService(
game_state_store=self.game_state_store,
snake_type=self.snake_type,
game_state_local_cache=self.game_state_local_cache,
stale_game_timeout_sec=self.stale_game_timeout_sec,
)
self.dashboard_ws_hub = DashboardWebSocketHub()
dashboard_event_origin = f'worker-{os.getpid()}-{int(time.time() * 1000)}'
dashboard_events_channel = os.getenv('DASHBOARD_EVENTS_CHANNEL', 'snake:dashboard:events')
dashboard_events_enabled = (self.metrics_backend_normalized == 'redis' and env_bool('DASHBOARD_EVENTS_ENABLED', True))
self.metrics_collector = MetricsCollector(
metrics_manager=MetricsStoreBuilder.build(
@@ -84,15 +86,18 @@ class Server:
metrics_backend=metrics_backend_normalized,
game_state_backend=game_state_backend,
stale_game_timeout_sec=self.stale_game_timeout_sec,
game_last_seen_unix=self.game_last_seen_unix,
game_move_counts=self.game_move_counts,
game_last_seen_unix=self.game_runtime.game_last_seen_unix,
game_move_counts=self.game_runtime.game_move_counts,
)
self.clear_worker_metrics_on_startup = self._env_bool('METRICS_CLEAR_WORKERS_ON_STARTUP', True)
self.worker_metrics_startup_lock_ttl_sec = self._env_int('METRICS_STARTUP_CLEANUP_LOCK_TTL_SEC', 300)
self.game_runtime.attach_metrics_collector(self.metrics_collector)
self.clear_worker_metrics_on_startup = env_bool('METRICS_CLEAR_WORKERS_ON_STARTUP', True)
self.worker_metrics_startup_lock_ttl_sec = env_int('METRICS_STARTUP_CLEANUP_LOCK_TTL_SEC', 300)
self.dashboard_running_game_stale_sec = 600
self._startup_worker_metrics_cleared = False
self.logger = build_logger('Battlesnake', debug_env_var='DEBUG_SERVER')
self.snake_builder = SnakeBuilder
self.snake_version = self._get_snake_version()
self.gameplay_database = None
if gameplay_db_enabled:
@@ -102,76 +107,34 @@ class Server:
busy_timeout_ms=gameplay_db_busy_timeout_ms,
)
self.gameplay_tracking = GameplayTrackingService(
gameplay_database=self.gameplay_database,
snake_type=self.snake_type,
snake_version=self.snake_version,
logger=self.logger,
)
self.dashboard_query = DashboardQueryService(
gameplay_database=self.gameplay_database,
ws_hub=self.dashboard_ws_hub,
logger=self.logger,
dashboard_running_game_stale_sec=self.dashboard_running_game_stale_sec,
)
self.dashboard_events_service = DashboardEventsService(
enabled=dashboard_events_enabled,
redis_url=self.metrics_redis_url,
channel=dashboard_events_channel,
event_origin=dashboard_event_origin,
shutdown_event=self.dashboard_ws_hub.shutdown_event,
on_notice=self._on_dashboard_games_update_notice,
logger=self.logger,
)
self.dashboard_query.set_publish_notice(self.dashboard_events_service.publish_notice)
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
# TIP: If you open your Battlesnake URL in a browser you should see this data
@self.app.get('/')
async def on_info():
self.metrics_collector.record_http_request('info')
snake_config = await self._read_json_config_or_create()
await await_log(self.logger.info(f'INFO Snake: {snake_config}'))
return snake_config
# start is called when your Battlesnake begins a game
@self.app.post('/start')
async def on_start():
self.metrics_collector.record_http_request('start')
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'
# move is called when your Battlesnake game is running game
@self.app.post('/move')
async def on_move():
self.metrics_collector.record_http_request('move')
game_state = await request.get_json()
move_started = time.perf_counter()
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)
if self.debug:
await await_log(self.logger.debug(f'TURN: {game_state['turn']:3}, MOVE: {next_move:5}'))
return {'move': next_move}
# end is called when your Battlesnake finishes a game
@self.app.post('/end')
async def on_end():
self.metrics_collector.record_http_request('end')
await self._prune_stale_games()
game_state = await request.get_json()
if self.store_game_state:
game_board = cast(GameBoard, await self._get_game_board(game_state, end=True))
if self.check_tls_security:
await game_board.save(
StorageLoader.build(self.storage_type),
file_path=os.path.join(self.data_path, 'data'),
database=os.getenv('EDGEDB_DATABASE', None),
tls_security=None,
)
else:
await game_board.save(
StorageLoader.build(self.storage_type),
file_path=os.path.join(self.data_path, 'data'),
database=os.getenv('EDGEDB_DATABASE', None),
)
await self._record_gameplay_end(game_state)
await self._push_dashboard_games_update(game_state)
await await_log(self.logger.info(f'GAME ENDED: Winner is {[x['name'] for x in game_state['board']['snakes']]}'))
await self._delete_game_board(game_state)
await self.metrics_collector.record_game_end(game_state)
return 'ok'
self.app.register_blueprint(create_battlesnake_blueprint(self))
self.app.register_blueprint(create_metrics_blueprint(self))
self.app.register_blueprint(create_dashboard_blueprint(self))
@self.app.after_request
async def identify_server(response):
@@ -187,138 +150,25 @@ class Server:
should_clear = await self.metrics_collector.should_clear_worker_metrics_on_startup(self.worker_metrics_startup_lock_ttl_sec)
if should_clear:
await self.metrics_collector.clear_worker_metrics()
await self._start_dashboard_events_listener()
await self.dashboard_events_service.start_listener()
@self.app.after_serving
async def shutdown_state_storage():
await self._stop_dashboard_events_listener()
await self.dashboard_events_service.stop_listener()
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():
results = self._cleanup_database()
return jsonify(data=json.loads(results), status=200)
@self.app.get('/metrics')
async def metrics():
snapshot = await self.metrics_collector.build_snapshot(self.game_last_seen_unix, self.game_move_counts)
return jsonify(snapshot)
@self.app.get('/metrics/prometheus')
async def metrics_prometheus():
snapshot = await self.metrics_collector.build_snapshot(self.game_last_seen_unix, self.game_move_counts)
return (
self.metrics_collector.build_prometheus_metrics(snapshot),
200,
{'Content-Type': 'text/plain; version=0.0.4; charset=utf-8'},
)
@self.app.get('/dashboard')
async def dashboard_view():
initial_game_id = request.args.get('game_id', '')
initial_summary = await self._get_dashboard_summary()
initial_games = await self._get_dashboard_games(limit=100)
return await render_template(
'dashboard.html',
initial_game_id=initial_game_id,
initial_summary=initial_summary,
initial_games=initial_games,
)
@self.app.get('/dashboard/customizations/<path:asset_path>')
async def dashboard_customizations_asset(asset_path: str):
customization_root = os.path.join(self.data_path, 'server', 'static', 'customizations')
return await send_from_directory(customization_root, asset_path)
@self.app.websocket('/dashboard/ws/games')
async def dashboard_games_ws():
websocket_task = asyncio.current_task()
if websocket_task is not None:
await self._register_dashboard_ws_task(websocket_task)
subscriber_queue: asyncio.Queue[str] = asyncio.Queue(maxsize=20)
await self._register_dashboard_game_subscriber(subscriber_queue)
try:
initial_payload = await self._build_dashboard_games_event()
await asyncio.wait_for(websocket.send(json.dumps(initial_payload)), timeout=1.5)
while True:
queue_task = asyncio.create_task(subscriber_queue.get())
receive_task = asyncio.create_task(websocket.receive())
try:
done, _ = await asyncio.wait(
{queue_task, receive_task},
timeout=1.0,
return_when=asyncio.FIRST_COMPLETED,
)
if len(done) == 0:
if self.dashboard_ws_shutdown_event.is_set():
await asyncio.wait_for(
websocket.send(self.dashboard_ws_shutdown_message),
timeout=1.5,
)
break
continue
if receive_task in done:
try:
request_payload_raw = receive_task.result()
except Exception:
break
response_event = await self._handle_dashboard_ws_request(
request_payload_raw
)
if response_event is not None:
await asyncio.wait_for(
websocket.send(json.dumps(response_event)),
timeout=1.5,
)
if queue_task in done:
event_payload = queue_task.result()
if event_payload == self.dashboard_ws_shutdown_message:
await asyncio.wait_for(
websocket.send(event_payload), timeout=1.5
)
break
await asyncio.wait_for(
websocket.send(event_payload), timeout=1.5
)
except asyncio.TimeoutError:
if self.dashboard_ws_shutdown_event.is_set():
await asyncio.wait_for(
websocket.send(self.dashboard_ws_shutdown_message),
timeout=1.5,
)
break
finally:
for pending_task in (queue_task, receive_task):
if not pending_task.done():
pending_task.cancel()
await asyncio.gather(
queue_task, receive_task, return_exceptions=True
)
except asyncio.CancelledError:
pass
except Exception:
pass
finally:
await self._unregister_dashboard_game_subscriber(subscriber_queue)
if websocket_task is not None:
await self._unregister_dashboard_ws_task(websocket_task)
async def run(self, host:str='0.0.0.0', port:int=8000, debug:bool=False):
logging.getLogger('werkzeug').setLevel(logging.ERROR)
loop = asyncio.get_running_loop()
shutdown_event = asyncio.Event()
installed_signal_handlers:list[signal.Signals] = []
shutdown_event = asyncio.Event()
def on_shutdown_signal() -> None:
self._request_dashboard_ws_shutdown()
self.dashboard_ws_hub.request_shutdown()
shutdown_event.set()
async def shutdown_trigger() -> None:
@@ -335,7 +185,7 @@ class Server:
try:
await self.app.run_task(host=host, port=port, debug=debug, shutdown_trigger=shutdown_trigger)
finally:
self._request_dashboard_ws_shutdown()
self.dashboard_ws_hub.request_shutdown()
for shutdown_signal in installed_signal_handlers:
try:
loop.remove_signal_handler(shutdown_signal)
@@ -380,86 +230,7 @@ class Server:
return str(version)
def _get_stale_game_timeout_sec(self) -> int:
value = os.getenv('SNAKE_STUCK_GAME_TIMEOUT_SEC', '180')
try:
return max(30, int(value))
except ValueError:
return 180
def _env_bool(self, name:str, default:bool=False) -> bool:
value = os.getenv(name)
if value is None:
return default
return value.strip().lower() in {'1', 'true', 'yes', 'on'}
def _env_int(self, name: str, default: int) -> int:
value = os.getenv(name)
if value is None:
return default
try:
return int(value)
except ValueError:
return default
async def _create_game_board(self, game_state:dict) -> GameBoard:
game_id = game_state['game']['id']
new_game_board = GameBoard(
game_id=game_id,
width=game_state['board']['width'],
height=game_state['board']['height'],
ruleset=game_state['game']['ruleset'],
source=game_state['game']['source'],
map=game_state['game']['map'],
snake_class=SnakeBuilder.build(self.snake_type),
)
await new_game_board.start_game(game_state)
if self.game_state_local_cache:
self.running_games[game_id] = new_game_board
await self.game_state_store.save(game_id, new_game_board)
self.game_move_counts[game_id] = 0
self.game_last_seen_unix[game_id] = int(time.time())
await self.metrics_collector.record_game_started(len(self.game_last_seen_unix))
return new_game_board
async def _persist_game_board(self, game_id:str, game_board:GameBoard):
if self.game_state_local_cache:
self.running_games[game_id] = game_board
await self.game_state_store.save(game_id, game_board)
async def _delete_game_board(self, game_state:dict):
game_id = game_state['game']['id']
self.running_games.pop(game_id, None)
self.game_move_counts.pop(game_id, None)
self.game_last_seen_unix.pop(game_id, None)
await self.game_state_store.delete(game_id)
async def _get_game_board(self, game_state:dict, end:bool=False) -> GameBoard:
game_id = game_state['game']['id']
game_board:GameBoard
if self.game_state_local_cache and game_id in self.running_games:
game_board = self.running_games[game_id]
else:
persisted_board = await self.game_state_store.load(game_id)
if persisted_board is not None:
game_board = cast(GameBoard, persisted_board)
if self.game_state_local_cache:
self.running_games[game_id] = game_board
else:
game_board = await self._create_game_board(game_state)
await self.metrics_collector.record_game_autocreated()
if not end:
self.game_move_counts[game_id] = self.game_move_counts.get(game_id, 0) + 1
self.game_last_seen_unix[game_id] = int(time.time())
game_board.read_game_data(game_state)
if end:
game_board.end_game(game_state)
await self._persist_game_board(game_id, game_board)
return game_board
return max(30, env_int('SNAKE_STUCK_GAME_TIMEOUT_SEC', 180))
def enable_store_game_state(self):
self.store_game_state = True
@@ -468,345 +239,5 @@ class Server:
storage = StorageLoader.build(self.storage_type)()
return storage.cleanup()
async def _prune_stale_games(self):
if not self.game_last_seen_unix:
return
now = int(time.time())
stale_ids = [
game_id
for game_id, last_seen in self.game_last_seen_unix.items()
if now - last_seen >= self.stale_game_timeout_sec
]
for game_id in stale_ids:
self.running_games.pop(game_id, None)
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,
snake_type=self.snake_type,
snake_version=self.snake_version,
)
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 _register_dashboard_game_subscriber(self, subscriber_queue:asyncio.Queue[str]) -> None:
async with self.dashboard_game_subscribers_lock:
self.dashboard_game_subscribers.add(subscriber_queue)
async def _unregister_dashboard_game_subscriber(self, subscriber_queue:asyncio.Queue[str]) -> None:
async with self.dashboard_game_subscribers_lock:
self.dashboard_game_subscribers.discard(subscriber_queue)
async def _broadcast_dashboard_game_event(self, payload:dict) -> None:
encoded_payload = json.dumps(payload)
async with self.dashboard_game_subscribers_lock:
subscribers = tuple(self.dashboard_game_subscribers)
for subscriber_queue in subscribers:
if subscriber_queue.full():
try:
subscriber_queue.get_nowait()
except asyncio.QueueEmpty:
pass
try:
subscriber_queue.put_nowait(encoded_payload)
except asyncio.QueueFull:
continue
async def _register_dashboard_ws_task(self, websocket_task:asyncio.Task) -> None:
async with self.dashboard_ws_tasks_lock:
self.dashboard_ws_tasks.add(websocket_task)
async def _unregister_dashboard_ws_task(self, websocket_task:asyncio.Task) -> None:
async with self.dashboard_ws_tasks_lock:
self.dashboard_ws_tasks.discard(websocket_task)
def _request_dashboard_ws_shutdown(self) -> None:
if self.dashboard_ws_shutdown_event.is_set():
return
self.dashboard_ws_shutdown_event.set()
for subscriber_queue in tuple(self.dashboard_game_subscribers):
if subscriber_queue.full():
try:
subscriber_queue.get_nowait()
except asyncio.QueueEmpty:
pass
try:
subscriber_queue.put_nowait(self.dashboard_ws_shutdown_message)
except asyncio.QueueFull:
continue
async def _start_dashboard_events_listener(self) -> None:
if not self.dashboard_events_enabled:
return
if self.dashboard_events_listener_task is not None:
return
try:
import redis.asyncio as aioredis # type: ignore[import-not-found]
self.dashboard_events_redis = aioredis.from_url(self.metrics_redis_url)
self.dashboard_events_pubsub = self.dashboard_events_redis.pubsub()
await self.dashboard_events_pubsub.subscribe(self.dashboard_events_channel)
self.dashboard_events_listener_task = asyncio.create_task(
self._dashboard_events_listener_loop()
)
except Exception as error:
self.dashboard_events_listener_task = None
self.dashboard_events_pubsub = None
self.dashboard_events_redis = None
await await_log(self.logger.warning(f'Dashboard events listener disabled (redis unavailable): {error}'))
async def _stop_dashboard_events_listener(self) -> None:
listener_task = self.dashboard_events_listener_task
self.dashboard_events_listener_task = None
if listener_task is not None:
listener_task.cancel()
await asyncio.gather(listener_task, return_exceptions=True)
pubsub = self.dashboard_events_pubsub
self.dashboard_events_pubsub = None
if pubsub is not None:
try:
await pubsub.unsubscribe(self.dashboard_events_channel)
except Exception:
pass
close_method = getattr(pubsub, 'aclose', None)
if callable(close_method):
try:
maybe_result = close_method()
if inspect.isawaitable(maybe_result):
await maybe_result
except Exception:
pass
redis_client = self.dashboard_events_redis
self.dashboard_events_redis = None
if redis_client is not None:
close_method = getattr(redis_client, 'aclose', None)
if callable(close_method):
try:
maybe_result = close_method()
if inspect.isawaitable(maybe_result):
await maybe_result
except Exception:
pass
async def _dashboard_events_listener_loop(self) -> None:
pubsub = self.dashboard_events_pubsub
if pubsub is None:
return
try:
while not self.dashboard_ws_shutdown_event.is_set():
message = await pubsub.get_message(
ignore_subscribe_messages=True,
timeout=1.0,
)
if message is None:
continue
raw_data = message.get('data')
if isinstance(raw_data, bytes):
payload_raw = raw_data.decode('utf-8', errors='replace')
else:
payload_raw = str(raw_data)
try:
payload = json.loads(payload_raw)
except json.JSONDecodeError:
continue
if not isinstance(payload, dict):
continue
if payload.get('type') != 'dashboard_games_update_notice':
continue
if payload.get('origin') == self.dashboard_event_origin:
continue
notice_trigger = str(payload.get('trigger') or 'game_saved')
await self._push_dashboard_games_update(
game_state=None,
publish_cluster=False,
trigger=notice_trigger,
)
except asyncio.CancelledError:
pass
except Exception as error:
await await_log(
self.logger.warning(f'Dashboard events listener stopped unexpectedly: {error}'))
async def _publish_dashboard_games_update_notice(self, trigger: str) -> None:
if not self.dashboard_events_enabled:
return
if self.dashboard_events_redis is None:
return
if trigger not in {'game_saved', 'stale_finalized', 'manual'}:
return
message = {
'type': 'dashboard_games_update_notice',
'origin': self.dashboard_event_origin,
'trigger': trigger,
'sent_at': int(time.time()),
}
try:
await self.dashboard_events_redis.publish(
self.dashboard_events_channel,
json.dumps(message),
)
except Exception as error:
await await_log(self.logger.warning(f'Dashboard events publish failed: {error}'))
async def _build_dashboard_games_event(self, game_state:dict|None=None, trigger_override:str|None=None) -> dict:
games_payload = await self._get_dashboard_games(limit=100)
summary_payload = await self._get_dashboard_summary()
game_id = None
if game_state is not None:
game_id = game_state.get('game', {}).get('id')
trigger = trigger_override or ('game_saved' if game_id else 'snapshot')
return {
'type': 'dashboard_games_update',
'trigger': trigger,
'games': games_payload,
'summary': summary_payload,
}
async def _build_dashboard_game_replay_event(self, game_id:str, request_id:str|None=None) -> dict:
replay_payload = await self._get_dashboard_game_replay(game_id)
if replay_payload is None:
return {
'type': 'dashboard_game_replay',
'request_id': request_id,
'game_id': game_id,
'error': 'game_not_found',
}
return {
'type': 'dashboard_game_replay',
'request_id': request_id,
'game_id': game_id,
'replay': replay_payload,
}
async def _handle_dashboard_ws_request(self, payload_raw: object) -> dict | None:
if not isinstance(payload_raw, str):
return None
try:
payload = json.loads(payload_raw)
except json.JSONDecodeError:
return None
if not isinstance(payload, dict):
return None
if payload.get('type') != 'dashboard_game_replay_request':
return None
game_id = str(payload.get('game_id') or '').strip()
request_id_raw = payload.get('request_id')
request_id = None if request_id_raw is None else str(request_id_raw)
if game_id == '':
return {
'type': 'dashboard_game_replay',
'request_id': request_id,
'error': 'missing_game_id',
}
return await self._build_dashboard_game_replay_event(
game_id=game_id,
request_id=request_id,
)
async def _push_dashboard_games_update(self, game_state:dict|None=None, publish_cluster:bool=True, trigger:str|None=None) -> None:
if self.gameplay_database is None:
return
event_payload = await self._build_dashboard_games_event(
game_state,
trigger_override=trigger,
)
await self._broadcast_dashboard_game_event(event_payload)
if publish_cluster:
await self._publish_dashboard_games_update_notice(trigger=str(event_payload.get('trigger') or ''))
async def _get_dashboard_summary(self) -> dict:
if self.gameplay_database is None:
return {'enabled': False}
try:
await self._finalize_stale_dashboard_games()
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:
await self._finalize_stale_dashboard_games()
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 _finalize_stale_dashboard_games(self) -> None:
if self.gameplay_database is None:
return
try:
await self.gameplay_database.finalize_stale_running_games(stale_after_seconds=self.dashboard_running_game_stale_sec)
except Exception as error:
await await_log(self.logger.warning(f'Gameplay DB stale running game finalize failed:{error}'))
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}
async def _on_dashboard_games_update_notice(self, trigger:str) -> None:
await self.dashboard_query.on_dashboard_games_update_notice(trigger)
+3
View File
@@ -0,0 +1,3 @@
from .battlesnake import create_battlesnake_blueprint
from .metrics import create_metrics_blueprint
from .dashboard import create_dashboard_blueprint
+83
View File
@@ -0,0 +1,83 @@
from typing import TYPE_CHECKING, cast
import json, time, os
from quart import Blueprint, request, jsonify
from quart_common.web.logger import await_log
from server.storage import StorageLoader
from server.GameBoard import GameBoard
if TYPE_CHECKING:
from server.Server import Server
def create_battlesnake_blueprint(server:'Server') -> Blueprint:
blueprint = Blueprint('battlesnake', __name__)
@blueprint.get('/')
async def on_info():
server.metrics_collector.record_http_request('info')
snake_config = await server._read_json_config_or_create()
await await_log(server.logger.info(f'INFO Snake: {snake_config}'))
return snake_config
@blueprint.post('/start')
async def on_start():
server.metrics_collector.record_http_request('start')
await server.game_runtime.prune_stale_games()
game_state = await request.get_json()
await server.game_runtime.create_game_board(game_state, snake_builder=server.snake_builder)
await server.gameplay_tracking.record_gameplay_start(game_state)
await await_log(server.logger.info(f'GAME START: {game_state['game']}'))
return 'ok'
@blueprint.post('/move')
async def on_move():
server.metrics_collector.record_http_request('move')
game_state = await request.get_json()
move_started = time.perf_counter()
game_board = cast(GameBoard, await server.game_runtime.get_game_board(game_state, snake_builder=server.snake_builder))
next_move = game_board.snake_neat_make_a_move()
await server.game_runtime.persist_game_board(game_state['game']['id'], game_board)
await server.gameplay_tracking.record_gameplay_turn(game_state, next_move, game_board)
elapsed_ms = (time.perf_counter() - move_started) * 1000.0
await server.metrics_collector.record_move(next_move, elapsed_ms)
if server.debug:
await await_log(server.logger.debug(f'TURN: {game_state['turn']:3}, MOVE: {next_move:5}'))
return {'move': next_move}
@blueprint.post('/end')
async def on_end():
server.metrics_collector.record_http_request('end')
await server.game_runtime.prune_stale_games()
game_state = await request.get_json()
if server.store_game_state:
game_board = cast(GameBoard, await server.game_runtime.get_game_board(game_state, snake_builder=server.snake_builder, end=True))
if server.check_tls_security:
await game_board.save(
StorageLoader.build(server.storage_type),
file_path=os.path.join(server.data_path, 'data'),
database=os.getenv('EDGEDB_DATABASE', None),
tls_security=None,
)
else:
await game_board.save(
StorageLoader.build(server.storage_type),
file_path=os.path.join(server.data_path, 'data'),
database=os.getenv('EDGEDB_DATABASE', None),
)
await server.gameplay_tracking.record_gameplay_end(game_state)
await server.dashboard_query.push_dashboard_games_update(game_state)
await await_log(server.logger.info(f'GAME ENDED: Winner is {[x['name'] for x in game_state['board']['snakes']]}'))
await server.game_runtime.delete_game_board(game_state)
await server.metrics_collector.record_game_end(game_state)
return 'ok'
@blueprint.get('/cleanup')
async def cleanup():
results = server._cleanup_database()
return jsonify(data=json.loads(results), status=200)
return blueprint
+119
View File
@@ -0,0 +1,119 @@
from typing import TYPE_CHECKING
import asyncio, json, os
from quart import (
Blueprint,
render_template,
send_from_directory,
request,
websocket,
)
if TYPE_CHECKING:
from server.Server import Server
def create_dashboard_blueprint(server:'Server') -> Blueprint:
blueprint = Blueprint('dashboard', __name__)
@blueprint.get('/dashboard')
async def dashboard_view():
initial_game_id = request.args.get('game_id', '')
initial_summary = await server.dashboard_query.get_dashboard_summary()
initial_games = await server.dashboard_query.get_dashboard_games(limit=100)
return await render_template(
'dashboard.html',
initial_game_id=initial_game_id,
initial_summary=initial_summary,
initial_games=initial_games,
)
@blueprint.get('/dashboard/customizations/<path:asset_path>')
async def dashboard_customizations_asset(asset_path:str):
customization_root = os.path.join(
server.data_path,
'server',
'static',
'customizations',
)
return await send_from_directory(customization_root, asset_path)
@blueprint.websocket('/dashboard/ws/games')
async def dashboard_games_ws():
ws_hub = server.dashboard_ws_hub
websocket_task = asyncio.current_task()
if websocket_task is not None:
await ws_hub.register_task(websocket_task)
subscriber_queue:asyncio.Queue[str] = asyncio.Queue(maxsize=20)
await ws_hub.register_subscriber(subscriber_queue)
try:
initial_payload = await server.dashboard_query.build_dashboard_games_event()
await asyncio.wait_for(
websocket.send(json.dumps(initial_payload)), timeout=1.5
)
while True:
queue_task = asyncio.create_task(subscriber_queue.get())
receive_task = asyncio.create_task(websocket.receive())
try:
done, _ = await asyncio.wait(
{queue_task, receive_task},
timeout=1.0,
return_when=asyncio.FIRST_COMPLETED,
)
if len(done) == 0:
if ws_hub.shutdown_event.is_set():
await asyncio.wait_for(
websocket.send(ws_hub.shutdown_message),
timeout=1.5,
)
break
continue
if receive_task in done:
try:
request_payload_raw = receive_task.result()
except Exception:
break
response_event = await server.dashboard_query.handle_dashboard_ws_request(request_payload_raw)
if response_event is not None:
await asyncio.wait_for(
websocket.send(json.dumps(response_event)),
timeout=1.5,
)
if queue_task in done:
event_payload = queue_task.result()
if event_payload == ws_hub.shutdown_message:
await asyncio.wait_for(
websocket.send(event_payload), timeout=1.5
)
break
await asyncio.wait_for(
websocket.send(event_payload), timeout=1.5
)
except asyncio.TimeoutError:
if ws_hub.shutdown_event.is_set():
await asyncio.wait_for(
websocket.send(ws_hub.shutdown_message),
timeout=1.5,
)
break
finally:
for pending_task in (queue_task, receive_task):
if not pending_task.done():
pending_task.cancel()
await asyncio.gather(
queue_task, receive_task, return_exceptions=True
)
except asyncio.CancelledError:
pass
except Exception:
pass
finally:
await ws_hub.unregister_subscriber(subscriber_queue)
if websocket_task is not None:
await ws_hub.unregister_task(websocket_task)
return blueprint
+30
View File
@@ -0,0 +1,30 @@
from quart import Blueprint, jsonify
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from server.Server import Server
def create_metrics_blueprint(server:'Server') -> Blueprint:
blueprint = Blueprint('metrics', __name__)
@blueprint.get('/metrics')
async def metrics():
snapshot = await server.metrics_collector.build_snapshot(
server.game_runtime.game_last_seen_unix,
server.game_runtime.game_move_counts,
)
return jsonify(snapshot)
@blueprint.get('/metrics/prometheus')
async def metrics_prometheus():
snapshot = await server.metrics_collector.build_snapshot(
server.game_runtime.game_last_seen_unix,
server.game_runtime.game_move_counts,
)
return (
server.metrics_collector.build_prometheus_metrics(snapshot),
200,
{'Content-Type': 'text/plain; version=0.0.4; charset=utf-8'},
)
return blueprint
+8 -11
View File
@@ -2,6 +2,7 @@ from typing import TypedDict
from pathlib import Path
import os
from quart_common.web.env import env_bool, env_int
from server.Server import Server
class RunConfig(TypedDict):
@@ -9,35 +10,31 @@ class RunConfig(TypedDict):
port: int
debug: bool
def env_bool(name:str, default:bool=False) -> bool:
value = os.environ.get(name)
if value is None:
return default
return value.lower() in {'1', 'true', 'yes', 'on'}
def build_server_from_env(default_snake_type:str) -> Server:
data_path = str(Path(__file__).resolve().parent.parent)
redis_url = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
game_state_backend = os.environ.get('GAME_STATE_BACKEND', 'memory')
game_state_redis_url = os.environ.get('GAME_STATE_REDIS_URL', 'redis://localhost:6379/0')
game_state_ttl_sec = int(os.environ.get('GAME_STATE_TTL_SEC', '900'))
game_state_redis_url = os.environ.get('GAME_STATE_REDIS_URL', redis_url)
game_state_ttl_sec = env_int('GAME_STATE_TTL_SEC', 900)
metrics_backend = os.environ.get('METRICS_BACKEND', None)
if metrics_backend is None:
metrics_backend = ('redis' if game_state_backend.strip().lower() == 'redis' else 'memory')
metrics_redis_url = os.environ.get('METRICS_REDIS_URL', game_state_redis_url)
metrics_redis_url = os.environ.get('METRICS_REDIS_URL', redis_url)
metrics_ttl_sec_raw = os.environ.get('METRICS_TTL_SEC', None)
if metrics_ttl_sec_raw is None:
metrics_ttl_sec = (game_state_ttl_sec if metrics_backend.strip().lower() == 'redis' else None)
else:
metrics_ttl_sec = int(metrics_ttl_sec_raw)
metrics_ttl_sec = env_int('METRICS_TTL_SEC', game_state_ttl_sec)
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'))
gameplay_db_busy_timeout_ms = env_int('GAMEPLAY_DB_BUSY_TIMEOUT_MS', 5000)
server = Server(
data_path=data_path,
@@ -65,6 +62,6 @@ def build_server_from_env(default_snake_type:str) -> Server:
def build_run_config() -> RunConfig:
return {
'host': os.environ.get('HOST', '0.0.0.0'),
'port': int(os.environ.get('PORT', '8000')),
'port': env_int('PORT', 8000),
'debug': env_bool('DEBUG'),
}
+5 -2
View File
@@ -1,9 +1,12 @@
from server.GameBoard import GameBoard
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from server.GameBoard import GameBoard
class Dataset:
VALID_MOVES = {"up", "down", "left", "right"}
def __init__(self, game_board:GameBoard):
def __init__(self, game_board:'GameBoard'):
self.game_board = game_board
def _did_we_win(self):
+3 -19
View File
@@ -2,34 +2,18 @@ from pathlib import Path
from typing import Any
import os
from quart_common.web.env import env_bool, env_int
from server.dataset.DatasetIO import DatasetIO
class RLBootstrapDataset:
def __init__(self):
self.enabled = self._env_bool("RL_BOOTSTRAP_ENABLED", default=False)
self.min_base_rows = self._env_int("RL_MIN_BASE_ROWS", default=5000)
self.enabled = env_bool("RL_BOOTSTRAP_ENABLED", default=False)
self.min_base_rows = env_int("RL_MIN_BASE_ROWS", default=5000)
self.base_dataset_path = Path(os.getenv("RL_BASE_DATASET", "data/dataset/best_moves.jsonl"))
self.output_path = Path(os.getenv("RL_BOOTSTRAP_OUTPUT", "data/dataset/rl_bootstrap.jsonl"))
self.max_bytes = int(float(os.getenv("RL_BOOTSTRAP_MAX_MB", "50")) * 1024 * 1024)
self.needs_more_data = False
@staticmethod
def _env_bool(name:str, default:bool=False) -> bool:
value = os.getenv(name)
if value is None:
return default
return value.lower() in {"1", "true", "yes", "on"}
@staticmethod
def _env_int(name:str, default:int) -> int:
value = os.getenv(name)
if value is None:
return default
try:
return int(value)
except ValueError:
return default
def refresh_state(self):
if not self.enabled:
self.needs_more_data = False
@@ -1,10 +1,13 @@
from server.GameBoard import GameBoard
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from server.GameBoard import GameBoard
class MemoryGameBoardStore:
def __init__(self, **kwargs):
self._state:dict[str, object] = {}
async def save(self, game_id:str, game_board:GameBoard) -> None:
async def save(self, game_id:str, game_board:'GameBoard') -> None:
self._state[game_id] = game_board
async def load(self, game_id:str):
@@ -1,6 +1,9 @@
from server.GameBoard import GameBoard
from typing import TYPE_CHECKING
import inspect, pickle
if TYPE_CHECKING:
from server.GameBoard import GameBoard
class RedisGameBoardStore:
def __init__(self, redis_url:str="redis://localhost:6379/0", key_prefix:str="snake:gameboard", ttl_seconds:int=900, **kwargs):
self.redis_url = redis_url
@@ -23,7 +26,7 @@ class RedisGameBoardStore:
def _key(self, game_id:str) -> str:
return f"{self.key_prefix}:{game_id}"
async def save(self, game_id:str, game_board:GameBoard) -> None:
async def save(self, game_id:str, game_board:'GameBoard') -> None:
redis = await self._get_redis()
payload = pickle.dumps(game_board, protocol=pickle.HIGHEST_PROTOCOL)
await redis.set(self._key(game_id), payload, ex=self.ttl_seconds)
+5
View File
@@ -0,0 +1,5 @@
from .dashboard_events import DashboardEventsService
from .dashboard_ws_hub import DashboardWebSocketHub
from .game_runtime import GameRuntimeService
from .gameplay_tracking import GameplayTrackingService
from .dashboard_query import DashboardQueryService
+127
View File
@@ -0,0 +1,127 @@
from quart_common.web.logger import await_log
from typing import Awaitable, Callable
import asyncio, inspect, json, time
class DashboardEventsService:
def __init__(self, enabled:bool, redis_url:str, channel:str, event_origin:str, shutdown_event:asyncio.Event, on_notice:Callable[[str], Awaitable[None]], logger):
self.enabled = enabled
self.redis_url = redis_url
self.channel = channel
self.event_origin = event_origin
self.shutdown_event = shutdown_event
self.on_notice = on_notice
self.logger = logger
self.listener_task:asyncio.Task|None=None
self.redis = None
self.pubsub = None
async def start_listener(self) -> None:
if not self.enabled:
return
if self.listener_task is not None:
return
try:
import redis.asyncio as aioredis # type: ignore[import-not-found]
self.redis = aioredis.from_url(self.redis_url)
self.pubsub = self.redis.pubsub()
await self.pubsub.subscribe(self.channel)
self.listener_task = asyncio.create_task(self._listener_loop())
except Exception as error:
self.listener_task = None
self.pubsub = None
self.redis = None
await await_log(self.logger.warning(f'Dashboard events listener disabled (redis unavailable): {error}'))
async def stop_listener(self) -> None:
listener_task = self.listener_task
self.listener_task = None
if listener_task is not None:
listener_task.cancel()
await asyncio.gather(listener_task, return_exceptions=True)
pubsub = self.pubsub
self.pubsub = None
if pubsub is not None:
try:
await pubsub.unsubscribe(self.channel)
except Exception:
pass
close_method = getattr(pubsub, 'aclose', None)
if callable(close_method):
try:
maybe_result = close_method()
if inspect.isawaitable(maybe_result):
await maybe_result
except Exception:
pass
redis_client = self.redis
self.redis = None
if redis_client is not None:
close_method = getattr(redis_client, 'aclose', None)
if callable(close_method):
try:
maybe_result = close_method()
if inspect.isawaitable(maybe_result):
await maybe_result
except Exception:
pass
async def publish_notice(self, trigger:str) -> None:
if not self.enabled:
return
if self.redis is None:
return
if trigger not in {'game_saved', 'stale_finalized', 'manual'}:
return
message = {
'type': 'dashboard_games_update_notice',
'origin': self.event_origin,
'trigger': trigger,
'sent_at': int(time.time()),
}
try:
await self.redis.publish(self.channel, json.dumps(message))
except Exception as error:
await await_log(self.logger.warning(f'Dashboard events publish failed: {error}'))
async def _listener_loop(self) -> None:
pubsub = self.pubsub
if pubsub is None:
return
try:
while not self.shutdown_event.is_set():
message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=1.0)
if message is None:
continue
raw_data = message.get('data')
if isinstance(raw_data, bytes):
payload_raw = raw_data.decode('utf-8', errors='replace')
else:
payload_raw = str(raw_data)
try:
payload = json.loads(payload_raw)
except json.JSONDecodeError:
continue
if not isinstance(payload, dict):
continue
if payload.get('type') != 'dashboard_games_update_notice':
continue
if payload.get('origin') == self.event_origin:
continue
notice_trigger = str(payload.get('trigger') or 'game_saved')
await self.on_notice(notice_trigger)
except asyncio.CancelledError:
pass
except Exception as error:
await await_log(self.logger.warning(f'Dashboard events listener stopped unexpectedly: {error}'))
+144
View File
@@ -0,0 +1,144 @@
from quart_common.web.logger import await_log, logging
from typing import Awaitable, Callable
import json
from .dashboard_ws_hub import DashboardWebSocketHub
from server.database import GameplayDatabase
class DashboardQueryService:
def __init__(self, gameplay_database:GameplayDatabase, ws_hub:DashboardWebSocketHub, logger:logging, dashboard_running_game_stale_sec:int):
self.gameplay_database = gameplay_database
self.ws_hub = ws_hub
self.logger = logger
self.dashboard_running_game_stale_sec = dashboard_running_game_stale_sec
self.publish_notice:Callable[[str], Awaitable[None]] | None = None
def set_publish_notice(self, publish_notice:Callable[[str], Awaitable[None]]) -> None:
self.publish_notice = publish_notice
async def on_dashboard_games_update_notice(self, trigger:str) -> None:
await self.push_dashboard_games_update(
game_state=None,
publish_cluster=False,
trigger=trigger,
)
async def build_dashboard_games_event(self, game_state:dict|None=None, trigger_override:str|None=None) -> dict:
games_payload = await self.get_dashboard_games(limit=100)
summary_payload = await self.get_dashboard_summary()
game_id = None
if game_state is not None:
game_id = game_state.get('game', {}).get('id')
trigger = trigger_override or ('game_saved' if game_id else 'snapshot')
return {
'type': 'dashboard_games_update',
'trigger': trigger,
'games': games_payload,
'summary': summary_payload,
}
async def build_dashboard_game_replay_event(self, game_id:str, request_id:str|None=None) -> dict:
replay_payload = await self.get_dashboard_game_replay(game_id)
if replay_payload is None:
return {
'type': 'dashboard_game_replay',
'request_id': request_id,
'game_id': game_id,
'error': 'game_not_found',
}
return {
'type': 'dashboard_game_replay',
'request_id': request_id,
'game_id': game_id,
'replay': replay_payload,
}
async def handle_dashboard_ws_request(self, payload_raw:object) -> dict|None:
if not isinstance(payload_raw, str):
return None
try:
payload = json.loads(payload_raw)
except json.JSONDecodeError:
return None
if not isinstance(payload, dict):
return None
if payload.get('type') != 'dashboard_game_replay_request':
return None
game_id = str(payload.get('game_id') or '').strip()
request_id_raw = payload.get('request_id')
request_id = None if request_id_raw is None else str(request_id_raw)
if game_id == '':
return {
'type': 'dashboard_game_replay',
'request_id': request_id,
'error': 'missing_game_id',
}
return await self.build_dashboard_game_replay_event(
game_id=game_id,
request_id=request_id,
)
async def push_dashboard_games_update(self, game_state:dict|None=None, publish_cluster:bool=True, trigger:str|None=None) -> None:
if self.gameplay_database is None:
return
event_payload = await self.build_dashboard_games_event(
game_state,
trigger_override=trigger,
)
await self.ws_hub.broadcast_payload(event_payload)
if publish_cluster and self.publish_notice is not None:
await self.publish_notice(str(event_payload.get('trigger') or ''))
async def get_dashboard_summary(self) -> dict:
if self.gameplay_database is None:
return {'enabled': False}
try:
await self._finalize_stale_dashboard_games()
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:
await self._finalize_stale_dashboard_games()
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}
async def _finalize_stale_dashboard_games(self) -> None:
if self.gameplay_database is None:
return
try:
await self.gameplay_database.finalize_stale_running_games(stale_after_seconds=self.dashboard_running_game_stale_sec)
except Exception as error:
await await_log(self.logger.warning(f'Gameplay DB stale running game finalize failed:{error}'))
+61
View File
@@ -0,0 +1,61 @@
import asyncio, json
class DashboardWebSocketHub:
def __init__(self):
self.subscribers:set[asyncio.Queue[str]] = set()
self.subscribers_lock = asyncio.Lock()
self.ws_tasks:set[asyncio.Task] = set()
self.ws_tasks_lock = asyncio.Lock()
self.shutdown_event = asyncio.Event()
self.shutdown_message = json.dumps({"type": "dashboard_ws_shutdown"})
async def register_subscriber(self, subscriber_queue:asyncio.Queue[str]) -> None:
async with self.subscribers_lock:
self.subscribers.add(subscriber_queue)
async def unregister_subscriber(self, subscriber_queue:asyncio.Queue[str]) -> None:
async with self.subscribers_lock:
self.subscribers.discard(subscriber_queue)
async def register_task(self, websocket_task:asyncio.Task) -> None:
async with self.ws_tasks_lock:
self.ws_tasks.add(websocket_task)
async def unregister_task(self, websocket_task:asyncio.Task) -> None:
async with self.ws_tasks_lock:
self.ws_tasks.discard(websocket_task)
async def broadcast_payload(self, payload:dict) -> None:
encoded_payload = json.dumps(payload)
async with self.subscribers_lock:
subscribers = tuple(self.subscribers)
for subscriber_queue in subscribers:
if subscriber_queue.full():
try:
subscriber_queue.get_nowait()
except asyncio.QueueEmpty:
pass
try:
subscriber_queue.put_nowait(encoded_payload)
except asyncio.QueueFull:
continue
def request_shutdown(self) -> None:
if self.shutdown_event.is_set():
return
self.shutdown_event.set()
for subscriber_queue in tuple(self.subscribers):
if subscriber_queue.full():
try:
subscriber_queue.get_nowait()
except asyncio.QueueEmpty:
pass
try:
subscriber_queue.put_nowait(self.shutdown_message)
except asyncio.QueueFull:
continue
+103
View File
@@ -0,0 +1,103 @@
from typing import cast
import time
from server.metrics import MetricsCollector
from server.GameBoard import GameBoard
from server.storage import StorageLoader
from snakes import SnakeBuilder
class GameRuntimeService:
def __init__(self, game_state_store:StorageLoader, snake_type:str, game_state_local_cache:bool, stale_game_timeout_sec:int):
self.game_state_store = game_state_store
self.snake_type = snake_type
self.game_state_local_cache = game_state_local_cache
self.stale_game_timeout_sec = stale_game_timeout_sec
self.metrics_collector = None
self.running_games: dict[str, GameBoard] = {}
self.game_move_counts: dict[str, int] = {}
self.game_last_seen_unix: dict[str, int] = {}
def attach_metrics_collector(self, metrics_collector:MetricsCollector) -> None:
self.metrics_collector = metrics_collector
async def create_game_board(self, game_state:dict, snake_builder:SnakeBuilder) -> GameBoard:
game_id = game_state['game']['id']
new_game_board = GameBoard(
game_id=game_id,
width=game_state['board']['width'],
height=game_state['board']['height'],
ruleset=game_state['game']['ruleset'],
source=game_state['game']['source'],
map=game_state['game']['map'],
snake_class=snake_builder.build(self.snake_type),
)
await new_game_board.start_game(game_state)
if self.game_state_local_cache:
self.running_games[game_id] = new_game_board
await self.game_state_store.save(game_id, new_game_board)
self.game_move_counts[game_id] = 0
self.game_last_seen_unix[game_id] = int(time.time())
if self.metrics_collector is not None:
await self.metrics_collector.record_game_started(len(self.game_last_seen_unix))
return new_game_board
async def persist_game_board(self, game_id:str, game_board:GameBoard) -> None:
if self.game_state_local_cache:
self.running_games[game_id] = game_board
await self.game_state_store.save(game_id, game_board)
async def delete_game_board(self, game_state:dict) -> None:
game_id = game_state['game']['id']
self.running_games.pop(game_id, None)
self.game_move_counts.pop(game_id, None)
self.game_last_seen_unix.pop(game_id, None)
await self.game_state_store.delete(game_id)
async def get_game_board(self, game_state:dict, snake_builder:SnakeBuilder, end:bool=False) -> GameBoard:
game_id = game_state['game']['id']
game_board: GameBoard
if self.game_state_local_cache and game_id in self.running_games:
game_board = self.running_games[game_id]
else:
persisted_board = await self.game_state_store.load(game_id)
if persisted_board is not None:
game_board = cast(GameBoard, persisted_board)
if self.game_state_local_cache:
self.running_games[game_id] = game_board
else:
game_board = await self.create_game_board(game_state, snake_builder)
if self.metrics_collector is not None:
await self.metrics_collector.record_game_autocreated()
if not end:
self.game_move_counts[game_id] = self.game_move_counts.get(game_id, 0) + 1
self.game_last_seen_unix[game_id] = int(time.time())
game_board.read_game_data(game_state)
if end:
game_board.end_game(game_state)
await self.persist_game_board(game_id, game_board)
return game_board
async def prune_stale_games(self) -> None:
if not self.game_last_seen_unix:
return
now = int(time.time())
stale_ids = [
game_id
for game_id, last_seen in self.game_last_seen_unix.items()
if now - last_seen >= self.stale_game_timeout_sec
]
for game_id in stale_ids:
self.running_games.pop(game_id, None)
self.game_move_counts.pop(game_id, None)
self.game_last_seen_unix.pop(game_id, None)
if self.metrics_collector is not None:
await self.metrics_collector.record_stuck_removed()
+50
View File
@@ -0,0 +1,50 @@
from quart_common.web.logger import await_log, logging
from server.database import GameplayDatabase
from server.GameBoard import GameBoard
class GameplayTrackingService:
def __init__(self, gameplay_database:GameplayDatabase, snake_type:str, snake_version:str, logger:logging):
self.gameplay_database = gameplay_database
self.snake_type = snake_type
self.snake_version = snake_version
self.logger = logger
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,
snake_type=self.snake_type,
snake_version=self.snake_version,
)
except Exception as error:
await await_log(self.logger.warning(f"Gameplay DB start record failed:{error}"))
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}"))
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
+11 -7
View File
@@ -1,5 +1,9 @@
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from server.GameBoard import GameBoard
from server.dataset.Dataset import Dataset
from server.GameBoard import GameBoard
from server.Files import save_file
import aiofiles
@@ -21,7 +25,7 @@ class LocalStorage:
self.dataset_compress_rotated = os.getenv("DATASET_COMPRESS_ROTATED", "true").strip().lower() in ("1", "true", "yes", "on")
self.dataset_max_bytes = int(float(os.getenv("DATASET_JSONL_MAX_MB", "50")) * 1024 * 1024)
def _get_active_dataset_path(self, game_board:GameBoard):
def _get_active_dataset_path(self, game_board:'GameBoard'):
if not self.dataset_rotate_daily:
return self.dataset_jsonl_path
@@ -60,7 +64,7 @@ class LocalStorage:
self._gzip_file(os.path.join(folder, name))
async def _rotate_if_needed(self, active_path:str, game_board:GameBoard):
async def _rotate_if_needed(self, active_path:str, game_board:'GameBoard'):
if self.dataset_max_bytes <= 0:
return
if not await aiofiles.os.path.exists(active_path):
@@ -81,7 +85,7 @@ class LocalStorage:
if self.dataset_compress_rotated:
self._gzip_file(rotated_path)
def _build_dataset_rows(self, dataset_payload:dict, game_board:GameBoard):
def _build_dataset_rows(self, dataset_payload:dict, game_board:'GameBoard'):
game_info = dataset_payload.get("game", {})
snake_info = dataset_payload.get("snake", {})
@@ -102,7 +106,7 @@ class LocalStorage:
})
return rows
async def _append_dataset_jsonl(self, dataset_payload:dict, game_board:GameBoard):
async def _append_dataset_jsonl(self, dataset_payload:dict, game_board:'GameBoard'):
rows = self._build_dataset_rows(dataset_payload, game_board)
if len(rows) == 0:
return
@@ -116,7 +120,7 @@ class LocalStorage:
for row in rows:
await f.write(json.dumps(row, ensure_ascii=False) + "\n")
def _get_correct_folder_for_save_file(self, game_board:GameBoard, file_name:str, game_type:str, leader_board:bool, winner:bool):
def _get_correct_folder_for_save_file(self, game_board:'GameBoard', file_name:str, game_type:str, leader_board:bool, winner:bool):
storage_folder = self.file_path
if leader_board:
storage_folder = os.path.join(storage_folder, "00_Leaderboards")
@@ -136,7 +140,7 @@ class LocalStorage:
return os.path.join(storage_folder, file_name)
async def save(self, game_board:GameBoard):
async def save(self, game_board:'GameBoard'):
game_type = game_board.get_type_of_game()
dataset = Dataset(game_board).build(only_good_moves=True)
+4 -1
View File
@@ -1,6 +1,9 @@
from .LocalStorage import LocalStorage
from .EdgeDB import EdgeDB
class StorageLoader:
@classmethod
def build(self, selected_storage: str):
def build(self, selected_storage:str) -> LocalStorage|EdgeDB:
storage_module = __import__(f"server.storage.{selected_storage}", fromlist=[selected_storage])
storage_class = getattr(storage_module, selected_storage)
return storage_class
+29 -16
View File
@@ -775,6 +775,20 @@
return String(value);
}
function formatObservedAtLocal(value) {
if (value === null || value === undefined || value === "") return "-";
const raw = String(value).trim();
const parsed = new Date(raw);
if (Number.isNaN(parsed.getTime())) {
return safeString(raw).slice(11, 19);
}
return parsed.toLocaleTimeString([], {
hour: "2-digit",
minute: "2-digit",
second: "2-digit",
});
}
function renderStats(summary) {
const finished = summary.finished_games || 0;
const wins = summary.wins || 0;
@@ -1295,27 +1309,12 @@
<div class="chip"><span class="k">Chosen Move</span><span class="v">${safeString(turn.my_move)}</span></div>
<div class="chip"><span class="k">Snake Type</span><span class="v">${snakeType}</span></div>
<div class="chip"><span class="k">Snake Version</span><span class="v">${snakeVersion}</span></div>
<div class="chip"><span class="k">Observed At</span><span class="v">${safeString(turn.observed_at).slice(11, 19)}</span></div>
<div class="chip"><span class="k">Observed At</span><span class="v">${formatObservedAtLocal(turn.observed_at)}</span></div>
<div class="chip"><span class="k">Food Count</span><span class="v">${Array.isArray(turn.food) ? turn.food.length : 0}</span></div>
<div class="chip"><span class="k">Hazard Count</span><span class="v">${Array.isArray(turn.hazards) ? turn.hazards.length : 0}</span></div>
</div>
<section>
<p class="section-title">Decision Summary</p>
<ul class="reason-list">${reasonList}</ul>
</section>
<section class="scores-section">
<p class="section-title">Move Scores</p>
<table class="score-table">
<colgroup><col style="width:50%"><col style="width:50%"></colgroup>
<thead><tr><th>Move</th><th>Score</th></tr></thead>
<tbody>${buildScoresRows(reasoning)}</tbody>
</table>
</section>
<section class="snakes-section">
<p class="section-title">Snake State This Turn</p>
<table class="score-table">
<colgroup>
<col style="width:32%">
@@ -1331,6 +1330,20 @@
</table>
</section>
<section class="scores-section">
<p class="section-title">Move Scores</p>
<table class="score-table">
<colgroup><col style="width:50%"><col style="width:50%"></colgroup>
<thead><tr><th>Move</th><th>Score</th></tr></thead>
<tbody>${buildScoresRows(reasoning)}</tbody>
</table>
</section>
<section>
<p class="section-title">Decision Summary</p>
<ul class="reason-list">${reasonList}</ul>
</section>
<section class="raw-block">
<p class="section-title">Raw Reasoning Payload</p>
<pre class="mono">${JSON.stringify(reasoning, null, 2)}</pre>
+4 -12
View File
@@ -4,6 +4,7 @@ from typing import Any, cast
from time import perf_counter
import os
from quart_common.web.env import env_int
from server.dataset.RLBootstrapDataset import RLBootstrapDataset
from snakes.TemplateSnake import TemplateSnake
@@ -43,9 +44,9 @@ class BestBattleSnake(TemplateSnake):
self.duel_style = self._get_duel_style()
self.timeout_buffer_ms = self._get_timeout_buffer_ms()
self.rl_bootstrap = RLBootstrapDataset()
self.future_planning_depth = max(1, min(4, self._env_int("BATTLE_FUTURE_PLANNING_DEPTH", default=2)))
self.future_planning_branch = max(1, min(3, self._env_int("BATTLE_FUTURE_PLANNING_BRANCH", default=2)))
self.future_planning_min_time_ms = max(25, self._env_int("BATTLE_FUTURE_PLANNING_MIN_MS", default=70))
self.future_planning_depth = max(1, min(4, env_int("BATTLE_FUTURE_PLANNING_DEPTH", default=2)))
self.future_planning_branch = max(1, min(3, env_int("BATTLE_FUTURE_PLANNING_BRANCH", default=2)))
self.future_planning_min_time_ms = max(25, env_int("BATTLE_FUTURE_PLANNING_MIN_MS", default=70))
def _get_duel_style(self) -> str:
"""Resolve duel tuning style from `BATTLE_SNAKE_DUEL_STYLE` or `DUEL_STYLE`."""
@@ -86,15 +87,6 @@ class BestBattleSnake(TemplateSnake):
except ValueError:
return 120
def _env_int(self, name:str, default:int) -> int:
value = os.getenv(name)
if value is None:
return default
try:
return int(value)
except ValueError:
return default
def choose_move(self, game_data:GameBoard) -> str:
"""Pick the next move from a Battlesnake move request.
+6 -2
View File
@@ -1,4 +1,8 @@
from server.GameBoard import GameBoard
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from server.GameBoard import GameBoard
import random
class TemplateSnake:
@@ -22,7 +26,7 @@ class TemplateSnake:
def add_calculations(self, calculations:dict):
self.calculations.append(calculations)
def choose_move(self, game_data:GameBoard):
def choose_move(self, game_data:'GameBoard'):
self.game_board = game_data
self.calculations = []
self.eat_the_snake_overwrite = False
+6 -10
View File
@@ -4,6 +4,8 @@ from typing import Any, cast
from time import perf_counter
import heapq, os
from quart_common.web.env import env_int
from snakes.TemplateSnake import TemplateSnake
from server.GameBoard import GameBoard
from server.dataset.RLBootstrapDataset import RLBootstrapDataset
@@ -92,9 +94,9 @@ class UltimateBattleSnake(TemplateSnake):
self._bfs_cache: dict[tuple, int] = {}
self._bfs_cache_turn: int = -1
# Config
self._planning_depth = max(1, min(4, self._env_int("BATTLE_FUTURE_PLANNING_DEPTH", 2)))
self._planning_branch = max(1, min(3, self._env_int("BATTLE_FUTURE_PLANNING_BRANCH", 2)))
self._planning_min_ms = max(25, self._env_int("BATTLE_FUTURE_PLANNING_MIN_MS", 70))
self._planning_depth = max(1, min(4, env_int("BATTLE_FUTURE_PLANNING_DEPTH", 2)))
self._planning_branch = max(1, min(3, env_int("BATTLE_FUTURE_PLANNING_BRANCH", 2)))
self._planning_min_ms = max(25, env_int("BATTLE_FUTURE_PLANNING_MIN_MS", 70))
# RL bootstrap dataset recorder
self.rl_bootstrap = RLBootstrapDataset()
@@ -106,12 +108,6 @@ class UltimateBattleSnake(TemplateSnake):
except ValueError:
return 130
def _env_int(self, name: str, default: int) -> int:
try:
return int(os.getenv(name, str(default)))
except ValueError:
return default
def _get_duel_style(self) -> str:
raw = os.getenv("BATTLE_SNAKE_DUEL_STYLE", os.getenv("DUEL_STYLE", "balanced"))
style = raw.strip().lower()
@@ -138,7 +134,7 @@ class UltimateBattleSnake(TemplateSnake):
self.game_board = game_data
self.calculations = []
timeout_ms = game_data.get_timeout() if hasattr(game_data, "get_timeout") else 500
timeout_ms = (game_data.get_timeout() if hasattr(game_data, "get_timeout") else 500)
deadline = perf_counter() + (max(50, timeout_ms - self._get_timeout_buffer_ms()) / 1000.0)
game_id = getattr(game_data, "id", None)