1539 lines
60 KiB
Python
1539 lines
60 KiB
Python
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,
|
|
)
|
|
|
|
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()
|
|
|
|
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 _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:
|
|
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()
|
|
|
|
player = Player(
|
|
id=player_id,
|
|
name=player_in.name,
|
|
color=player_in.color,
|
|
strength=player_in.strength,
|
|
score=0,
|
|
x=spawn_x,
|
|
y=spawn_y,
|
|
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()
|
|
|
|
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,
|
|
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:
|
|
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"
|
|
|
|
return BotRadarResponse(
|
|
player_id=player.id,
|
|
current_x=player.x,
|
|
current_y=player.y,
|
|
bot_goal=bot_goal,
|
|
targets=targets,
|
|
nearest_target=nearest,
|
|
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")
|
|
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")
|
|
|
|
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.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)
|
|
|
|
# ==========================================
|
|
# 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 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
|
|
|
|
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}).",
|
|
)
|
|
|
|
if (tx, ty) in self.obstacles:
|
|
obs = self.obstacles[(tx, ty)]
|
|
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}' blocked by impassable {obs.type} at ({tx}, {ty}).",
|
|
)
|
|
|
|
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,
|
|
)
|
|
|
|
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}'.",
|
|
)
|
|
|
|
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_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")
|
|
|
|
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
|
|
for mid in party.member_ids:
|
|
p = self.players.get(mid)
|
|
if p:
|
|
total += p.strength
|
|
party.total_strength = max(1, total)
|
|
|
|
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
|
|
p2_total_score = 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 = party1.total_strength * r1
|
|
score2 = party2.total_strength * r2
|
|
|
|
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] = []
|
|
|
|
if killed_leader:
|
|
killed_leader.score -= 1
|
|
|
|
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}
|
|
|
|
for mid in list(defeated_party.member_ids):
|
|
if mid != defeated_party.leader_id:
|
|
m = self.players.get(mid)
|
|
if m:
|
|
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,
|
|
)
|
|
|
|
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 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)
|
|
|
|
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)
|
|
|
|
if len(self.parties) == 1:
|
|
only_party = next(iter(self.parties.values()))
|
|
if len(only_party.member_ids) == len(self.players):
|
|
rankings = sorted(
|
|
self.players.values(),
|
|
key=lambda p: (p.score, p.strength, p.name),
|
|
reverse=True,
|
|
)
|
|
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=len(self.players),
|
|
rankings=rankings,
|
|
)
|
|
|
|
return GameConclusion(concluded=False)
|
|
|
|
def _check_adjacent_encounter(
|
|
self, player: Player
|
|
) -> Tuple[Optional[Party], Optional[BattleResult]]:
|
|
for other in self.players.values():
|
|
if other.id == player.id:
|
|
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 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 unison or single bot
|
|
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 = check.target_x
|
|
player.y = check.target_y
|
|
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 (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")
|
|
|
|
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")
|
|
|
|
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,
|
|
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:
|
|
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,
|
|
)
|
|
|
|
|
|
# Global game engine instance
|
|
game_engine = GameEngine()
|