100 lines
3.3 KiB
Python
100 lines
3.3 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, Player, PlayerCreate
|
||
|
|
|
||
|
|
|
||
|
|
class GameEngine:
|
||
|
|
def __init__(self):
|
||
|
|
self._lock = asyncio.Lock()
|
||
|
|
self.players: Dict[str, Player] = {}
|
||
|
|
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) -> set[Tuple[int, int]]:
|
||
|
|
return {(p.x, p.y) 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:
|
||
|
|
# All spots taken, fallback to random any spot
|
||
|
|
return (
|
||
|
|
random.randint(self.config.min_x, self.config.max_x),
|
||
|
|
random.randint(self.config.min_y, self.config.max_y),
|
||
|
|
)
|
||
|
|
|
||
|
|
# Fast random sampling
|
||
|
|
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)
|
||
|
|
|
||
|
|
# Fallback exhaustive search if dense
|
||
|
|
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)
|
||
|
|
|
||
|
|
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
|
||
|
|
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 in self.players:
|
||
|
|
del self.players[player_id]
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
|
||
|
|
async def reset(self) -> None:
|
||
|
|
async with self._lock:
|
||
|
|
self.players.clear()
|
||
|
|
|
||
|
|
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,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# Global game engine instance
|
||
|
|
game_engine = GameEngine()
|