import asyncio import random import uuid from typing import Dict, List, Optional, Set, Tuple from app.config import settings from app.models import ( AvailableMovesResponse, BattleResult, BoardConfig, BoardState, DIRECTION_OFFSETS, MoveCheckResult, MoveResponse, Party, PartyDefeatResult, PartyInvite, Player, PlayerCreate, TurnInfo, ) 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.parties: Dict[str, Party] = {} self.invites: Dict[str, PartyInvite] = {} 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_active_turn_actors(self) -> List[str]: # Return actors that can take turns: party leaders and solo players actors = [] for pid in self.turn_order: p = self.players.get(pid) if not p: continue if not p.party_id or p.is_party_leader: actors.append(pid) return actors def _get_current_player(self) -> Optional[Player]: actors = self._get_active_turn_actors() if not actors: return None safe_index = self.current_turn_index % len(actors) player_id = actors[safe_index] return self.players.get(player_id) def _get_turn_info(self) -> TurnInfo: current = self._get_current_player() actors = self._get_active_turn_actors() 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=actors, ) def _advance_turn(self): actors = self._get_active_turn_actors() if not actors: self.current_turn_index = 0 return self.turn_number += 1 self.current_turn_index = (self.current_turn_index + 1) % len(actors) 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, score=0, party_id=None, is_party_leader=False, ) 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 player = self.players[player_id] # If player was in a party, clean up party membership if player.party_id and player.party_id in self.parties: party = self.parties[player.party_id] if player_id in party.member_ids: party.member_ids.remove(player_id) if party.leader_id == player_id: if party.member_ids: new_leader_id = random.choice(party.member_ids) party.leader_id = new_leader_id new_lead = self.players.get(new_leader_id) if new_lead: new_lead.is_party_leader = True party.leader_name = new_lead.name else: del self.parties[party.id] elif not party.member_ids: del self.parties[party.id] del self.players[player_id] if player_id in self.turn_order: self.turn_order.remove(player_id) actors = self._get_active_turn_actors() if actors: self.current_turn_index %= len(actors) else: self.current_turn_index = 0 return True async def reset(self) -> None: async with self._lock: self.players.clear() self.parties.clear() self.invites.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()) parties_list = list(self.parties.values()) return BoardState( config=self.config, player_count=len(players_list), players=players_list, parties=parties_list, turn=self._get_turn_info(), ) # ========================================== # Party Logic & Graph Adjacency # ========================================== @staticmethod def _are_adjacent(p1: Player, p2: Player) -> bool: # Distance <= 1 (cardinal or diagonal Chebyshev distance) return max(abs(p1.x - p2.x), abs(p1.y - p2.y)) <= 1 def _verify_party_connectivity(self, member_players: List[Player]) -> bool: """Verifies all party members form a single connected component where each is within 1 distance.""" if len(member_players) <= 1: return True visited: Set[str] = set() queue = [member_players[0]] visited.add(member_players[0].id) while queue: curr = queue.pop(0) for other in member_players: if other.id not in visited and self._are_adjacent(curr, other): visited.add(other.id) queue.append(other) return len(visited) == len(member_players) async def form_party( self, member_ids: List[str], leader_id: str, name: Optional[str] = None ) -> Party: async with self._lock: if len(member_ids) < 2: raise ValueError("A party must have at least 2 members") if leader_id not in member_ids: raise ValueError("Agreed leader must be one of the party members") member_players: List[Player] = [] for mid in member_ids: p = self.players.get(mid) if not p: raise KeyError(f"Player '{mid}' not found") member_players.append(p) # Check adjacency / connected link if not self._verify_party_connectivity(member_players): raise ValueError( "All party members must be linked within 1 distance of each other (directly or via connected teammates)." ) # Clean up any existing party memberships for p in member_players: if p.party_id and p.party_id in self.parties: old_party = self.parties[p.party_id] if p.id in old_party.member_ids: old_party.member_ids.remove(p.id) if not old_party.member_ids: del self.parties[old_party.id] party_id = f"party_{uuid.uuid4().hex[:8]}" leader = self.players[leader_id] party_name = name or f"Squad {leader.name}" party = Party( id=party_id, name=party_name, leader_id=leader_id, leader_name=leader.name, member_ids=list(member_ids), ) for p in member_players: p.party_id = party_id p.is_party_leader = p.id == leader_id self.parties[party_id] = party return party async def invite_to_party( self, inviter_id: str, invitee_id: str, proposed_leader_id: str, party_name: Optional[str] = None, ) -> PartyInvite: async with self._lock: inviter = self.players.get(inviter_id) invitee = self.players.get(invitee_id) if not inviter or not invitee: raise KeyError("Inviter or invitee not found") # Must be adjacent to inviter or to at least one member of inviter's party eligible_hosts = [inviter] if inviter.party_id and inviter.party_id in self.parties: party = self.parties[inviter.party_id] eligible_hosts = [self.players[mid] for mid in party.member_ids if mid in self.players] is_adjacent = any(self._are_adjacent(invitee, host) for host in eligible_hosts) if not is_adjacent: raise ValueError( f"Bot '{invitee.name}' is too far away. Must be within 1 distance of a party member to join." ) invite_id = f"inv_{uuid.uuid4().hex[:8]}" invite = PartyInvite( id=invite_id, inviter_id=inviter_id, invitee_id=invitee_id, proposed_leader_id=proposed_leader_id, party_id=inviter.party_id, ) self.invites[invite_id] = invite return invite async def respond_to_invite(self, invite_id: str, accept: bool) -> Optional[Party]: async with self._lock: invite = self.invites.get(invite_id) if not invite: raise KeyError(f"Invite '{invite_id}' not found") if invite.status != "pending": raise ValueError(f"Invite already {invite.status}") if not accept: invite.status = "rejected" return None invite.status = "accepted" inviter = self.players.get(invite.inviter_id) invitee = self.players.get(invite.invitee_id) if not inviter or not invitee: raise KeyError("Inviter or invitee no longer active") # If inviter is already in a party, add invitee to that party if inviter.party_id and inviter.party_id in self.parties: party = self.parties[inviter.party_id] if invitee.id not in party.member_ids: party.member_ids.append(invitee.id) invitee.party_id = party.id # Set agreed leader if invite.proposed_leader_id in party.member_ids: party.leader_id = invite.proposed_leader_id for mid in party.member_ids: p = self.players.get(mid) if p: p.is_party_leader = p.id == party.leader_id party.leader_name = self.players[party.leader_id].name return party # Otherwise form new party with both members = [inviter.id, invitee.id] leader_id = invite.proposed_leader_id if invite.proposed_leader_id in members else inviter.id party_id = f"party_{uuid.uuid4().hex[:8]}" leader = self.players[leader_id] party = Party( id=party_id, name=f"Squad {leader.name}", leader_id=leader_id, leader_name=leader.name, member_ids=members, ) inviter.party_id = party_id inviter.is_party_leader = inviter.id == leader_id invitee.party_id = party_id invitee.is_party_leader = invitee.id == leader_id self.parties[party_id] = party return party async def get_all_parties(self) -> List[Party]: async with self._lock: return list(self.parties.values()) async def get_party(self, party_id: str) -> Optional[Party]: async with self._lock: return self.parties.get(party_id) # ========================================== # Movement Checking (Single bot or Party) # ========================================== def _check_move_internal( self, player: Player, dx: int, dy: int, direction_name: str, occupied_map: Dict[Tuple[int, int], Player], ) -> MoveCheckResult: # If player is part of a party and IS the leader, check if ALL members can translate by (dx, dy) if player.party_id and player.party_id in self.parties and player.is_party_leader: party = self.parties[player.party_id] party_members = [self.players[mid] for mid in party.member_ids if mid in self.players] party_member_ids = {m.id for m in party_members} for m in party_members: tx = m.x + dx ty = m.y + dy # Wall boundary if tx < self.config.min_x or tx > self.config.max_x or ty < self.config.min_y or ty > self.config.max_y: return MoveCheckResult( direction=direction_name, dx=dx, dy=dy, target_x=player.x + dx, target_y=player.y + dy, available=False, reason=f"Party member '{m.name}' would hit boundary wall at ({tx}, {ty}).", ) # Collision with non-party bot occupant = occupied_map.get((tx, ty)) if occupant is not None and occupant.id not in party_member_ids: return MoveCheckResult( direction=direction_name, dx=dx, dy=dy, target_x=player.x + dx, target_y=player.y + dy, available=False, reason=f"Party member '{m.name}' path blocked by '{occupant.name}' at ({tx}, {ty}).", ) return MoveCheckResult( direction=direction_name, dx=dx, dy=dy, target_x=player.x + dx, target_y=player.y + dy, available=True, reason=None, ) # Single player or non-leader target_x = player.x + dx target_y = player.y + dy 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}).", ) 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") # Non-leaders cannot move independently if player.party_id and not player.is_party_leader: party = self.parties.get(player.party_id) leader_name = party.leader_name if party else "Leader" return MoveCheckResult( direction=direction_input, dx=0, dy=0, target_x=player.x, target_y=player.y, available=False, reason=f"Only party leader '{leader_name}' controls movement for the party.", ) 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" ) 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() party = self.parties.get(player.party_id) if player.party_id else None is_leader = player.is_party_leader or party is None member_count = len(party.member_ids) if party else 1 moves: Dict[str, MoveCheckResult] = {} for name, dx, dy in STANDARD_DIRECTIONS: if player.party_id and not player.is_party_leader: moves[name] = MoveCheckResult( direction=name, dx=dx, dy=dy, target_x=player.x + dx, target_y=player.y + dy, available=False, reason=f"Only party leader '{party.leader_name if party else 'Leader'}' controls group movement.", ) else: 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 and is_leader, is_party_leader=player.is_party_leader, party_id=player.party_id, party_member_count=member_count, 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") # Non-leaders cannot move if player.party_id and not player.is_party_leader: party = self.parties.get(player.party_id) leader_name = party.leader_name if party else "Leader" raise PermissionError( f"Party member '{player.name}' cannot move individually. Only party leader '{leader_name}' controls movement for the party." ) 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} affected_players: List[Player] = [] # If party leader, move all party members in unison if player.party_id and player.party_id in self.parties: party = self.parties[player.party_id] for mid in party.member_ids: m = self.players.get(mid) if m: m.x += dx m.y += dy affected_players.append(m) else: player.x = check.target_x player.y = check.target_y affected_players.append(player) new_pos = {"x": player.x, "y": player.y} self._advance_turn() turn_info = self._get_turn_info() return MoveResponse( success=True, player=player, direction=direction_name, party_moved=bool(player.party_id), affected_players=affected_players, 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() # ========================================== # Battle & Party Defeat Logic # ========================================== async def defeat_party(self, party_id: str) -> PartyDefeatResult: """Kills the party leader: score -= 1, respawns at random position, removed from party, and randomly assigns a new leader from the remainder of the party. """ async with self._lock: party = self.parties.get(party_id) if not party: raise KeyError(f"Party '{party_id}' not found") old_leader_id = party.leader_id leader = self.players.get(old_leader_id) if not leader: raise KeyError(f"Leader '{old_leader_id}' not found") # 1. Leader killed: score goes down by 1 leader.score -= 1 # 2. Leader is removed from party leader.party_id = None leader.is_party_leader = False if old_leader_id in party.member_ids: party.member_ids.remove(old_leader_id) # 3. Leader respawns at random free position rx, ry = self._find_random_free_position() leader.x = rx leader.y = ry respawn_pos = {"x": rx, "y": ry} # 4. New leader is randomly assigned from the remainder of the party new_leader_id = None new_leader_name = None party_dissolved = False if len(party.member_ids) > 0: new_leader_id = random.choice(party.member_ids) party.leader_id = new_leader_id new_leader = self.players[new_leader_id] new_leader.is_party_leader = True new_leader_name = new_leader.name party.leader_name = new_leader_name else: # No members left in party del self.parties[party_id] party_dissolved = True return PartyDefeatResult( party_id=party_id, killed_leader_id=leader.id, killed_leader_name=leader.name, killed_leader_new_score=leader.score, killed_leader_respawn_position=respawn_pos, new_leader_id=new_leader_id, new_leader_name=new_leader_name, remaining_members=list(party.member_ids), party_dissolved=party_dissolved, ) async def battle(self, challenger_id: str, defender_id: str) -> BattleResult: """Resolves a battle between two adjacent players/bots. The losing party's leader is killed, respawns, score -1, and remainder assigns new leader. """ async with self._lock: p1 = self.players.get(challenger_id) p2 = self.players.get(defender_id) if not p1 or not p2: raise KeyError("Both combatants must exist") if p1.party_id and p1.party_id == p2.party_id: raise ValueError("Cannot battle your own party member") if not self._are_adjacent(p1, p2): raise ValueError( f"Players are not adjacent. Combat distance must be <= 1 (current distance: ({abs(p1.x - p2.x)}, {abs(p1.y - p2.y)}))" ) # Party size calculation for battle advantage size1 = len(self.parties[p1.party_id].member_ids) if p1.party_id and p1.party_id in self.parties else 1 size2 = len(self.parties[p2.party_id].member_ids) if p2.party_id and p2.party_id in self.parties else 1 if size1 > size2: winner, loser = p1, p2 elif size2 > size1: winner, loser = p2, p1 else: # Tied party sizes: 50/50 roll if random.random() < 0.5: winner, loser = p1, p2 else: winner, loser = p2, p1 # Winner score +1 winner.score += 1 # Resolve defeat for loser if loser.party_id and loser.party_id in self.parties: # Must release lock temporarily or run defeat internally party_id = loser.party_id party = self.parties[party_id] old_leader_id = party.leader_id leader = self.players[old_leader_id] leader.score -= 1 leader.party_id = None leader.is_party_leader = False if old_leader_id in party.member_ids: party.member_ids.remove(old_leader_id) rx, ry = self._find_random_free_position() leader.x = rx leader.y = ry respawn_pos = {"x": rx, "y": ry} new_leader_id = None new_leader_name = None party_dissolved = False if len(party.member_ids) > 0: new_leader_id = random.choice(party.member_ids) party.leader_id = new_leader_id new_leader = self.players[new_leader_id] new_leader.is_party_leader = True new_leader_name = new_leader.name party.leader_name = new_leader_name else: del self.parties[party_id] party_dissolved = True defeat_result = PartyDefeatResult( party_id=party_id, killed_leader_id=leader.id, killed_leader_name=leader.name, killed_leader_new_score=leader.score, killed_leader_respawn_position=respawn_pos, new_leader_id=new_leader_id, new_leader_name=new_leader_name, remaining_members=list(party.member_ids), party_dissolved=party_dissolved, ) else: # Solo loser loser.score -= 1 rx, ry = self._find_random_free_position() loser.x = rx loser.y = ry defeat_result = PartyDefeatResult( party_id="none", killed_leader_id=loser.id, killed_leader_name=loser.name, killed_leader_new_score=loser.score, killed_leader_respawn_position={"x": rx, "y": ry}, new_leader_id=None, new_leader_name=None, remaining_members=[], party_dissolved=True, ) return BattleResult( winner_id=winner.id, winner_name=winner.name, loser_id=loser.id, loser_name=loser.name, defeat_details=defeat_result, ) # Global game engine instance game_engine = GameEngine()