diff --git a/main.py b/main.py index 602a28f..272ae5a 100755 --- a/main.py +++ b/main.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python3 +#!/usr/bin/env -S uv run --script # Welcome to # __________ __ __ .__ __ diff --git a/server/Files.py b/server/Files.py index 22d9c2f..ffa1bbb 100644 --- a/server/Files.py +++ b/server/Files.py @@ -1,16 +1,24 @@ +import aiofiles.os +import aiofiles import os -def read_file(path, callback=None): - if os.path.exists(path): - with open(path, 'r') as f: - data = callback(f) - return data - else: +async def read_file(path: str, callback=None): + if not await aiofiles.os.path.exists(path): return None -def save_file(path, data, callback=None, *args, **kwargs): - if not os.path.exists(path): - os.makedirs(os.path.dirname(path), exist_ok=True) + async with aiofiles.open(path, "r") as f: + if callback: + return await callback(f) + return await f.read() - with open(path, 'w') as f: - callback(data, f, *args, **kwargs) + +async def save_file(path: str, data, callback=None, *args, **kwargs): + dir_path = os.path.dirname(path) + if dir_path: + await aiofiles.os.makedirs(dir_path, exist_ok=True) + + async with aiofiles.open(path, "w") as f: + if callback: + await callback(data, f, *args, **kwargs) + else: + await f.write(data) diff --git a/server/GameBoard.py b/server/GameBoard.py index f0b6c25..48fed3d 100644 --- a/server/GameBoard.py +++ b/server/GameBoard.py @@ -92,7 +92,7 @@ class GameBoard: self._set_turn(game_data["turn"]) - def start_game(self, game_data:dict): + async def start_game(self, game_data:dict): self.init_snakes = len(game_data['board']['snakes']) def end_game(self, game_data:dict): @@ -134,7 +134,7 @@ class GameBoard: return {"name": self.type, "is_ladder": self.is_ladder} - def save(self, store_class, **kwargs): + async def save(self, store_class, **kwargs): store = store_class(**kwargs) - store.save(self) + await store.save(self) del store diff --git a/server/Server.py b/server/Server.py index 673da51..a58ca36 100644 --- a/server/Server.py +++ b/server/Server.py @@ -4,8 +4,7 @@ from server.SnakeBuilder import SnakeBuilder from server.storage.StorageLoader import StorageLoader -from flask import Flask, jsonify -from flask import request +from quart import Quart, request, jsonify import logging, json, os, re class Server: @@ -25,33 +24,67 @@ class Server: self.running_games:dict[str, GameBoard] = {} - self.app = Flask("Battlesnake") + 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(): - return self._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 = request.get_json() - self._start(game_state) + 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 = request.get_json() - return self._move(game_state) + 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 = request.get_json() - self._end(game_state) + 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/github/starter-snake-python" + "server", "battlesnake/gitea/snake-python" ) return response @@ -66,20 +99,20 @@ class Server: 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) - def _read_json_config_or_create(self): - snake_config = read_file(self.config_file, json.load) + async def _read_json_config_or_create(self): + snake_config = await read_file(self.config_file, json.load) if not snake_config: - snake_config = self._override_snake_config_with_environment_variables(self.default_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) - return self._override_snake_config_with_environment_variables(snake_config) - - def _override_snake_config_with_environment_variables(self, config:dict[str]): - for key in ["author", "color", "head", "tail"]: - if os.environ.get(f"SNAKE_{key.upper()}", None): - config[key.lower()] = os.environ.get(f"SNAKE_{key.upper()}") + 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 - def _create_game_board(self, game_state:dict): + async def _create_game_board(self, game_state:dict): new_game_board = GameBoard( game_id=game_state["game"]["id"], width=game_state['board']['width'], @@ -89,7 +122,7 @@ class Server: map=game_state['game']['map'], snake_class=SnakeBuilder.build(self.snake_type) ) - new_game_board.start_game(game_state) + await new_game_board.start_game(game_state) self.running_games[game_state["game"]["id"]] = new_game_board return new_game_board @@ -97,11 +130,11 @@ class Server: def _delete_game_board(self, game_state): del self.running_games[game_state["game"]["id"]] - def _get_game_board(self, game_state:str, end:bool=False): + 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 = self._create_game_board(game_state) + game_board = await self._create_game_board(game_state) game_board.read_game_data(game_state) if end: @@ -112,52 +145,6 @@ class Server: def enable_store_game_state(self): self.store_game_state = True - # 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 - def _info(self) -> dict: - snake_config = self._read_json_config_or_create() - - print("INFO Snake:", snake_config) - return snake_config - - # start is called when your Battlesnake begins a game - def _start(self, game_state:dict): - self._create_game_board(game_state) - print("GAME START:", game_state["game"]) - - # move is called when your Battlesnake game is running game - def _move(self, game_state:dict) -> dict: - game_board = 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 - def _end(self, game_state:dict): - if self.store_game_state: - game_board = 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: - 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: - 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) - def _cleanup_database(self): storage = StorageLoader.build(self.storage_type)() return storage.cleanup() diff --git a/server/storage/EdgeDB.py b/server/storage/EdgeDB.py index 3a884b2..e656c52 100644 --- a/server/storage/EdgeDB.py +++ b/server/storage/EdgeDB.py @@ -45,7 +45,7 @@ class EdgeDB: return data - def insert(self, game_board:GameBoard): + async def insert(self, game_board:GameBoard): game_type = game_board.get_type_of_game() self.run_query_with_reconnection( @@ -104,8 +104,8 @@ class EdgeDB: snake_type=game_board.snake_class.__class__.__name__, ) - def save(self, game_board:GameBoard): - self.insert(game_board) + async def save(self, game_board:GameBoard): + await self.insert(game_board) def __del__(self): self.client.close() diff --git a/server/storage/LocalStorage.py b/server/storage/LocalStorage.py index 0d93eee..88ee4fd 100644 --- a/server/storage/LocalStorage.py +++ b/server/storage/LocalStorage.py @@ -28,7 +28,7 @@ class LocalStorage: return os.path.join(storage_folder, file_name) - def save(self, game_board:GameBoard): + async def save(self, game_board:GameBoard): game_type = game_board.get_type_of_game() save_file_path = self._get_correct_folder_for_save_file( game_board, @@ -38,7 +38,7 @@ class LocalStorage: True if game_board.winner_snake_names and "me" in game_board.winner_snake_names else False ) - save_file(save_file_path, { + await save_file(save_file_path, { "winner": game_board.winner_snake_names, "game": { "url": game_board.url,