155 lines
4.7 KiB
Python
155 lines
4.7 KiB
Python
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),
|
|
can_be_deactivated=False,
|
|
)
|
|
players.append(player)
|
|
logging.info(f"Test player created: {player}")
|