Project rename and restructure

hopper/api/dependencies.py

@@ -0,0 +1,13 @@
+from hopper.engine import GameEngine, GameEngineFactory
+
+game_engine: GameEngine
+
+
+def create_game_engine() -> GameEngine:
+    global game_engine
+    game_engine = GameEngineFactory.create_default()
+    return game_engine
+
+
+def get_game_engine() -> GameEngine:
+    return game_engine

hopper/api/dto.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+from pydantic import BaseModel as PydanticBaseModel
+
+from hopper.models.board import GameBoard
+from hopper.models.player import Player, Position
+
+
+class BaseModel(PydanticBaseModel):
+    class Config:
+        orm_mode = True
+
+
+class PingResponse(BaseModel):
+    message: str
+
+
+class BoardDto(BaseModel):
+    width: int
+    height: int
+
+    @staticmethod
+    def from_model(board: GameBoard) -> BoardDto:
+        return BoardDto.from_orm(board)
+
+
+class PositionDto(BaseModel):
+    x: int
+    y: int
+
+    @staticmethod
+    def from_model(position: Position) -> PositionDto:
+        return PositionDto.from_orm(position)
+
+
+class PlayerDto(BaseModel):
+    uuid: str
+    position: PositionDto
+    move_count: int
+    move_attempt_count: int
+
+    @staticmethod
+    def from_model(player: Player) -> PlayerDto:
+        return PlayerDto.from_orm(player)
+
+
+class DestinationDto(BaseModel):
+    position: PositionDto
+
+
+class StartGameRequestDto(BaseModel):
+    player_name: str
+
+
+class GameInfoDto(BaseModel):
+    board: BoardDto
+    destination: DestinationDto
+
+
+class StartGameResponseDto(GameInfoDto):
+    player: PlayerDto
+
+
+class MovePlayerResponseDto(BaseModel):
+    player: PlayerDto
+
+
+class PlayerInfoResponseDto(MovePlayerResponseDto):
+    ...
+
+
+class ErrorResponseDto(BaseModel):
+    detail: str

hopper/api/views.py

@@ -0,0 +1,132 @@
+from fastapi import APIRouter, Depends, HTTPException, Response
+from starlette import status
+
+from hopper.api.dependencies import get_game_engine
+from hopper.api.dto import (
+    BoardDto,
+    DestinationDto,
+    ErrorResponseDto,
+    GameInfoDto,
+    MovePlayerResponseDto,
+    PingResponse,
+    PlayerDto,
+    PlayerInfoResponseDto,
+    PositionDto,
+    StartGameRequestDto,
+    StartGameResponseDto,
+)
+from hopper.engine import GameEngine
+from hopper.enums import Direction, PlayerMoveResult
+from hopper.errors import Collision, PositionOutOfBounds
+
+router = APIRouter()
+
+
+@router.get("/ping", response_model=PingResponse)
+async def ping() -> PingResponse:
+    return PingResponse(
+        message="Pong!",
+    )
+
+
+@router.get("/game", response_model=GameInfoDto)
+async def get_game_info(
+    engine: GameEngine = Depends(get_game_engine),
+) -> GameInfoDto:
+    return GameInfoDto(
+        board=BoardDto.from_model(engine.board),
+        destination=DestinationDto(
+            position=PositionDto.from_model(engine.board.destination.position)
+        ),
+    )
+
+
+@router.post("/game", response_model=StartGameResponseDto)
+async def start_game(
+    body: StartGameRequestDto,
+    engine: GameEngine = Depends(get_game_engine),
+) -> StartGameResponseDto:
+    new_player = engine.start_game(player_name=body.player_name)
+
+    return StartGameResponseDto(
+        board=BoardDto.from_model(engine.board),
+        player=PlayerDto.from_model(new_player),
+        destination=DestinationDto(
+            position=PositionDto.from_model(engine.board.destination.position)
+        ),
+    )
+
+
+@router.get(
+    "/player/{uuid}",
+    response_model=PlayerInfoResponseDto,
+    status_code=status.HTTP_201_CREATED,
+)
+async def get_player_info(
+    uuid: str,
+    engine: GameEngine = Depends(get_game_engine),
+) -> MovePlayerResponseDto:
+    player = engine.players.find(uuid)
+    if player is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Player not found"
+        )
+    if not player.active:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Player kicked out due to inactivity",
+        )
+    return PlayerInfoResponseDto(player=PlayerDto.from_model(player))
+
+
+@router.post(
+    "/player/{uuid}/move/{direction}",
+    response_model=MovePlayerResponseDto,
+    status_code=status.HTTP_201_CREATED,
+    responses={
+        status.HTTP_200_OK: {
+            "model": MovePlayerResponseDto,
+            "description": "Destination reached!",
+        },
+        status.HTTP_403_FORBIDDEN: {
+            "model": ErrorResponseDto,
+            "description": " Player uuid not valid, probably due to inactivity",
+        },
+        status.HTTP_409_CONFLICT: {
+            "model": ErrorResponseDto,
+            "description": " Position out of bounds or collision with an object",
+        },
+    },
+)
+async def move_player(
+    uuid: str,
+    direction: Direction,
+    response: Response,
+    engine: GameEngine = Depends(get_game_engine),
+) -> MovePlayerResponseDto:
+    player = engine.players.find(uuid)
+    if player is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Player not found"
+        )
+    if not player.active:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Player kicked out due to inactivity",
+        )
+
+    try:
+        move_result = engine.move_player(player, direction)
+    except PositionOutOfBounds:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Position out of bounds"
+        )
+    except Collision:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Collision with an object"
+        )
+
+    if move_result == PlayerMoveResult.DESTINATION_REACHED:
+        response.status_code = status.HTTP_200_OK
+
+    return MovePlayerResponseDto(player=PlayerDto.from_model(player))

