314 lines
11 KiB
Python
314 lines
11 KiB
Python
import asyncio
|
|
import random
|
|
import uuid
|
|
from typing import Dict, List, Optional, Tuple
|
|
from app.config import settings
|
|
from app.models import (
|
|
BoardConfig,
|
|
BoardState,
|
|
MoveCheckResult,
|
|
MoveResponse,
|
|
Player,
|
|
PlayerCreate,
|
|
TurnInfo,
|
|
DIRECTION_OFFSETS,
|
|
AvailableMovesResponse,
|
|
)
|
|
|
|
STANDARD_DIRECTIONS = [
|
|
("UP", 0, -1),
|
|
("UP_RIGHT", 1, -1),
|
|
("RIGHT", 1, 0),
|
|
("DOWN_RIGHT", 1, 1),
|
|
("DOWN", 0, 1),
|
|
("DOWN_LEFT", -1, 1),
|
|
("LEFT", -1, 0),
|
|
("UP_LEFT", -1, -1),
|
|
]
|
|
|
|
|
|
class GameEngine:
|
|
def __init__(self):
|
|
self._lock = asyncio.Lock()
|
|
self.players: Dict[str, Player] = {}
|
|
self.turn_order: List[str] = []
|
|
self.current_turn_index: int = 0
|
|
self.round_number: int = 1
|
|
self.turn_number: int = 0
|
|
|
|
self.config = BoardConfig(
|
|
min_x=settings.GRID_MIN_X,
|
|
max_x=settings.GRID_MAX_X,
|
|
min_y=settings.GRID_MIN_Y,
|
|
max_y=settings.GRID_MAX_Y,
|
|
grid_cells_x=settings.GRID_MAX_X - settings.GRID_MIN_X,
|
|
grid_cells_y=settings.GRID_MAX_Y - settings.GRID_MIN_Y,
|
|
)
|
|
|
|
def _get_occupied_coordinates(self) -> Dict[Tuple[int, int], Player]:
|
|
return {(p.x, p.y): p for p in self.players.values()}
|
|
|
|
def _find_random_free_position(self) -> Tuple[int, int]:
|
|
occupied = self._get_occupied_coordinates()
|
|
total_possible = (self.config.max_x - self.config.min_x + 1) * (
|
|
self.config.max_y - self.config.min_y + 1
|
|
)
|
|
|
|
if len(occupied) >= total_possible:
|
|
return (
|
|
random.randint(self.config.min_x, self.config.max_x),
|
|
random.randint(self.config.min_y, self.config.max_y),
|
|
)
|
|
|
|
for _ in range(100):
|
|
rx = random.randint(self.config.min_x, self.config.max_x)
|
|
ry = random.randint(self.config.min_y, self.config.max_y)
|
|
if (rx, ry) not in occupied:
|
|
return (rx, ry)
|
|
|
|
all_coords = [
|
|
(x, y)
|
|
for x in range(self.config.min_x, self.config.max_x + 1)
|
|
for y in range(self.config.min_y, self.config.max_y + 1)
|
|
if (x, y) not in occupied
|
|
]
|
|
return random.choice(all_coords)
|
|
|
|
def _get_current_player(self) -> Optional[Player]:
|
|
if not self.turn_order:
|
|
return None
|
|
safe_index = self.current_turn_index % len(self.turn_order)
|
|
player_id = self.turn_order[safe_index]
|
|
return self.players.get(player_id)
|
|
|
|
def _get_turn_info(self) -> TurnInfo:
|
|
current = self._get_current_player()
|
|
return TurnInfo(
|
|
current_player_id=current.id if current else None,
|
|
current_player_name=current.name if current else None,
|
|
round_number=self.round_number,
|
|
turn_number=self.turn_number,
|
|
turn_order=list(self.turn_order),
|
|
)
|
|
|
|
def _advance_turn(self):
|
|
if not self.turn_order:
|
|
self.current_turn_index = 0
|
|
return
|
|
self.turn_number += 1
|
|
self.current_turn_index = (self.current_turn_index + 1) % len(self.turn_order)
|
|
if self.current_turn_index == 0:
|
|
self.round_number += 1
|
|
|
|
async def register_player(self, player_in: PlayerCreate) -> Player:
|
|
async with self._lock:
|
|
player_id = f"bot_{uuid.uuid4().hex[:8]}"
|
|
x, y = self._find_random_free_position()
|
|
|
|
player = Player(
|
|
id=player_id,
|
|
name=player_in.name,
|
|
color=player_in.color,
|
|
x=x,
|
|
y=y,
|
|
)
|
|
self.players[player_id] = player
|
|
self.turn_order.append(player_id)
|
|
return player
|
|
|
|
async def get_player(self, player_id: str) -> Optional[Player]:
|
|
async with self._lock:
|
|
return self.players.get(player_id)
|
|
|
|
async def get_all_players(self) -> List[Player]:
|
|
async with self._lock:
|
|
return list(self.players.values())
|
|
|
|
async def remove_player(self, player_id: str) -> bool:
|
|
async with self._lock:
|
|
if player_id not in self.players:
|
|
return False
|
|
|
|
del self.players[player_id]
|
|
|
|
if player_id in self.turn_order:
|
|
idx = self.turn_order.index(player_id)
|
|
self.turn_order.remove(player_id)
|
|
|
|
if self.turn_order:
|
|
if idx < self.current_turn_index:
|
|
self.current_turn_index -= 1
|
|
self.current_turn_index %= len(self.turn_order)
|
|
else:
|
|
self.current_turn_index = 0
|
|
|
|
return True
|
|
|
|
async def reset(self) -> None:
|
|
async with self._lock:
|
|
self.players.clear()
|
|
self.turn_order.clear()
|
|
self.current_turn_index = 0
|
|
self.round_number = 1
|
|
self.turn_number = 0
|
|
|
|
async def get_board_state(self) -> BoardState:
|
|
async with self._lock:
|
|
players_list = list(self.players.values())
|
|
return BoardState(
|
|
config=self.config,
|
|
player_count=len(players_list),
|
|
players=players_list,
|
|
turn=self._get_turn_info(),
|
|
)
|
|
|
|
def _check_move_internal(
|
|
self,
|
|
player: Player,
|
|
dx: int,
|
|
dy: int,
|
|
direction_name: str,
|
|
occupied_map: Dict[Tuple[int, int], Player],
|
|
) -> MoveCheckResult:
|
|
target_x = player.x + dx
|
|
target_y = player.y + dy
|
|
|
|
# 1. Boundary Wall Check
|
|
if (
|
|
target_x < self.config.min_x
|
|
or target_x > self.config.max_x
|
|
or target_y < self.config.min_y
|
|
or target_y > self.config.max_y
|
|
):
|
|
return MoveCheckResult(
|
|
direction=direction_name,
|
|
dx=dx,
|
|
dy=dy,
|
|
target_x=target_x,
|
|
target_y=target_y,
|
|
available=False,
|
|
reason=f"Wall collision at ({target_x}, {target_y}). Grid boundaries are [{self.config.min_x}..{self.config.max_x}, {self.config.min_y}..{self.config.max_y}].",
|
|
)
|
|
|
|
# 2. Occupancy Check
|
|
occupant = occupied_map.get((target_x, target_y))
|
|
if occupant is not None and occupant.id != player.id:
|
|
return MoveCheckResult(
|
|
direction=direction_name,
|
|
dx=dx,
|
|
dy=dy,
|
|
target_x=target_x,
|
|
target_y=target_y,
|
|
available=False,
|
|
reason=f"Space occupied by {occupant.name} ({occupant.id}) at ({target_x}, {target_y}).",
|
|
)
|
|
|
|
return MoveCheckResult(
|
|
direction=direction_name,
|
|
dx=dx,
|
|
dy=dy,
|
|
target_x=target_x,
|
|
target_y=target_y,
|
|
available=True,
|
|
reason=None,
|
|
)
|
|
|
|
async def check_single_move(self, player_id: str, direction_input: str) -> MoveCheckResult:
|
|
async with self._lock:
|
|
player = self.players.get(player_id)
|
|
if not player:
|
|
raise KeyError(f"Player '{player_id}' not found")
|
|
|
|
normalized = direction_input.strip().upper().replace(" ", "_")
|
|
if normalized not in DIRECTION_OFFSETS:
|
|
raise ValueError(
|
|
f"Unknown direction '{direction_input}'. Allowed: UP, DOWN, LEFT, RIGHT, UP_LEFT, UP_RIGHT, DOWN_LEFT, DOWN_RIGHT (or N, S, W, E, NW, NE, SW, SE)"
|
|
)
|
|
|
|
dx, dy = DIRECTION_OFFSETS[normalized]
|
|
occupied = self._get_occupied_coordinates()
|
|
return self._check_move_internal(player, dx, dy, normalized, occupied)
|
|
|
|
async def get_available_moves(self, player_id: str) -> AvailableMovesResponse:
|
|
async with self._lock:
|
|
player = self.players.get(player_id)
|
|
if not player:
|
|
raise KeyError(f"Player '{player_id}' not found")
|
|
|
|
current_turn_player = self._get_current_player()
|
|
is_turn = current_turn_player is not None and current_turn_player.id == player_id
|
|
occupied = self._get_occupied_coordinates()
|
|
|
|
moves: Dict[str, MoveCheckResult] = {}
|
|
for name, dx, dy in STANDARD_DIRECTIONS:
|
|
moves[name] = self._check_move_internal(player, dx, dy, name, occupied)
|
|
|
|
return AvailableMovesResponse(
|
|
player_id=player.id,
|
|
player_name=player.name,
|
|
current_x=player.x,
|
|
current_y=player.y,
|
|
is_turn=is_turn,
|
|
current_turn_player_id=current_turn_player.id if current_turn_player else None,
|
|
moves=moves,
|
|
)
|
|
|
|
async def move_player(self, player_id: str, dx: int, dy: int, direction_name: str) -> MoveResponse:
|
|
async with self._lock:
|
|
player = self.players.get(player_id)
|
|
if not player:
|
|
raise KeyError(f"Player '{player_id}' not found")
|
|
|
|
current_turn_player = self._get_current_player()
|
|
if not current_turn_player or current_turn_player.id != player_id:
|
|
curr_name = current_turn_player.name if current_turn_player else "Nobody"
|
|
curr_id = current_turn_player.id if current_turn_player else "None"
|
|
raise PermissionError(
|
|
f"It is not your turn. Current turn belongs to '{curr_name}' ({curr_id})."
|
|
)
|
|
|
|
occupied = self._get_occupied_coordinates()
|
|
check = self._check_move_internal(player, dx, dy, direction_name, occupied)
|
|
|
|
if not check.available:
|
|
raise ValueError(f"Illegal move: {check.reason}")
|
|
|
|
prev_pos = {"x": player.x, "y": player.y}
|
|
player.x = check.target_x
|
|
player.y = check.target_y
|
|
new_pos = {"x": player.x, "y": player.y}
|
|
|
|
# Advance turn order to next player
|
|
self._advance_turn()
|
|
turn_info = self._get_turn_info()
|
|
|
|
return MoveResponse(
|
|
success=True,
|
|
player=player,
|
|
direction=direction_name,
|
|
previous_position=prev_pos,
|
|
new_position=new_pos,
|
|
turn=turn_info,
|
|
)
|
|
|
|
async def pass_turn(self, player_id: str) -> TurnInfo:
|
|
async with self._lock:
|
|
player = self.players.get(player_id)
|
|
if not player:
|
|
raise KeyError(f"Player '{player_id}' not found")
|
|
|
|
current_turn_player = self._get_current_player()
|
|
if not current_turn_player or current_turn_player.id != player_id:
|
|
curr_name = current_turn_player.name if current_turn_player else "Nobody"
|
|
curr_id = current_turn_player.id if current_turn_player else "None"
|
|
raise PermissionError(
|
|
f"Cannot pass: it is not your turn. Current turn belongs to '{curr_name}' ({curr_id})."
|
|
)
|
|
|
|
self._advance_turn()
|
|
return self._get_turn_info()
|
|
|
|
|
|
# Global game engine instance
|
|
game_engine = GameEngine()
|