From 72ea0126f796eccfb0c4f088478c200c92fe16dc Mon Sep 17 00:00:00 2001 From: shillerben Date: Sat, 9 Sep 2023 20:45:24 -0500 Subject: [PATCH] Switch to FastAPI --- Dockerfile | 16 +-- requirements.txt | 27 +++- server.py | 63 --------- src/__init__.py | 0 src/api.py | 166 ++++++++++++++++++++++++ src/server.py | 45 +++++++ {snakes => src/snakes}/__init__.py | 0 {snakes => src/snakes}/snake.py | 10 +- {snakes => src/snakes}/starter_snake.py | 50 +++---- 9 files changed, 275 insertions(+), 102 deletions(-) delete mode 100644 server.py create mode 100644 src/__init__.py create mode 100644 src/api.py create mode 100644 src/server.py rename {snakes => src/snakes}/__init__.py (100%) rename {snakes => src/snakes}/snake.py (77%) rename {snakes => src/snakes}/starter_snake.py (53%) diff --git a/Dockerfile b/Dockerfile index 01fe337..fb52670 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,16 +1,14 @@ FROM python:3.11.4-slim -# FROM continuumio/conda-ci-linux-64-python3.8 -# FROM continuumio/miniconda3 -# Install app -COPY . /app WORKDIR /app +COPY . . -# Install dependencies RUN pip install --upgrade pip && pip install -r requirements.txt -# RUN conda env create -f environment.yaml -# ENV CONDA_DEFAULT_ENV battlesnake + +ENV HOST 0.0.0.0 +ENV PORT 8000 +ENV ROOT_PATH / +ENV SNAKE starter_snake # Run Battlesnake -ENTRYPOINT [ "python", "server.py" ] -CMD [ "starter_snake" ] \ No newline at end of file +ENTRYPOINT uvicorn src.server:app --host $HOST --port $PORT --root-path $ROOT_PATH \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 31fcf80..a784cc4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,10 +1,33 @@ +annotated-types==0.5.0 +anyio==3.7.1 blinker==1.6.2 click==8.1.7 +fastapi==0.103.1 Flask==2.3.3 +h11==0.14.0 +h2==4.1.0 +hpack==4.0.0 +httptools==0.6.0 +hyperframe==6.0.1 +idna==3.4 itsdangerous==2.1.2 Jinja2==3.1.2 MarkupSafe==2.1.3 -mypy==1.5.1 -mypy-extensions==1.0.0 +marshmallow==3.20.1 +packaging==23.1 +priority==2.0.0 +pydantic==2.3.0 +pydantic-extra-types==2.1.0 +pydantic_core==2.6.3 +python-dotenv==1.0.0 +PyYAML==6.0.1 +sniffio==1.3.0 +starlette==0.27.0 +typing-inspect==0.9.0 typing_extensions==4.7.1 +uvicorn==0.23.2 +uvloop==0.17.0 +watchfiles==0.20.0 +websockets==11.0.3 Werkzeug==2.3.7 +wsproto==1.2.0 diff --git a/server.py b/server.py deleted file mode 100644 index 43b3a1b..0000000 --- a/server.py +++ /dev/null @@ -1,63 +0,0 @@ -import argparse -import logging -import os -import typing - -from flask import Flask -from flask import request - -from snakes.snake import Snake - - -def run_server(snake: Snake, host: str, port: int): - path = snake.name - app = Flask("Battlesnake") - - @app.get(f"/{path}/") - def on_info(): - return snake.info() - - @app.post(f"/{path}/start") - def on_start(): - game_state = request.get_json() - snake.start(game_state) - return "ok" - - @app.post(f"/{path}/move") - def on_move(): - game_state = request.get_json() - return snake.move(game_state) - - @app.post(f"/{path}/end") - def on_end(): - game_state = request.get_json() - snake.end(game_state) - return "ok" - - @app.after_request - def identify_server(response): - response.headers.set( - "server", "shillerben/gitea/starter-snake-python" - ) - return response - - logging.getLogger("werkzeug").setLevel(logging.ERROR) - - print(f"\nRunning Battlesnake at http://{host}:{port}/{path}") - app.run(host=host, port=port) - - -if __name__ == "__main__": - parser = argparse.ArgumentParser() - parser.add_argument("-b", "--bind", default="0.0.0.0", help="host ip to listen on") - parser.add_argument("-p", "--port", type=int, default=8000, help="host port to listen on") - parser.add_argument("snake", help="snake to run") - args = parser.parse_args() - - host = args.bind - port = args.port - snake_name = args.snake - - snake = Snake.SNAKES[snake_name]() - - run_server(snake, host, port) diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/api.py b/src/api.py new file mode 100644 index 0000000..758eb39 --- /dev/null +++ b/src/api.py @@ -0,0 +1,166 @@ +from typing import Union, List +from enum import StrEnum, auto +from pydantic import BaseModel, PositiveInt, NonNegativeInt, conint, constr + + +''' +Enums +''' + +class GameMode(StrEnum): + EMPTY_STR = "" + STANDARD = auto() + SOLO = auto() + ROYALE = auto() + SQUAD = auto() + CONSTRICTOR = auto() + WRAPPED = auto() + +class Source(StrEnum): + TOURNAMENT = auto() + LEAGUE = auto() + ARENA = auto() + CHALLENGE = auto() + CUSTOM = auto() + +class GameMap(StrEnum): + EMPTY_STR = "" + STANDARD = auto() + EMPTY = auto() + ARCADE_MAZE = auto() + ROYALE = auto() + SOLO_MAZE = auto() + HZ_INNER_WALL = auto() + HZ_RINGS = auto() + HZ_COLUMNS = auto() + HZ_RIVERS_BRIDGES = auto() + HZ_SPIRAL = auto() + HZ_SCATTER = auto() + HZ_GROW_BOX = auto() + HZ_EXPAND_BOX = auto() + HZ_EXPAND_SCATTER = auto() + +class GameMove(StrEnum): + UP = auto() + DOWN = auto() + LEFT = auto() + RIGHT = auto() + + +''' +Models +''' + +class RoyaleRulesetSettings(BaseModel): + shrinkEveryNTurns: NonNegativeInt + +class SquadRulesetSettings(BaseModel): + allowBodyCollisions: bool + sharedElimination: bool + sharedHealth: bool + sharedLength: bool + +class RulesetSettings(BaseModel): + foodSpawnChance: conint(ge=0, le=100, strict=True) + minimumFood: NonNegativeInt + hazardDamagePerTurn: NonNegativeInt + royale: RoyaleRulesetSettings + squad: SquadRulesetSettings + +class Ruleset(BaseModel): + name: GameMode + version: str + settings: RulesetSettings + + +class Coordinate(BaseModel): + x: NonNegativeInt + y: NonNegativeInt + + +class BattlesnakeCustomizations(BaseModel): + color: str + head: str + tail: str + + +class Battlesnake(BaseModel): + id: str + name: str + health: conint(ge=0, le=100, strict=True) + body: List[Coordinate] + latency: str # previous latency in milliseconds, str(int) + head: Coordinate + length: PositiveInt + shout: constr(max_length=256) + squad: str + customizations: BattlesnakeCustomizations + + + +class Board(BaseModel): + height: PositiveInt + width: PositiveInt + food: List[Coordinate] + hazards: List[Coordinate] + snakes: List[Battlesnake] + + +class Game(BaseModel): + id: str + ruleset: Ruleset + map: GameMap + timeout: NonNegativeInt # milliseconds + source: Source + + +''' +GET / +''' +class GetResponse(BaseModel): + apiversion: str + author: Union[str, None] = None + color: Union[str, None] = None + head: Union[str, None] = None + tail: Union[str, None] = None + version: Union[str, None] = None + + +''' +POST /start +''' +class StartRequest(BaseModel): + game: Game + turn: NonNegativeInt + board: Board + you: Battlesnake + +class StartResponse(BaseModel): + pass + + +''' +POST /move +''' +class MoveRequest(BaseModel): + game: Game + turn: NonNegativeInt + board: Board + you: Battlesnake + +class MoveResponse(BaseModel): + move: GameMove + shout: constr(max_length=256) = "" + + +''' +POST /end +''' +class EndRequest(BaseModel): + game: Game + turn: NonNegativeInt + board: Board + you: Battlesnake + +class EndResponse(BaseModel): + pass \ No newline at end of file diff --git a/src/server.py b/src/server.py new file mode 100644 index 0000000..b94898e --- /dev/null +++ b/src/server.py @@ -0,0 +1,45 @@ +import logging +import os + +from fastapi import FastAPI, Request + +from .snakes.snake import Snake +from . import api + + +snake_name = os.getenv("SNAKE") + +if snake_name is None: + print("SNAKE env variable is not set!") + exit(1) + +snake = Snake.SNAKES[snake_name]() + +app = FastAPI(title=snake.name, + summary=f"API for {snake.name}") + +@app.get("/") +def on_info() -> api.GetResponse: + print("INFO") + return snake.info() + +@app.post("/start") +def on_start(request: api.StartRequest) -> api.StartResponse: + return snake.start(request) + +@app.post("/move") +def on_move(request: api.MoveRequest) -> api.MoveResponse: + return snake.move(request) + +@app.post("/end") +def on_end(request: api.EndRequest) -> api.EndResponse: + return snake.end(request) + + +@app.middleware("http") +async def identify_server(request: Request, call_next): + response = await call_next(request) + response.headers["server"] = "shillerben/gitea/starter-snake-python" + return response + +logging.getLogger("werkzeug").setLevel(logging.ERROR) diff --git a/snakes/__init__.py b/src/snakes/__init__.py similarity index 100% rename from snakes/__init__.py rename to src/snakes/__init__.py diff --git a/snakes/snake.py b/src/snakes/snake.py similarity index 77% rename from snakes/snake.py rename to src/snakes/snake.py index 3939d6c..f45e4ac 100644 --- a/snakes/snake.py +++ b/src/snakes/snake.py @@ -1,5 +1,7 @@ import typing +from .. import api + class Snake: @@ -9,7 +11,7 @@ class Snake: def __init_subclass__(cls) -> None: cls.SNAKES[cls.name] = cls - def info(self) -> typing.Dict: + def info(self) -> api.GetResponse: '''info is called when you create your Battlesnake on play.battlesnake.com and controls your Battlesnake's appearance @@ -17,15 +19,15 @@ class Snake: ''' raise NotImplementedError() - def start(self, game_state: typing.Dict) -> None: + def start(self, game_state: api.StartRequest) -> api.StartResponse: '''start is called when your Battlesnake begins a game''' raise NotImplementedError() - def end(self, game_state: typing.Dict) -> None: + def end(self, game_state: api.EndRequest) -> api.EndResponse: '''end is called when your Battlesnake finishes a game''' raise NotImplementedError() - def move(self, game_state: typing.Dict) -> typing.Dict: + def move(self, game_state: api.MoveRequest) -> api.MoveResponse: '''move is called on every turn and returns your next move Valid moves are "up", "down", "left", or "right" See https://docs.battlesnake.com/api/example-move for available data diff --git a/snakes/starter_snake.py b/src/snakes/starter_snake.py similarity index 53% rename from snakes/starter_snake.py rename to src/snakes/starter_snake.py index b00430a..2e68dd5 100644 --- a/snakes/starter_snake.py +++ b/src/snakes/starter_snake.py @@ -1,6 +1,7 @@ import random -import typing + +from .. import api from .snake import Snake @@ -10,40 +11,41 @@ class StarterSnake(Snake): name = "starter_snake" - def info(self) -> typing.Dict: + def info(self) -> api.GetResponse: print("INFO") + return api.GetResponse( + apiversion="1", + author="shillerben", + color="#BBDEFB", + head="default", + tail="default", + ) - return { - "apiversion": "1", - "author": "", # TODO: Your Battlesnake Username - "color": "#888888", # TODO: Choose color - "head": "default", # TODO: Choose head - "tail": "default", # TODO: Choose tail - } - - def start(self, game_state: typing.Dict): + def start(self, game_state: api.StartRequest) -> api.StartResponse: print("GAME START") + return api.StartResponse() - def end(self, game_state: typing.Dict): - print("GAME OVER\n") + def end(self, game_state: api.EndRequest) -> api.EndResponse: + print("GAME OVER") + return api.EndResponse() - def move(self, game_state: typing.Dict) -> typing.Dict: + def move(self, game_state: api.MoveRequest) -> api.MoveResponse: is_move_safe = {"up": True, "down": True, "left": True, "right": True} # We've included code to prevent your Battlesnake from moving backwards - my_head = game_state["you"]["body"][0] # Coordinates of your head - my_neck = game_state["you"]["body"][1] # Coordinates of your "neck" + my_head = game_state.you.body[0] # Coordinates of your head + my_neck = game_state.you.body[1] # Coordinates of your "neck" - if my_neck["x"] < my_head["x"]: # Neck is left of head, don't move left + if my_neck.x < my_head.x: # Neck is left of head, don't move left is_move_safe["left"] = False - elif my_neck["x"] > my_head["x"]: # Neck is right of head, don't move right + elif my_neck.x > my_head.x: # Neck is right of head, don't move right is_move_safe["right"] = False - elif my_neck["y"] < my_head["y"]: # Neck is below head, don't move down + elif my_neck.y < my_head.y: # Neck is below head, don't move down is_move_safe["down"] = False - elif my_neck["y"] > my_head["y"]: # Neck is above head, don't move up + elif my_neck.y > my_head.y: # Neck is above head, don't move up is_move_safe["up"] = False # TODO: Step 1 - Prevent your Battlesnake from moving out of bounds @@ -63,8 +65,8 @@ class StarterSnake(Snake): safe_moves.append(move) if len(safe_moves) == 0: - print(f"MOVE {game_state['turn']}: No safe moves detected! Moving down") - return {"move": "down"} + print(f"MOVE {game_state.turn}: No safe moves detected! Moving down") + return api.MoveResponse(move=api.GameMove.UP) # Choose a random move from the safe ones next_move = random.choice(safe_moves) @@ -72,5 +74,5 @@ class StarterSnake(Snake): # TODO: Step 4 - Move towards food instead of random, to regain health and survive longer # food = game_state['board']['food'] - print(f"MOVE {game_state['turn']}: {next_move}") - return {"move": next_move} + print(f"MOVE {game_state.turn}: {next_move}") + return api.MoveResponse(move=api.GameMove(next_move))