import asyncio import random import uuid from typing import Dict, List, Optional, Set, Tuple from app.config import settings from app.models import ( AiStepResponse, AvailableMovesResponse, BattleBout, BattleResult, BoardConfig, BoardState, BotMemoryResponse, BotRadarResponse, DIRECTION_OFFSETS, GameConclusion, MoveCheckResult, MoveResponse, Obstacle, Party, PartyDefeatResult, PartyInvite, Player, PlayerCreate, RadarTarget, TurnInfo, WizardChallengeBout, WizardChallengeResult, WizardNPC, WizardRadarTarget, ) 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.game_started: bool = False 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, ) # Procedural Impassable Obstacles (Mountains & Valleys) # Guaranteed: <= 50% impassable, and all passable tiles form a single connected component self.obstacles: Dict[Tuple[int, int], Obstacle] = self._generate_terrain() self.wizard: WizardNPC = self._spawn_wizard() def _spawn_wizard(self) -> WizardNPC: """Spawn the wandering Wizard NPC at a random free passable coordinate.""" wx, wy = self._find_random_free_position() return WizardNPC( id="wizard_npc", name="Gary the Wizard", x=wx, y=wy, strength=3.0, color="#A855F7", dialogue="Greetings, traveler! Do you dare challenge my arcane arts?", ) def _generate_terrain(self) -> Dict[Tuple[int, int], Obstacle]: """Generates procedural mountain ranges and valley chasms subject to: 1. Obstacles must NOT exceed 50% of the map (kept at ~20-35%). 2. Every passable coordinate must belong to a single connected component (no isolated pockets). """ min_x, max_x = self.config.min_x, self.config.max_x min_y, max_y = self.config.min_y, self.config.max_y total_cells = (max_x - min_x + 1) * (max_y - min_y + 1) max_allowed_obstacles = int(total_cells * 0.40) # Safe margin below 50% obstacles: Dict[Tuple[int, int], Obstacle] = {} def add_obstacle(x: int, y: int, obs_type: str): if min_x <= x <= max_x and min_y <= y <= max_y: obstacles[(x, y)] = Obstacle(x=x, y=y, type=obs_type) # 1. Procedural Mountain ranges (6-8 organic ridges) for _ in range(random.randint(6, 8)): curr_x = random.randint(min_x + 5, max_x - 5) curr_y = random.randint(min_y + 5, max_y - 5) length = random.randint(18, 32) dx = random.choice([-1, 0, 1]) dy = random.choice([-1, 0, 1]) if dx == 0 and dy == 0: dx, dy = 1, 1 for _ in range(length): add_obstacle(curr_x, curr_y, "mountain") if random.random() < 0.35: add_obstacle(curr_x + dy, curr_y - dx, "mountain") if random.random() < 0.3: dx += random.choice([-1, 0, 1]) dx = max(-1, min(1, dx)) if random.random() < 0.3: dy += random.choice([-1, 0, 1]) dy = max(-1, min(1, dy)) if dx == 0 and dy == 0: dx = 1 curr_x += dx curr_y += dy # 2. Procedural Dense Forests / Woodlands (6-8 organic forest groves) for _ in range(random.randint(6, 8)): center_x = random.randint(min_x + 6, max_x - 6) center_y = random.randint(min_y + 6, max_y - 6) grove_cells = [(center_x, center_y)] target_trees = random.randint(22, 36) for _ in range(target_trees): if not grove_cells: break cx, cy = random.choice(grove_cells) if (cx, cy) not in obstacles: add_obstacle(cx, cy, "forest") for nx, ny in [(cx + 1, cy), (cx - 1, cy), (cx, cy + 1), (cx, cy - 1)]: if min_x <= nx <= max_x and min_y <= ny <= max_y and (nx, ny) not in obstacles: if random.random() < 0.65: grove_cells.append((nx, ny)) # Enforce max allowed obstacles (cap strictly below 50%) if len(obstacles) > max_allowed_obstacles: keys_to_remove = random.sample(list(obstacles.keys()), len(obstacles) - max_allowed_obstacles) for k in keys_to_remove: del obstacles[k] # 3. Connectivity Verification & Single Component Guarantee: # All passable cells must form exactly ONE connected component. all_coords = set( (x, y) for x in range(min_x, max_x + 1) for y in range(min_y, max_y + 1) ) passable = all_coords - set(obstacles.keys()) visited: Set[Tuple[int, int]] = set() components: List[Set[Tuple[int, int]]] = [] for cell in passable: if cell in visited: continue comp: Set[Tuple[int, int]] = set() queue = [cell] visited.add(cell) comp.add(cell) while queue: cx, cy = queue.pop(0) for nx, ny in [(cx + 1, cy), (cx - 1, cy), (cx, cy + 1), (cx, cy - 1)]: if min_x <= nx <= max_x and min_y <= ny <= max_y: if (nx, ny) in passable and (nx, ny) not in visited: visited.add((nx, ny)) comp.add((nx, ny)) queue.append((nx, ny)) components.append(comp) if not components: return {} components.sort(key=lambda c: len(c), reverse=True) main_component = components[0] # Connect all isolated components to main_component by carving passes through obstacles for other_comp in components[1:]: src = next(iter(other_comp)) sample_size = min(len(main_component), 50) sample_targets = random.sample(list(main_component), sample_size) target = min(sample_targets, key=lambda c: abs(c[0] - src[0]) + abs(c[1] - src[1])) cx, cy = src tx, ty = target while (cx, cy) != (tx, ty): if cx < tx: cx += 1 elif cx > tx: cx -= 1 elif cy < ty: cy += 1 elif cy > ty: cy -= 1 if (cx, cy) in obstacles: del obstacles[(cx, cy)] main_component.add((cx, cy)) main_component.update(other_comp) return obstacles def _find_path_bfs( self, start: Tuple[int, int], goal: Tuple[int, int], max_depth: int = 50 ) -> Optional[List[Tuple[int, int]]]: """Breadth-first search pathfinder finding shortest passable path avoiding obstacles.""" if start == goal: return [start] queue: List[Tuple[Tuple[int, int], List[Tuple[int, int]]]] = [(start, [start])] visited: Set[Tuple[int, int]] = {start} while queue: (curr_x, curr_y), path = queue.pop(0) if len(path) > max_depth: break for _, dx, dy in STANDARD_DIRECTIONS: nx, ny = curr_x + dx, curr_y + dy if ( self.config.min_x <= nx <= self.config.max_x and self.config.min_y <= ny <= self.config.max_y and (nx, ny) not in self.obstacles and (nx, ny) not in visited ): new_path = path + [(nx, ny)] if (nx, ny) == goal: return new_path visited.add((nx, ny)) queue.append(((nx, ny), new_path)) return None 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]: """Finds a random free position strictly excluding occupied tiles AND impassable obstacles.""" occupied = self._get_occupied_coordinates() # Fast random sampling for _ in range(200): 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 and (rx, ry) not in self.obstacles: return (rx, ry) # Fallback to list of all valid passable, unoccupied coordinates 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 and (x, y) not in self.obstacles ] if all_coords: return random.choice(all_coords) # Absolute safety fallback return ( random.randint(self.config.min_x, self.config.max_x), random.randint(self.config.min_y, self.config.max_y), ) def _check_and_apply_death(self, player: Player) -> bool: """If player's health <= 0, mark dead, disconnect from party, and remove from turn rotation.""" if player.health <= 0: player.health = 0 player.is_alive = False # Disconnect from any existing party 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: surviving = [ self.players[mid] for mid in party.member_ids if mid in self.players and self.players[mid].is_alive ] if surviving: surviving.sort(key=lambda p: (p.strength, p.score), reverse=True) new_leader = surviving[0] party.leader_id = new_leader.id party.leader_name = new_leader.name new_leader.is_party_leader = True self._update_party_strength(party) else: if party.id in self.parties: del self.parties[party.id] elif party.member_ids: self._update_party_strength(party) else: if party.id in self.parties: del self.parties[party.id] player.party_id = None player.is_party_leader = False 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 return False def _get_active_turn_actors(self) -> List[str]: if not self.game_started: return [] actors = [] for pid in self.turn_order: p = self.players.get(pid) if not p or not p.is_alive: 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: if not self.game_started: return TurnInfo( game_started=False, current_player_id=None, current_player_name=None, round_number=0, turn_number=0, turn_order=[], ) current = self._get_current_player() actors = self._get_active_turn_actors() return TurnInfo( game_started=True, 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.turn_number += 1 return self.current_turn_index = (self.current_turn_index + 1) % len(actors) self.turn_number += 1 if self.current_turn_index == 0: self.round_number += 1 async def register_player(self, player_in: PlayerCreate) -> Player: async with self._lock: if len(self.players) >= settings.MAX_PLAYERS: raise ValueError( f"Maximum players ({settings.MAX_PLAYERS}) reached. Cannot register more players." ) for existing in self.players.values(): if existing.name.lower() == player_in.name.lower(): raise ValueError(f"Player with name '{player_in.name}' is already registered.") player_id = f"player_{uuid.uuid4().hex[:8]}" spawn_x, spawn_y = self._find_random_free_position() # Determine piece class (knight or warrior) piece_type = getattr(player_in, "piece_type", None) if not piece_type: name_lower = player_in.name.lower() if "warrior" in name_lower or "striker" in name_lower or "scout" in name_lower: piece_type = "warrior" elif "knight" in name_lower or "tank" in name_lower or "titan" in name_lower: piece_type = "knight" else: piece_type = "knight" if len(player_in.name) % 2 == 0 else "warrior" health = getattr(player_in, "health", 10) if health is None: health = 10 player = Player( id=player_id, name=player_in.name, color=player_in.color, strength=player_in.strength, score=0, health=health, max_health=health, x=spawn_x, y=spawn_y, piece_type=piece_type, party_id=None, is_party_leader=False, visited_locations=[{"x": spawn_x, "y": spawn_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()) # Alias remove_player to delete_player async def remove_player(self, player_id: str) -> bool: return await self.delete_player(player_id) async def delete_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.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 self._update_party_strength(party) else: del self.parties[party.id] elif party.member_ids: self._update_party_strength(party) else: 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 start_game(self) -> TurnInfo: async with self._lock: if self.game_started: return self._get_turn_info() if len(self.players) == 0: raise ValueError("Cannot start game: no bots have joined yet. Spawn or register bots first.") self.game_started = True # Build turn order from registered players self.turn_order = list(self.players.keys()) self.current_turn_index = 0 self.round_number = 1 self.turn_number = 1 return self._get_turn_info() 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 self.game_started = False # Regenerate fresh procedural mountain ranges and valley trenches on reset self.obstacles = self._generate_terrain() self.wizard = self._spawn_wizard() async def get_board_state(self) -> BoardState: async with self._lock: players_list = list(self.players.values()) parties_list = list(self.parties.values()) obstacles_list = list(self.obstacles.values()) return BoardState( config=self.config, player_count=len(players_list), players=players_list, parties=parties_list, obstacles=obstacles_list, wizard=self.wizard, turn=self._get_turn_info(), game_started=self.game_started, conclusion=self._check_game_concluded(), ) # ========================================== # Bot Memory & Radar Awareness # ========================================== async def get_bot_memory( self, player_id: str, check_x: Optional[int] = None, check_y: Optional[int] = None ) -> BotMemoryResponse: async with self._lock: player = self.players.get(player_id) if not player: raise KeyError(f"Player '{player_id}' not found") has_visited = True if check_x is not None and check_y is not None: has_visited = any(loc["x"] == check_x and loc["y"] == check_y for loc in player.visited_locations) return BotMemoryResponse( player_id=player.id, player_name=player.name, current_x=player.x, current_y=player.y, visited_count=len(player.visited_locations), visited_history=list(player.visited_locations), has_visited_current=has_visited, ) async def get_bot_radar(self, player_id: str) -> BotRadarResponse: async with self._lock: player = self.players.get(player_id) if not player: raise KeyError(f"Player '{player_id}' not found") bot_goal = "find_and_defeat_all_parties" if player.party_id else "form_party" targets: List[RadarTarget] = [] for other in self.players.values(): if other.id == player.id or not other.is_alive or other.health <= 0: continue dist = max(abs(player.x - other.x), abs(player.y - other.y)) is_ally = bool(player.party_id and player.party_id == other.party_id) is_enemy = not is_ally party_obj = self.parties.get(other.party_id) if other.party_id else None if player.party_id and player.is_party_leader and other.party_id is None: can_recruit = other.strength <= player.strength can_battle = other.strength > player.strength elif player.party_id is None and other.party_id is None: can_recruit = True can_battle = False elif player.party_id is None and other.party_id: other_lead = self.players.get(party_obj.leader_id) if party_obj else None lead_str = other_lead.strength if other_lead else 1 can_recruit = player.strength <= lead_str can_battle = player.strength > lead_str else: can_recruit = False can_battle = is_enemy and bool(player.party_id and other.party_id) targets.append( RadarTarget( id=other.id, name=other.name, color=other.color, x=other.x, y=other.y, strength=other.strength, distance=dist, party_id=other.party_id, party_name=party_obj.name if party_obj else None, is_ally=is_ally, is_enemy=is_enemy, can_recruit=can_recruit, can_battle=can_battle, ) ) targets.sort(key=lambda t: t.distance) primary_targets = [] if bot_goal == "form_party": recruit_targets = [t for t in targets if t.can_recruit or not t.party_id] primary_targets = recruit_targets if recruit_targets else targets else: battle_targets = [t for t in targets if t.is_enemy and (t.party_id or t.can_battle)] primary_targets = battle_targets if battle_targets else [t for t in targets if t.is_enemy] nearest = primary_targets[0] if primary_targets else (targets[0] if targets else None) rec_dir = None rec_act = "explore_unvisited" if nearest: # Use BFS pathfinder to recommend direction navigating around obstacles! bfs_path = self._find_path_bfs((player.x, player.y), (nearest.x, nearest.y)) if bfs_path and len(bfs_path) >= 2: step_x, step_y = bfs_path[1] dx = step_x - player.x dy = step_y - player.y else: dx = 1 if nearest.x > player.x else (-1 if nearest.x < player.x else 0) dy = 1 if nearest.y > player.y else (-1 if nearest.y < player.y else 0) for name, (ox, oy) in DIRECTION_OFFSETS.items(): if ox == dx and oy == dy and "_" in name: rec_dir = name break if not rec_dir: for name, (ox, oy) in DIRECTION_OFFSETS.items(): if ox == dx and oy == dy: rec_dir = name break if nearest.distance <= 1: if nearest.can_recruit: rec_act = "form_party" else: rec_act = "engage_battle" else: if bot_goal == "form_party": rec_act = "seek_partner" else: rec_act = "hunt_party" wiz_dist = max(abs(player.x - self.wizard.x), abs(player.y - self.wizard.y)) wiz_radar = WizardRadarTarget( id=self.wizard.id, name=self.wizard.name, x=self.wizard.x, y=self.wizard.y, distance=wiz_dist, strength=self.wizard.strength, can_challenge=(wiz_dist <= 1), ) return BotRadarResponse( player_id=player.id, current_x=player.x, current_y=player.y, bot_goal=bot_goal, targets=targets, nearest_target=nearest, wizard=wiz_radar, recommended_direction=rec_dir, recommended_action=rec_act, ) # ========================================== # Party Logic & Leadership Negotiation # ========================================== @staticmethod def _are_adjacent(p1: Player, p2: Player) -> bool: return max(abs(p1.x - p2.x), abs(p1.y - p2.y)) <= 1 def _verify_party_connectivity(self, member_players: List[Player]) -> bool: 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) def _negotiate_party_leader_id(self, bot_a: Player, bot_b: Player) -> str: if bot_a.strength > bot_b.strength: return bot_a.id elif bot_b.strength > bot_a.strength: return bot_b.id else: if bot_a.score > bot_b.score: return bot_a.id elif bot_b.score > bot_a.score: return bot_b.id return bot_a.id async def form_party( self, member_ids: List[str], leader_id: str, name: Optional[str] = None ) -> Party: async with self._lock: if not self.game_started: raise ValueError("Game has not started yet. Waiting for Start Game button in UI.") 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") if not p.is_alive or p.health <= 0: raise ValueError(f"Player '{p.name}' is dead and cannot join a party.") member_players.append(p) 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)." ) 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), total_strength=sum(p.strength for p in member_players), ) 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") if not inviter.is_alive or inviter.health <= 0: raise ValueError(f"Player '{inviter.name}' is dead and cannot invite players.") if not invitee.is_alive or invitee.health <= 0: raise ValueError(f"Player '{invitee.name}' is dead and cannot be invited.") 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 not inviter.is_alive or not invitee.is_alive: raise ValueError("Cannot join party with deceased player.") 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 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 self._update_party_strength(party) return party 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, total_strength=inviter.strength + invitee.strength, ) 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) def _get_ordered_party_chain(self, party: Party, leader: Player) -> List[Player]: chain = [leader] for mid in party.member_ids: if mid != leader.id and mid in self.players: chain.append(self.players[mid]) return chain def _compute_party_move( self, party: Party, leader: Player, dx: int, dy: int, occupied_map: Dict[Tuple[int, int], Player], ) -> Tuple[Optional[Dict[str, Tuple[int, int]]], Optional[str], float]: target_x = leader.x + dx target_y = leader.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 ( None, f"Hit boundary wall at ({target_x}, {target_y}). Coordinates must remain between {self.config.min_x} and {self.config.max_x}.", 0.0, ) if (target_x, target_y) in self.obstacles: obs = self.obstacles[(target_x, target_y)] return None, f"Target square ({target_x}, {target_y}) is impassable {obs.type} terrain.", 0.0 party_member_ids = set(party.member_ids) occupant = occupied_map.get((target_x, target_y)) if occupant is not None and occupant.id not in party_member_ids: return None, f"Target square ({target_x}, {target_y}) is occupied by player '{occupant.name}'.", 0.0 strength_penalty = 0.0 if dx != 0 and dy != 0: if (leader.x + dx, leader.y) in self.obstacles and (leader.x, leader.y + dy) in self.obstacles: strength_penalty = 0.2 chain = self._get_ordered_party_chain(party, leader) old_positions = {p.id: (p.x, p.y) for p in chain} new_positions: Dict[str, Tuple[int, int]] = {leader.id: (target_x, target_y)} occupied_new: Set[Tuple[int, int]] = {(target_x, target_y)} for i in range(1, len(chain)): curr = chain[i] pred = chain[i - 1] target_pref = old_positions[pred.id] dist_to_pref = max(abs(curr.x - target_pref[0]), abs(curr.y - target_pref[1])) if ( dist_to_pref <= 1 and target_pref not in occupied_new and target_pref not in self.obstacles and ( occupied_map.get(target_pref) is None or occupied_map.get(target_pref).id in party_member_ids ) ): chosen = target_pref else: pred_new = new_positions[pred.id] candidates = [] for cdx in (-1, 0, 1): for cdy in (-1, 0, 1): cand = (curr.x + cdx, curr.y + cdy) if ( self.config.min_x <= cand[0] <= self.config.max_x and self.config.min_y <= cand[1] <= self.config.max_y and cand not in self.obstacles and cand not in occupied_new ): occ = occupied_map.get(cand) if occ is None or occ.id in party_member_ids: if max(abs(cand[0] - pred_new[0]), abs(cand[1] - pred_new[1])) <= 1: candidates.append(cand) if not candidates: for cdx in (-1, 0, 1): for cdy in (-1, 0, 1): cand = (curr.x + cdx, curr.y + cdy) if ( self.config.min_x <= cand[0] <= self.config.max_x and self.config.min_y <= cand[1] <= self.config.max_y and cand not in self.obstacles and cand not in occupied_new ): occ = occupied_map.get(cand) if occ is None or occ.id in party_member_ids: if any( max(abs(cand[0] - pos[0]), abs(cand[1] - pos[1])) <= 1 for pos in new_positions.values() ): candidates.append(cand) if not candidates: return None, f"Party member '{curr.name}' has no available moves to maintain party connection.", 0.0 chosen = min( candidates, key=lambda c: ( max(abs(c[0] - target_pref[0]), abs(c[1] - target_pref[1])), max(abs(c[0] - target_x), abs(c[1] - target_y)), ), ) new_positions[curr.id] = chosen occupied_new.add(chosen) return new_positions, None, strength_penalty # ========================================== # Movement Checking & Execution # ========================================== def _check_move_internal( self, player: Player, dx: int, dy: int, direction_name: str, occupied_map: Dict[Tuple[int, int], Player], ) -> MoveCheckResult: if not player.is_alive or player.health <= 0: return MoveCheckResult( direction=direction_name, dx=dx, dy=dy, target_x=player.x + dx, target_y=player.y + dy, available=False, reason="Player is dead (0 HP)", ) if player.party_id and player.party_id in self.parties and player.is_party_leader: party = self.parties[player.party_id] new_positions, failure_reason, strength_penalty = self._compute_party_move( party, player, dx, dy, occupied_map ) if new_positions is None: return MoveCheckResult( direction=direction_name, dx=dx, dy=dy, target_x=player.x + dx, target_y=player.y + dy, available=False, reason=failure_reason or "Blocked", strength_penalty=0.0, ) return MoveCheckResult( direction=direction_name, dx=dx, dy=dy, target_x=player.x + dx, target_y=player.y + dy, available=True, reason=None, strength_penalty=strength_penalty, ) 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"Hit boundary wall at ({target_x}, {target_y}). Coordinates must remain between {self.config.min_x} and {self.config.max_x}.", ) if (target_x, target_y) in self.obstacles: obs = self.obstacles[(target_x, target_y)] return MoveCheckResult( direction=direction_name, dx=dx, dy=dy, target_x=target_x, target_y=target_y, available=False, reason=f"Target square ({target_x}, {target_y}) is impassable {obs.type} terrain.", ) 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"Target square ({target_x}, {target_y}) is occupied by player '{occupant.name}'.", ) strength_penalty = 0.0 if dx != 0 and dy != 0: if (player.x + dx, player.y) in self.obstacles and (player.x, player.y + dy) in self.obstacles: strength_penalty = 0.1 return MoveCheckResult( direction=direction_name, dx=dx, dy=dy, target_x=target_x, target_y=target_y, available=True, reason=None, strength_penalty=strength_penalty, ) async def check_single_move(self, player_id: str, direction_name: 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_name.strip().upper().replace(" ", "_") if normalized not in DIRECTION_OFFSETS: raise ValueError(f"Unknown direction '{direction_name}'") 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") if not player.is_alive or player.health <= 0: moves = { name: MoveCheckResult( direction=name, dx=dx, dy=dy, target_x=player.x + dx, target_y=player.y + dy, available=False, reason="Player is dead (0 HP)", ) for name, dx, dy in STANDARD_DIRECTIONS } return AvailableMovesResponse( player_id=player.id, player_name=player.name, current_x=player.x, current_y=player.y, is_turn=False, is_party_leader=False, party_id=None, party_member_count=0, current_turn_player_id=None, moves=moves, ) occupied = self._get_occupied_coordinates() current_turn = self._get_current_player() is_turn = bool(current_turn and current_turn.id == player_id) moves: Dict[str, MoveCheckResult] = {} for name, dx, dy in STANDARD_DIRECTIONS: if not self.game_started: moves[name] = MoveCheckResult( direction=name, dx=dx, dy=dy, target_x=player.x + dx, target_y=player.y + dy, available=False, reason="Game has not started yet. Waiting for Start Game button in UI.", ) else: moves[name] = self._check_move_internal(player, dx, dy, name, occupied) member_count = 1 if player.party_id and player.party_id in self.parties: member_count = len(self.parties[player.party_id].member_ids) return AvailableMovesResponse( player_id=player.id, player_name=player.name, current_x=player.x, current_y=player.y, is_turn=is_turn, is_party_leader=player.is_party_leader, party_id=player.party_id, party_member_count=member_count, current_turn_player_id=current_turn.id if current_turn else None, moves=moves, ) def _update_party_strength(self, party: Party): total = 0.0 for mid in party.member_ids: p = self.players.get(mid) if p: total += p.strength party.total_strength = max(0.1, round(total, 1)) def _resolve_3bout_battle_internal( self, party1: Party, party2: Party ) -> BattleResult: bouts: List[BattleBout] = [] p1_bouts_won = 0 p2_bouts_won = 0 p1_total_score = 0.0 p2_total_score = 0.0 self._update_party_strength(party1) self._update_party_strength(party2) for bout_idx in range(1, 4): r1 = random.randint(1, 20) r2 = random.randint(1, 20) score1 = round(party1.total_strength * r1, 1) score2 = round(party2.total_strength * r2, 1) p1_total_score += score1 p2_total_score += score2 if score1 > score2: winner_name = party1.name p1_bouts_won += 1 elif score2 > score1: winner_name = party2.name p2_bouts_won += 1 else: if party1.total_strength >= party2.total_strength: winner_name = party1.name p1_bouts_won += 1 else: winner_name = party2.name p2_bouts_won += 1 bouts.append( BattleBout( bout_number=bout_idx, party1_roll=r1, party1_strength=party1.total_strength, party1_score=score1, party2_roll=r2, party2_strength=party2.total_strength, party2_score=score2, winner_name=winner_name, ) ) if p1_bouts_won > p2_bouts_won: winner_party = party1 defeated_party = party2 elif p2_bouts_won > p1_bouts_won: winner_party = party2 defeated_party = party1 else: if p1_total_score >= p2_total_score: winner_party = party1 defeated_party = party2 else: winner_party = party2 defeated_party = party1 winner_lead = self.players.get(winner_party.leader_id) if winner_lead: winner_lead.score += 2 for mid in winner_party.member_ids: if mid != winner_party.leader_id and mid in self.players: self.players[mid].score += 1 killed_leader = self.players.get(defeated_party.leader_id) absorbed_members: List[str] = [] dead_players: List[str] = [] # Defeated party members (including leader) all lose 1 to 3 health points (randomized) health_losses: Dict[str, int] = {} defeated_all_ids = list(defeated_party.member_ids) if defeated_party.leader_id and defeated_party.leader_id not in defeated_all_ids: defeated_all_ids.append(defeated_party.leader_id) for mid in defeated_all_ids: p = self.players.get(mid) if p: hp_loss = random.randint(1, 3) p.health = max(0, p.health - hp_loss) health_losses[mid] = hp_loss if p.health == 0 and p.is_alive: self._check_and_apply_death(p) dead_players.append(mid) if killed_leader: killed_leader.score -= 1 if killed_leader.is_alive: if len(defeated_party.member_ids) == 1: killed_leader.party_id = winner_party.id killed_leader.is_party_leader = False if killed_leader.id not in winner_party.member_ids: winner_party.member_ids.append(killed_leader.id) absorbed_members.append(killed_leader.id) respawn_pos = {"x": killed_leader.x, "y": killed_leader.y} else: respawn_x, respawn_y = self._find_random_free_position() killed_leader.x = respawn_x killed_leader.y = respawn_y killed_leader.party_id = None killed_leader.is_party_leader = False respawn_pos = {"x": respawn_x, "y": respawn_y} else: # Leader is dead: remains at final coordinates as gravestone, disconnected from party respawn_pos = {"x": killed_leader.x, "y": killed_leader.y} # Only surviving defeated party followers are absorbed into winning party for mid in list(defeated_party.member_ids): if mid != defeated_party.leader_id: m = self.players.get(mid) if m and m.is_alive: m.party_id = winner_party.id m.is_party_leader = False if m.id not in winner_party.member_ids: winner_party.member_ids.append(m.id) absorbed_members.append(m.id) if defeated_party.id in self.parties: del self.parties[defeated_party.id] self._update_party_strength(winner_party) return BattleResult( bouts=bouts, party1_name=party1.name, party2_name=party2.name, party1_bouts_won=p1_bouts_won, party2_bouts_won=p2_bouts_won, party1_total_score=p1_total_score, party2_total_score=p2_total_score, winner_party_id=winner_party.id, winner_party_name=winner_party.name, winner_leader_id=winner_party.leader_id, winner_leader_name=winner_party.leader_name, defeated_party_id=defeated_party.id, defeated_party_name=defeated_party.name, killed_leader_id=killed_leader.id if killed_leader else "", killed_leader_name=killed_leader.name if killed_leader else "Unknown", killed_leader_new_score=killed_leader.score if killed_leader else 0, killed_leader_respawn_position=respawn_pos, absorbed_members=absorbed_members, new_party_size=len(winner_party.member_ids), new_party_strength=winner_party.total_strength, health_losses=health_losses, dead_players=dead_players, ) async def fight_battle(self, challenger_id: str, defender_id: str) -> BattleResult: async with self._lock: if not self.game_started: raise ValueError("Game has not started yet. Waiting for Start Game button in UI.") p1 = self.players.get(challenger_id) p2 = self.players.get(defender_id) if not p1 or not p2: raise KeyError("Challenger or defender not found") if not p1.is_alive or p1.health <= 0: raise ValueError(f"Challenger '{p1.name}' is dead and cannot battle.") if not p2.is_alive or p2.health <= 0: raise ValueError(f"Defender '{p2.name}' is dead and cannot battle.") if p1.party_id and p1.party_id == p2.party_id: raise ValueError("Cannot battle members of your own party") if p1.party_id and not p1.is_party_leader: raise PermissionError("Only party leader can initiate a battle for the party") party1 = self.parties.get(p1.party_id) if p1.party_id else None party2 = self.parties.get(p2.party_id) if p2.party_id else None if not party1: temp_p1_id = f"party_{uuid.uuid4().hex[:8]}" party1 = Party( id=temp_p1_id, name=f"Squad {p1.name}", leader_id=p1.id, leader_name=p1.name, member_ids=[p1.id], total_strength=p1.strength, ) p1.party_id = temp_p1_id p1.is_party_leader = True self.parties[temp_p1_id] = party1 if not party2: temp_p2_id = f"party_{uuid.uuid4().hex[:8]}" party2 = Party( id=temp_p2_id, name=f"Squad {p2.name}", leader_id=p2.id, leader_name=p2.name, member_ids=[p2.id], total_strength=p2.strength, ) p2.party_id = temp_p2_id p2.is_party_leader = True self.parties[temp_p2_id] = party2 result = self._resolve_3bout_battle_internal(party1, party2) self._advance_turn() return result async def battle(self, challenger_id: str, defender_id: str) -> BattleResult: return await self.fight_battle(challenger_id, defender_id) def _resolve_wizard_challenge_internal( self, player: Player, reward_choice: str = "score" ) -> WizardChallengeResult: """Resolve a 3-bout D20 challenge between player/party and the Wizard NPC.""" effective_strength = player.strength party_id = player.party_id if party_id and party_id in self.parties: party = self.parties[party_id] self._update_party_strength(party) effective_strength = party.total_strength bouts: List[WizardChallengeBout] = [] player_bouts_won = 0 wizard_bouts_won = 0 for bout_idx in range(1, 4): r_player = random.randint(1, 20) r_wiz = random.randint(1, 20) score_player = round(effective_strength * r_player, 1) score_wiz = round(self.wizard.strength * r_wiz, 1) if score_player > score_wiz: winner = "player" player_bouts_won += 1 elif score_wiz > score_player: winner = "wizard" wizard_bouts_won += 1 else: if effective_strength >= self.wizard.strength: winner = "player" player_bouts_won += 1 else: winner = "wizard" wizard_bouts_won += 1 bouts.append( WizardChallengeBout( bout_number=bout_idx, player_roll=r_player, player_strength=effective_strength, player_score=score_player, wizard_roll=r_wiz, wizard_strength=self.wizard.strength, wizard_score=score_wiz, winner=winner, ) ) player_won = player_bouts_won >= 2 score_change = 0 strength_change = 0.0 health_change = 0 choice = (reward_choice or "score").lower().strip() if choice not in ("score", "strength", "health"): choice = "score" if player_won: # Player wins challenge: choice of +2 score, +2 strength, or +2 health if choice == "strength": player.strength = round(player.strength + 2.0, 1) strength_change = 2.0 if player.party_id and player.party_id in self.parties: self._update_party_strength(self.parties[player.party_id]) elif choice == "health": player.health += 2 if player.health > player.max_health: player.max_health = player.health health_change = 2 else: # "score" player.score += 2 score_change = 2 player_died = False if not player_won: # Player loses challenge: lose 2 health (or points if they do not have health to lose) if player.health >= 2: player.health -= 2 health_change = -2 score_change = 0 elif player.health > 0: pts_lost = 2 - player.health health_change = -player.health player.health = 0 player.score -= pts_lost score_change = -pts_lost else: health_change = 0 player.score -= 2 score_change = -2 if player.health == 0 and player.is_alive: player_died = self._check_and_apply_death(player) # Wizard teleports to a new random open coordinate on the map new_wx, new_wy = self._find_random_free_position() self.wizard.x = new_wx self.wizard.y = new_wy respawn_pos = {"x": new_wx, "y": new_wy} self._advance_turn() return WizardChallengeResult( challenger_id=player.id, challenger_name=player.name, wizard_name=self.wizard.name if self.wizard else "Gary the Wizard", party_id=party_id, bouts=bouts, player_bouts_won=player_bouts_won, wizard_bouts_won=wizard_bouts_won, player_won=player_won, reward_chosen=choice if player_won else None, score_change=score_change, strength_change=strength_change, health_change=health_change, new_score=player.score, new_strength=player.strength, new_health=player.health, player_died=player_died, wizard_respawn_position=respawn_pos, ) async def challenge_wizard(self, player_id: str, reward_choice: str = "score") -> WizardChallengeResult: async with self._lock: if not self.game_started: raise ValueError("Game has not started yet. Waiting for Start Game button in UI.") player = self.players.get(player_id) if not player: raise KeyError(f"Player '{player_id}' not found") if not player.is_alive or player.health <= 0: raise ValueError(f"Player '{player.name}' is dead and cannot challenge the wizard.") 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})." ) 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 challenge the wizard individually. Only party leader '{leader_name}' can initiate challenges." ) dist = max(abs(player.x - self.wizard.x), abs(player.y - self.wizard.y)) if dist > 1: wiz_name = self.wizard.name if self.wizard else "Gary the Wizard" raise ValueError( f"Player '{player.name}' is not adjacent to {wiz_name} (distance {dist}). Must be within 1 distance." ) return self._resolve_wizard_challenge_internal(player, reward_choice=reward_choice) async def get_game_conclusion(self) -> GameConclusion: async with self._lock: return self._check_game_concluded() def _check_game_concluded(self) -> GameConclusion: if len(self.players) < 2: return GameConclusion(concluded=False) living_players = [p for p in self.players.values() if p.is_alive] total_bots = len(self.players) rankings = sorted( self.players.values(), key=lambda p: (p.score, p.strength, p.name), reverse=True, ) # Case 1: All bots died if len(living_players) == 0: return GameConclusion( concluded=True, total_bots=total_bots, rankings=rankings, ) # Case 2: Exactly 1 living bot remains (sole survivor) if len(living_players) == 1: survivor = living_players[0] party = self.parties.get(survivor.party_id) if survivor.party_id else None return GameConclusion( concluded=True, winning_party_id=party.id if party else None, winning_party_name=party.name if party else f"Squad {survivor.name}", winning_leader_id=party.leader_id if party else survivor.id, winning_leader_name=party.leader_name if party else survivor.name, total_bots=total_bots, rankings=rankings, ) # Case 3: All living bots are united into a single remaining party if len(self.parties) == 1: only_party = next(iter(self.parties.values())) living_member_ids = { mid for mid in only_party.member_ids if self.players.get(mid) and self.players[mid].is_alive } living_player_ids = {p.id for p in living_players} if living_member_ids == living_player_ids and len(living_player_ids) >= 1: return GameConclusion( concluded=True, winning_party_id=only_party.id, winning_party_name=only_party.name, winning_leader_id=only_party.leader_id, winning_leader_name=only_party.leader_name, total_bots=total_bots, rankings=rankings, ) return GameConclusion(concluded=False) def _check_adjacent_encounter( self, player: Player ) -> Tuple[Optional[Party], Optional[BattleResult]]: if not player.is_alive or player.health <= 0: return None, None for other in self.players.values(): if other.id == player.id: continue if not other.is_alive or other.health <= 0: continue if player.party_id and player.party_id == other.party_id: continue if not self._are_adjacent(player, other): continue # Case A: Solo + Solo encounter if not player.party_id and not other.party_id: leader_id = self._negotiate_party_leader_id(player, other) members = [player.id, other.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, total_strength=player.strength + other.strength, ) player.party_id = party_id player.is_party_leader = player.id == leader_id other.party_id = party_id other.is_party_leader = other.id == leader_id self.parties[party_id] = party return party, None # Case B: Party + Opposing Party encounter if player.party_id and other.party_id and player.party_id != other.party_id: if player.is_party_leader: p1 = self.parties[player.party_id] p2 = self.parties[other.party_id] battle_res = self._resolve_3bout_battle_internal(p1, p2) return None, battle_res # Case C: Party Leader encounters Solo Bot if player.party_id and player.is_party_leader and not other.party_id: if other.strength <= player.strength: other.party_id = player.party_id other.is_party_leader = False party = self.parties[player.party_id] if other.id not in party.member_ids: party.member_ids.append(other.id) self._update_party_strength(party) return party, None else: solo_party_id = f"party_{uuid.uuid4().hex[:8]}" solo_party = Party( id=solo_party_id, name=f"Squad {other.name}", leader_id=other.id, leader_name=other.name, member_ids=[other.id], total_strength=other.strength, ) other.party_id = solo_party_id other.is_party_leader = True self.parties[solo_party_id] = solo_party active_party = self.parties[player.party_id] battle_res = self._resolve_3bout_battle_internal(active_party, solo_party) return None, battle_res # Case D: Solo Bot encounters Party if not player.party_id and other.party_id: other_party = self.parties.get(other.party_id) if other_party: leader_player = self.players.get(other_party.leader_id) leader_strength = leader_player.strength if leader_player else other_party.total_strength if player.strength <= leader_strength: player.party_id = other_party.id player.is_party_leader = False if player.id not in other_party.member_ids: other_party.member_ids.append(player.id) self._update_party_strength(other_party) return other_party, None else: solo_party_id = f"party_{uuid.uuid4().hex[:8]}" solo_party = Party( id=solo_party_id, name=f"Squad {player.name}", leader_id=player.id, leader_name=player.name, member_ids=[player.id], total_strength=player.strength, ) player.party_id = solo_party_id player.is_party_leader = True self.parties[solo_party_id] = solo_party battle_res = self._resolve_3bout_battle_internal(other_party, solo_party) return None, battle_res return None, None def _check_and_auto_form_party(self, player: Player) -> Optional[Party]: party, _ = self._check_adjacent_encounter(player) return party async def move_player( self, player_id: str, dx: int, dy: int, direction_name: str ) -> MoveResponse: async with self._lock: if not self.game_started: raise ValueError("Game has not started yet. Waiting for Start Game button in UI.") player = self.players.get(player_id) if not player: raise KeyError(f"Player '{player_id}' not found") if not player.is_alive or player.health <= 0: raise ValueError(f"Player '{player.name}' is dead and 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] = [] # Move party in follow-the-leader chain (forming a line) or single bot if player.party_id and player.party_id in self.parties: party = self.parties[player.party_id] new_positions, failure_reason, _ = self._compute_party_move( party, player, dx, dy, occupied ) if not new_positions: raise ValueError(f"Illegal move: {failure_reason or 'Blocked'}") chain = self._get_ordered_party_chain(party, player) for m in chain: old_mx, old_my = m.x, m.y new_mx, new_my = new_positions[m.id] step_dx = new_mx - old_mx step_dy = new_my - old_my m.x = new_mx m.y = new_my m.visited_locations.append({"x": m.x, "y": m.y}) if step_dx != 0 and step_dy != 0: if (old_mx + step_dx, old_my) in self.obstacles and (old_mx, old_my + step_dy) in self.obstacles: penalty = 0.2 if m.is_party_leader else 0.1 m.strength = round(max(0.1, m.strength - penalty), 1) affected_players.append(m) self._update_party_strength(party) else: player.x = check.target_x player.y = check.target_y player.visited_locations.append({"x": player.x, "y": player.y}) if check.strength_penalty > 0: player.strength = round(max(0.1, player.strength - 0.1), 1) affected_players.append(player) new_pos = {"x": player.x, "y": player.y} # Check for adjacent interactions (party formation or battle) formed_party, battle_result = self._check_adjacent_encounter(player) self._advance_turn() turn_info = self._get_turn_info() conclusion = self._check_game_concluded() 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, party_formed_triggered=bool(formed_party), formed_party=formed_party, battle_triggered=bool(battle_result), battle_result=battle_result, game_concluded=conclusion if conclusion.concluded else None, turn=turn_info, ) async def pass_turn(self, player_id: str) -> TurnInfo: async with self._lock: if not self.game_started: raise ValueError("Game has not started yet. Waiting for Start Game button in UI.") player = self.players.get(player_id) if not player: raise KeyError(f"Player '{player_id}' not found") if not player.is_alive or player.health <= 0: raise ValueError(f"Player '{player.name}' is dead and has no actions.") 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 to pass. Current turn belongs to '{curr_name}' ({curr_id})." ) self._advance_turn() return self._get_turn_info() async def step_bot_ai(self, player_id: str) -> AiStepResponse: async with self._lock: if not self.game_started: raise ValueError("Game has not started yet. Waiting for Start Game button in UI.") player = self.players.get(player_id) if not player: raise KeyError(f"Player '{player_id}' not found") if not player.is_alive or player.health <= 0: raise ValueError(f"Player '{player.name}' is dead and cannot take actions.") 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})." ) bot_goal = "find_and_defeat_all_parties" if player.party_id else "form_party" # 1. Check if already adjacent to encounter before moving formed_party, battle_res = self._check_adjacent_encounter(player) if battle_res: self._advance_turn() conclusion = self._check_game_concluded() return AiStepResponse( action_taken="battled", player_id=player.id, player_name=player.name, bot_goal=bot_goal, direction=None, move_result=None, formed_party=None, battle_result=battle_res, game_concluded=conclusion if conclusion.concluded else None, turn=self._get_turn_info(), ) if formed_party: self._advance_turn() conclusion = self._check_game_concluded() return AiStepResponse( action_taken="formed_party", player_id=player.id, player_name=player.name, bot_goal=bot_goal, direction=None, move_result=None, formed_party=formed_party, battle_result=None, wizard_challenge_result=None, game_concluded=conclusion if conclusion.concluded else None, turn=self._get_turn_info(), ) # 1b. Check if adjacent to Gary the Wizard NPC and choose to challenge wiz_dist = max(abs(player.x - self.wizard.x), abs(player.y - self.wizard.y)) if wiz_dist <= 1 and (not player.party_id or player.is_party_leader): effective_str = player.strength if player.party_id and player.party_id in self.parties: effective_str = self.parties[player.party_id].total_strength if effective_str >= self.wizard.strength or player.health >= 4: if player.health <= 5: bot_reward = "health" elif player.strength < 4.0: bot_reward = "strength" else: bot_reward = "score" challenge_res = self._resolve_wizard_challenge_internal(player, reward_choice=bot_reward) conclusion = self._check_game_concluded() return AiStepResponse( action_taken="challenged_wizard", player_id=player.id, player_name=player.name, bot_goal=bot_goal, direction=None, move_result=None, formed_party=None, battle_result=None, wizard_challenge_result=challenge_res, game_concluded=conclusion if conclusion.concluded else None, turn=self._get_turn_info(), ) # 2. Navigate towards target using BFS pathfinder & memory occupied = self._get_occupied_coordinates() moves_map: Dict[str, MoveCheckResult] = {} for name, dx, dy in STANDARD_DIRECTIONS: moves_map[name] = self._check_move_internal(player, dx, dy, name, occupied) available_dirs = [name for name, chk in moves_map.items() if chk.available] if not available_dirs: self._advance_turn() conclusion = self._check_game_concluded() return AiStepResponse( action_taken="passed", player_id=player.id, player_name=player.name, bot_goal=bot_goal, direction=None, move_result=None, formed_party=None, battle_result=None, game_concluded=conclusion if conclusion.concluded else None, turn=self._get_turn_info(), ) # Find nearest target according to goal targets = [] for other in self.players.values(): if other.id == player.id or not other.is_alive or other.health <= 0: continue if player.party_id and player.party_id == other.party_id: continue dist = max(abs(player.x - other.x), abs(player.y - other.y)) targets.append((dist, other)) targets.sort(key=lambda t: t[0]) if bot_goal == "form_party": preferred = [t for t in targets if not t[1].party_id] target_bot = preferred[0][1] if preferred else (targets[0][1] if targets else None) else: preferred = [t for t in targets if t[1].party_id] target_bot = preferred[0][1] if preferred else (targets[0][1] if targets else None) # Check if BFS pathfinder finds optimal route around obstacles bfs_next_step: Optional[Tuple[int, int]] = None if target_bot: path = self._find_path_bfs((player.x, player.y), (target_bot.x, target_bot.y)) if path and len(path) >= 2: bfs_next_step = (path[1][0] - player.x, path[1][1] - player.y) visited_set = {(loc["x"], loc["y"]) for loc in player.visited_locations} best_dir = available_dirs[0] best_score = float("-inf") for dir_name in available_dirs: dx, dy = DIRECTION_OFFSETS[dir_name] tx = player.x + dx ty = player.y + dy score = 0.0 # Massive bonus if this step aligns with the shortest BFS path around obstacles! if bfs_next_step and (dx, dy) == bfs_next_step: score += 50.0 if target_bot: old_dist = max(abs(player.x - target_bot.x), abs(player.y - target_bot.y)) new_dist = max(abs(tx - target_bot.x), abs(ty - target_bot.y)) score += (old_dist - new_dist) * 10.0 if (tx, ty) not in visited_set: score += 2.0 if score > best_score: best_score = score best_dir = dir_name dx, dy = DIRECTION_OFFSETS[best_dir] prev_pos = {"x": player.x, "y": player.y} affected_players: List[Player] = [] 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 m.visited_locations.append({"x": m.x, "y": m.y}) affected_players.append(m) else: player.x += dx player.y += dy player.visited_locations.append({"x": player.x, "y": player.y}) affected_players.append(player) new_pos = {"x": player.x, "y": player.y} # Check for adjacent interactions after movement formed_party, battle_result = self._check_adjacent_encounter(player) self._advance_turn() turn_info = self._get_turn_info() conclusion = self._check_game_concluded() move_res = MoveResponse( success=True, player=player, direction=best_dir, party_moved=bool(player.party_id), affected_players=affected_players, previous_position=prev_pos, new_position=new_pos, party_formed_triggered=bool(formed_party), formed_party=formed_party, battle_triggered=bool(battle_result), battle_result=battle_result, game_concluded=conclusion if conclusion.concluded else None, turn=turn_info, ) action_name = "moved" if battle_result: action_name = "battled" elif formed_party: action_name = "formed_party" return AiStepResponse( action_taken=action_name, player_id=player.id, player_name=player.name, bot_goal=bot_goal, direction=best_dir, move_result=move_res, formed_party=formed_party, battle_result=battle_result, game_concluded=conclusion if conclusion.concluded else None, turn=turn_info, ) async def defeat_party(self, party_id: str) -> PartyDefeatResult: async with self._lock: if party_id not in self.parties: raise KeyError(f"Party '{party_id}' not found") party = self.parties[party_id] leader = self.players.get(party.leader_id) leader_id = party.leader_id leader_name = party.leader_name if leader: leader.score -= 1 hp_loss = random.randint(1, 3) leader.health = max(0, leader.health - hp_loss) if leader.health == 0 and leader.is_alive: self._check_and_apply_death(leader) respawn_pos = {"x": leader.x, "y": leader.y} else: rx, ry = self._find_random_free_position() leader.x = rx leader.y = ry leader.party_id = None leader.is_party_leader = False respawn_pos = {"x": rx, "y": ry} else: respawn_pos = {"x": 0, "y": 0} if leader_id in party.member_ids: party.member_ids.remove(leader_id) remaining_alive = [ mid for mid in party.member_ids if mid in self.players and self.players[mid].is_alive ] new_leader_id = None new_leader_name = None party_dissolved = False if remaining_alive: candidates = [self.players[mid] for mid in remaining_alive] candidates.sort(key=lambda p: (p.strength, p.score), reverse=True) new_lead = candidates[0] party.leader_id = new_lead.id party.leader_name = new_lead.name new_lead.is_party_leader = True new_leader_id = new_lead.id new_leader_name = new_lead.name self._update_party_strength(party) else: party_dissolved = True if party.id in self.parties: del self.parties[party.id] self._advance_turn() return PartyDefeatResult( party_id=party_id, killed_leader_id=leader_id, killed_leader_name=leader_name, killed_leader_new_score=leader.score if leader else 0, killed_leader_respawn_position=respawn_pos, new_leader_id=new_leader_id, new_leader_name=new_leader_name, remaining_members=remaining_alive, party_dissolved=party_dissolved, ) # Global game engine instance game_engine = GameEngine()