snake-python/server/Server.py

from server.Files import read_file
from server.GameBoard import GameBoard
from server.SnakeBuilder import SnakeBuilder

from server.storage.StorageLoader import StorageLoader

from quart import Quart, request, jsonify
import logging, json, os, re

class Server:
  default_snake_config = {"apiversion":"1","author":"","color":"#888888","head":"default","tail":"default"}

  def __init__(self, data_path:str, snake_type:str, storage_type:str, debug:bool=False, store_game_when_win_and_moves_are_bigger_as:int=10, check_tls_security:bool=False):
    self.debug = debug
    self.snake_type = snake_type
    self.storage_type = storage_type

    self.config_file = os.path.join(data_path, 'data', 'snake-config.json')
    self.data_path = data_path
    self.check_tls_security = check_tls_security

    self.store_game_state = False
    self.store_game_when_win_and_moves_are_bigger_as = store_game_when_win_and_moves_are_bigger_as

    self.running_games:dict[str, GameBoard] = {}

    self.app = Quart("Battlesnake")

    # 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():
      snake_config = await self._read_json_config_or_create()

      print("INFO Snake:", snake_config)
      return snake_config

    # start is called when your Battlesnake begins a game
    @self.app.post("/start")
    async def on_start():
      game_state = await request.get_json()
      await self._create_game_board(game_state)
      print("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():
      game_state = await request.get_json()
      game_board = await self._get_game_board(game_state)
      next_move = game_board.snake_neat_make_a_move()

      if self.debug:
        print("TURN:", f'{game_state["turn"]:3},', "MOVE:", f"{next_move:5}")

      return {"move": next_move}

    # end is called when your Battlesnake finishes a game
    @self.app.post("/end")
    async def on_end():
      game_state = await request.get_json()
      if self.store_game_state:
        game_board = await self._get_game_board(game_state, end=True)
        #if not game_board.get_winner() == "me" and not game_board.get_turn() <= self.store_game_when_win_and_moves_are_bigger_as:
        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),
          )

      print("GAME ENDED: Winner is", [ x["name"] for x in game_state["board"]['snakes']])
      self._delete_game_board(game_state)
      return "ok"

    @self.app.after_request
    async def identify_server(response):
      response.headers.set(
        "server", "battlesnake/gitea/snake-python"
      )
      return response

    @self.app.get("/cleanup")
    async def cleanup():
      results = self._cleanup_database()
      return jsonify(data=json.loads(results), status=200)

  def run(self, host:str="0.0.0.0", port:str="8000", debug:bool=False):
    logging.getLogger("werkzeug").setLevel(logging.ERROR)

    print(f"\nRunning Battlesnake at http://{host}:{port} with the {' '.join(re.findall('[A-Z][^A-Z]*', self.snake_type))}")
    self.app.run(host=host, port=port, debug=debug)

  async def _read_json_config_or_create(self):
    snake_config = await read_file(self.config_file, json.load)
    if not snake_config:
      return await self._override_snake_config_with_environment_variables(self.default_snake_config)
    return await self._override_snake_config_with_environment_variables(snake_config)

  async def _override_snake_config_with_environment_variables(self, config: dict[str, str]) -> dict[str, str]:
    for key in ("author", "color", "head", "tail"):
      value = os.environ.get(f"SNAKE_{key.upper()}")
      if value is not None:
        config[key] = value
    return config

  async def _create_game_board(self, game_state:dict):
    new_game_board = GameBoard(
      game_id=game_state["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)

    self.running_games[game_state["game"]["id"]] = new_game_board
    return new_game_board

  def _delete_game_board(self, game_state):
    del self.running_games[game_state["game"]["id"]]

  async def _get_game_board(self, game_state:str, end:bool=False):
    try:
      game_board = self.running_games[game_state["game"]["id"]]
    except KeyError:
      game_board = await self._create_game_board(game_state)

    game_board.read_game_data(game_state)
    if end:
      game_board.end_game(game_state)

    return game_board

  def enable_store_game_state(self):
    self.store_game_state = True

  def _cleanup_database(self):
    storage = StorageLoader.build(self.storage_type)()
    return storage.cleanup()