hopper/api_tests/requests.http

@@ -0,0 +1,34 @@
+GET http://localhost:8010/ping
+###
+
+# create new game
+POST http://localhost:8010/game
+
+{
+  "player_name": "Mirko"
+}
+###
+
+# get game info
+GET http://localhost:8010/game
+###
+
+# get player info
+GET http://localhost:8010/player/test-player-id
+###
+
+# move player left
+POST http://localhost:8010/player/test-player-id/move/left
+###
+
+# move player right
+POST http://localhost:8010/player/test-player-id/move/right
+###
+
+# move player up
+POST http://localhost:8010/player/test-player-id/move/up
+###
+
+# move player down
+POST http://localhost:8010/player/test-player-id/move/down
+###

hopper/engine.py

@@ -0,0 +1,153 @@
+import logging
+
+from hopper.enums import Direction, PlayerMoveResult
+from hopper.errors import Collision, PositionOutOfBounds
+from hopper.models.board import (
+    BOARD_DUMP_CHARS,
+    Destination,
+    GameBoard,
+    Layer,
+    LayerObject,
+    ObjectType,
+    create_random_position,
+)
+from hopper.models.player import Player, PlayerList, Position
+from hopper.watchdog import InactivityWatchdog
+from settings import settings
+
+
+class GameEngine:
+    def __init__(self, board: GameBoard) -> None:
+        self.board = board
+        self.players = PlayerList()
+        self._inacivity_watchdog = None
+        self.__debug_print_board()
+
+    def dump_board(self) -> list[list[str]]:
+        dump = self.board.dump()
+
+        for player in self.players:
+            dump[player.position.y][player.position.x] = BOARD_DUMP_CHARS[
+                ObjectType.PLAYER
+            ]
+
+        return dump
+
+    def __debug_print_board(self):
+        if not (settings.debug and settings.debug.PRINT_BOARD):
+            return
+        for line in self.dump_board():
+            print(" ".join(line))
+
+    def _start_inactivity_watchdog(self) -> None:
+        if not self._inacivity_watchdog:
+            self._inacivity_watchdog = InactivityWatchdog(
+                players=self.players, daemon=True
+            )
+            self._inacivity_watchdog.start()
+
+    def start_game(self, player_name: str) -> Player:
+        self._start_inactivity_watchdog()
+        player = Player(
+            name=player_name,
+            position=Position(0, 0),
+        )
+        self.players.append(player)
+
+        logging.info(f"Starting new game for player: {player}")
+        self.__debug_print_board()
+
+        return player
+
+    def move_player(self, player: Player, direction: Direction) -> PlayerMoveResult:
+        player.reset_timeout()
+
+        new_position = Position(player.position.x, player.position.y)
+        logging.info(f"Player {player} move to {direction}")
+
+        player.move_attempt_count += 1
+
+        if direction == Direction.LEFT:
+            new_position.x -= 1
+        elif direction == Direction.RIGHT:
+            new_position.x += 1
+        elif direction == Direction.UP:
+            new_position.y -= 1
+        elif direction == Direction.DOWN:
+            new_position.y += 1
+        else:
+            raise ValueError(f"Unhandled direction: {direction}")
+
+        if not self.position_in_board_bounds(new_position):
+            raise PositionOutOfBounds()
+
+        if self.colided_with_obstacle(new_position):
+            raise Collision()
+
+        player.position = new_position
+        player.move_count += 1
+
+        if self.is_player_on_destination(player):
+            logging.info(f"Player {player} reached destination!")
+            return PlayerMoveResult.DESTINATION_REACHED
+
+        self.__debug_print_board()
+        return PlayerMoveResult.OK
+
+    def is_player_on_destination(self, player: Player) -> bool:
+        return player.position == self.board.destination.position
+
+    def position_in_board_bounds(self, position: Position) -> bool:
+        return (
+            0 <= position.x < self.board.width and 0 <= position.y < self.board.height
+        )
+
+    def colided_with_obstacle(self, position: Position) -> bool:
+        return self.board.get_object_at_position(position) is not None
+
+
+class GameEngineFactory:
+    @staticmethod
+    def create(
+        board_width: int,
+        board_height: int,
+        obstacle_count: int = 0,
+    ) -> GameEngine:
+        board = GameBoard(
+            width=board_width,
+            height=board_height,
+            destination=Destination(Position(board_height // 2, board_height // 2)),
+        )
+        obstacle_layer = Layer(name="obstacles")
+        for _ in range(obstacle_count):
+            obstacle_layer.objects.append(
+                LayerObject(
+                    type_=ObjectType.OBSTACLE,
+                    position=create_random_position(board_width, board_height),
+                ),
+            )
+        board.layers.append(obstacle_layer)
+
+        game = GameEngine(board=board)
+        GameEngineFactory.__add_test_player(game.players)
+        return game
+
+    @staticmethod
+    def create_default() -> GameEngine:
+        return GameEngineFactory.create(
+            board_width=settings.board.WIDTH,
+            board_height=settings.board.HEIGHT,
+            obstacle_count=settings.board.OBSTACLE_COUNT,
+        )
+
+    @staticmethod
+    def __add_test_player(players: PlayerList) -> None:
+        if not (settings.debug and settings.debug.CREATE_TEST_PLAYER):
+            return
+        player = Player(
+            name="Pero",
+            uuid="test-player-id",
+            position=Position(2, 2),
+        )
+        players.append(player)
+        logging.info(f"Test player created: {player}")

hopper/enums.py

@@ -0,0 +1,20 @@
+from enum import Enum, auto
+
+
+class Direction(Enum):
+    LEFT = "left"
+    RIGHT = "right"
+    UP = "up"
+    DOWN = "down"
+
+
+class ObjectType(str, Enum):
+    NONE = auto()
+    OBSTACLE = auto()
+    PLAYER = auto()
+    DESTINATION = auto()
+
+
+class PlayerMoveResult(Enum):
+    OK = auto()
+    DESTINATION_REACHED = auto()

hopper/errors.py

@@ -0,0 +1,10 @@
+class BaseError(Exception):
+    ...
+
+
+class PositionOutOfBounds(BaseError):
+    ...
+
+
+class Collision(BaseError):
+    ...

hopper/models/board.py

@@ -0,0 +1,111 @@
+import random
+from copy import copy
+from dataclasses import dataclass, field
+from typing import Optional
+
+from hopper.enums import ObjectType
+from hopper.models.player import Player, Position
+
+BOARD_DUMP_CHARS: dict[ObjectType, str] = {
+    ObjectType.NONE: "·",
+    ObjectType.OBSTACLE: "✘",
+    ObjectType.PLAYER: "⯧",
+    ObjectType.DESTINATION: "⬤",
+}
+
+
+@dataclass
+class LayerObject:
+    type_: ObjectType
+    position: Position
+
+
+@dataclass
+class Layer:
+    name: Optional[str] = None
+    objects: list[LayerObject] = field(default_factory=list)
+
+    def get_object_at_position(self, position: Position) -> Optional[LayerObject]:
+        for obj in self.objects:
+            if obj.position == position:
+                return obj
+        return None
+
+
+@dataclass
+class Destination:
+    position: Position
+
+
+@dataclass
+class GameBoard:
+    width: int
+    height: int
+    destination: Destination
+    layers: list[Layer] = field(default_factory=list)
+
+    def dump(self) -> list[list[str]]:
+        board = [
+            [BOARD_DUMP_CHARS[ObjectType.NONE] for _ in range(self.width)]
+            for _ in range(self.height)
+        ]
+
+        for layer in self.layers:
+            for obj in layer.objects:
+                board[obj.position.y][obj.position.x] = BOARD_DUMP_CHARS[
+                    ObjectType.OBSTACLE
+                ]
+
+        board[self.destination.position.y][
+            self.destination.position.x
+        ] = BOARD_DUMP_CHARS[ObjectType.DESTINATION]
+
+        return board
+
+    def get_object_at_position(self, position: Position) -> Optional[LayerObject]:
+        for layer in self.layers:
+            obj = layer.get_object_at_position(position)
+            if obj is not None:
+                return obj
+        return None
+
+
+class BoardLayout:
+    def __init__(self, board: GameBoard, players: list[Player]) -> None:
+        self.board = board
+        self.players = players
+        self.layers = self.__create_layers()
+
+    def __create_layers(self) -> list[Layer]:
+        layers = copy(self.board.layers)
+        layers.append(
+            Layer(
+                name="destination",
+                objects=[
+                    LayerObject(
+                        type_=ObjectType.DESTINATION,
+                        position=self.board.destination.position,
+                    ),
+                ],
+            )
+        )
+        layers.append(
+            Layer(
+                name="players",
+                objects=[
+                    LayerObject(
+                        type_=ObjectType.PLAYER,
+                        position=player.position,
+                    )
+                    for player in self.players
+                ],
+            )
+        )
+        return layers
+
+
+def create_random_position(board_width: int, board_height: int) -> Position:
+    return Position(
+        x=random.randint(0, board_width - 1),
+        y=random.randint(0, board_height - 1),
+    )

hopper/models/config.py

@@ -0,0 +1,29 @@
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class BoardSettings:
+    WIDTH: int = 21
+    HEIGHT: int = 21
+    OBSTACLE_COUNT: int = 10
+
+
+@dataclass
+class InactivityWatchdogSettings:
+    INACIVITY_TIMEOUT: int = 10  # seconds
+    KICK_TIMEOUT: int = 60 * 10  # seconds
+    TICK_INTERVAL: int = 1  # seconds
+
+
+@dataclass
+class DebugSettings:
+    PRINT_BOARD: bool = False
+    CREATE_TEST_PLAYER: bool = False
+
+
+@dataclass
+class Settings:
+    board: BoardSettings
+    inacivity_watchdog: InactivityWatchdogSettings
+    debug: Optional[DebugSettings] = None

hopper/models/player.py

@@ -0,0 +1,34 @@
+import datetime
+import uuid
+from dataclasses import dataclass, field
+from typing import Optional
+
+
+@dataclass
+class Position:
+    x: int
+    y: int
+
+
+@dataclass
+class Player:
+    name: str
+    uuid: str = field(default_factory=lambda: str(uuid.uuid4()))
+    position: Position = field(default_factory=lambda: Position(0, 0))
+    move_count: int = 0
+    move_attempt_count: int = 0
+    last_seen: datetime.datetime = field(
+        default_factory=lambda: datetime.datetime.now()
+    )
+    active: bool = True
+
+    def reset_timeout(self) -> None:
+        self.last_seen = datetime.datetime.now()
+
+
+class PlayerList(list[Player]):
+    def find(self, uuid: str) -> Optional[Player]:
+        for player in self:
+            if player.uuid == uuid:
+                return player
+        return None

hopper/watchdog.py

@@ -0,0 +1,47 @@
+import datetime
+import logging
+import time
+from threading import Thread
+
+from hopper.models.player import PlayerList
+from settings import settings
+
+
+class InactivityWatchdog(Thread):
+    def __init__(self, players: PlayerList, *args, **kwargs) -> None:
+        self.players = players
+        self.stopped = False
+        super().__init__(*args, **kwargs)
+
+    def run(self) -> None:
+        logging.info("Starting inactivity watchdog")
+        while not self.stopped:
+            self.cleanup_players()
+            time.sleep(settings.inacivity_watchdog.TICK_INTERVAL)
+
+    def cleanup_players(self) -> None:
+        now = datetime.datetime.now()
+        inactivity_threshold = now - datetime.timedelta(
+            seconds=settings.inacivity_watchdog.INACIVITY_TIMEOUT
+        )
+        kick_threshold = now - datetime.timedelta(
+            seconds=settings.inacivity_watchdog.KICK_TIMEOUT
+        )
+
+        for player in self.players:
+            if player.active and player.last_seen < inactivity_threshold:
+                player.active = False
+                logging.info(f"Player {player} set as inactive")
+
+        # safe remove from list
+        n = 0
+        while n < len(self.players):
+            player = self.players[n]
+            if player.last_seen < kick_threshold:
+                self.players.pop(n)
+                logging.info(f"Player {player} kicked out")
+            else:
+                n += 1
+
+    def stop(self) -> None:
+        self.stopped = True