51 lines
1.8 KiB
Python
51 lines
1.8 KiB
Python
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, logger:logging):
|
|
self.gameplay_database = gameplay_database
|
|
self.logger = logger
|
|
|
|
async def record_gameplay_start(self, game_state:dict, game_board:GameBoard) -> None:
|
|
snake_name, snake_version = game_board.get_snake_name_and_version()
|
|
|
|
if self.gameplay_database is None:
|
|
return
|
|
try:
|
|
await self.gameplay_database.record_game_start(
|
|
game_state,
|
|
snake_type=snake_name,
|
|
snake_version=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
|