|
from typing import Dict, List, Optional, Set, Tuple
|
|
from server.models import Player, Building, BuildingType, BUILDING_CONFIGS
|
|
import time
|
|
from server.logger import logger
|
|
|
|
class GameState:
|
|
"""Manages the complete game state"""
|
|
|
|
def __init__(self):
|
|
self.players: Dict[str, Player] = {}
|
|
self.buildings: Dict[Tuple[int, int], Building] = {}
|
|
self.road_network: Set[Tuple[int, int]] = set()
|
|
self.connected_zones: List[Set[Tuple[int, int]]] = []
|
|
|
|
def get_or_create_player(self, nickname: str, player_id: str) -> Player:
|
|
"""Get existing player or create new one"""
|
|
if player_id in self.players:
|
|
player = self.players[player_id]
|
|
player.is_online = True
|
|
player.last_online = time.time()
|
|
logger.debug(f"Returning existing player '{nickname}' ({player_id}).")
|
|
return player
|
|
|
|
player = Player(
|
|
player_id=player_id,
|
|
nickname=nickname,
|
|
last_online=time.time()
|
|
)
|
|
self.players[player_id] = player
|
|
logger.info(f"Created new player '{nickname}' ({player_id}) with ${player.money} starting money.")
|
|
return player
|
|
|
|
def place_building(self, player_id: str, building_type: str, x: int, y: int) -> dict:
|
|
"""Place a building on the map"""
|
|
logger.debug(f"Attempting to place '{building_type}' at ({x},{y}) for player {player_id}.")
|
|
|
|
if (x, y) in self.buildings:
|
|
logger.warning(f"Placement failed for {player_id}: Tile ({x},{y}) already occupied.")
|
|
return {"success": False, "error": "Tile already occupied"}
|
|
|
|
player = self.players.get(player_id)
|
|
if not player:
|
|
logger.error(f"Placement failed: Player {player_id} not found.")
|
|
return {"success": False, "error": "Player not found"}
|
|
|
|
try:
|
|
b_type = BuildingType(building_type)
|
|
config = BUILDING_CONFIGS[b_type]
|
|
except (ValueError, KeyError):
|
|
logger.error(f"Placement failed for {player_id}: Invalid building type '{building_type}'.")
|
|
return {"success": False, "error": "Invalid building type"}
|
|
|
|
if not player.can_afford(config.cost):
|
|
logger.debug(f"Placement failed for {player_id}: Cannot afford '{building_type}'. Cost: {config.cost}, Player Money: {player.money}.")
|
|
return {"success": False, "error": "Not enough money"}
|
|
|
|
if config.requires_population > player.population:
|
|
logger.debug(f"Placement failed for {player_id}: '{building_type}' requires {config.requires_population} pop, has {player.population}.")
|
|
return {"success": False, "error": f"Requires {config.requires_population} population"}
|
|
|
|
if config.power_required and not self._has_power_plant(player_id):
|
|
logger.debug(f"Placement failed for {player_id}: '{building_type}' requires a power plant.")
|
|
return {"success": False, "error": "Requires power plant"}
|
|
|
|
building = Building(building_type=b_type, x=x, y=y, owner_id=player_id, placed_at=time.time())
|
|
|
|
self.buildings[(x, y)] = building
|
|
player.deduct_money(config.cost)
|
|
|
|
logger.info(f"Player {player_id} placed '{building_type}' at ({x},{y}). New balance: ${player.money}.")
|
|
|
|
if b_type == BuildingType.ROAD:
|
|
self.road_network.add((x, y))
|
|
self._update_connected_zones()
|
|
|
|
return {"success": True, "building": building.to_dict()}
|
|
|
|
def remove_building(self, player_id: str, x: int, y: int) -> dict:
|
|
"""Remove a building"""
|
|
logger.debug(f"Attempting to remove building at ({x},{y}) for player {player_id}.")
|
|
building = self.buildings.get((x, y))
|
|
|
|
if not building:
|
|
logger.warning(f"Removal failed for {player_id}: No building at ({x},{y}).")
|
|
return {"success": False, "error": "No building at this location"}
|
|
|
|
if building.owner_id != player_id:
|
|
logger.warning(f"Removal failed for {player_id}: Does not own building at ({x},{y}). Owner is {building.owner_id}.")
|
|
return {"success": False, "error": "You don't own this building"}
|
|
|
|
del self.buildings[(x, y)]
|
|
logger.info(f"Player {player_id} removed '{building.building_type.value}' from ({x},{y}).")
|
|
|
|
if building.building_type == BuildingType.ROAD:
|
|
self.road_network.discard((x, y))
|
|
self._update_connected_zones()
|
|
|
|
return {"success": True}
|
|
|
|
def edit_building_name(self, player_id: str, x: int, y: int, name: str) -> dict:
|
|
"""Edit building name"""
|
|
building = self.buildings.get((x, y))
|
|
if not building:
|
|
return {"success": False, "error": "No building at this location"}
|
|
if building.owner_id != player_id:
|
|
return {"success": False, "error": "You don't own this building"}
|
|
|
|
building.name = name
|
|
logger.info(f"Player {player_id} renamed building at ({x},{y}) to '{name}'.")
|
|
return {"success": True}
|
|
|
|
def _has_power_plant(self, player_id: str) -> bool:
|
|
"""Check if player has a power plant"""
|
|
for building in self.buildings.values():
|
|
if (building.owner_id == player_id and building.building_type == BuildingType.POWER_PLANT):
|
|
return True
|
|
return False
|
|
|
|
def _update_connected_zones(self):
|
|
"""Update connected zones based on road network using flood fill"""
|
|
if not self.road_network:
|
|
self.connected_zones = []
|
|
return
|
|
|
|
logger.debug("Recalculating road network connected zones...")
|
|
start_time = time.perf_counter()
|
|
|
|
visited = set()
|
|
self.connected_zones = []
|
|
|
|
for road_pos in self.road_network:
|
|
if road_pos in visited:
|
|
continue
|
|
|
|
zone = set()
|
|
stack = [road_pos]
|
|
while stack:
|
|
pos = stack.pop()
|
|
if pos in visited: continue
|
|
visited.add(pos)
|
|
zone.add(pos)
|
|
x, y = pos
|
|
for dx, dy in [(0, 1), (1, 0), (0, -1), (-1, 0)]:
|
|
adj_pos = (x + dx, y + dy)
|
|
if adj_pos in self.road_network and adj_pos not in visited:
|
|
stack.append(adj_pos)
|
|
|
|
self.connected_zones.append(zone)
|
|
|
|
duration = time.perf_counter() - start_time
|
|
logger.debug(f"Found {len(self.connected_zones)} road zones in {duration:.4f} seconds.")
|
|
|
|
def get_building_zone_size(self, x: int, y: int) -> int:
|
|
"""Get the size of the connected zone for a building"""
|
|
for dx, dy in [(0, 1), (1, 0), (0, -1), (-1, 0), (1, 1), (-1, 1), (1, -1), (-1, -1)]:
|
|
road_pos = (x + dx, y + dy)
|
|
if road_pos in self.road_network:
|
|
for zone in self.connected_zones:
|
|
if road_pos in zone:
|
|
return len(zone)
|
|
return 0
|
|
|
|
def get_player_buildings(self, player_id: str) -> List[Building]:
|
|
"""Get all buildings owned by a player"""
|
|
return [b for b in self.buildings.values() if b.owner_id == player_id]
|
|
|
|
def get_state(self) -> dict:
|
|
"""Get complete game state for broadcasting"""
|
|
return {
|
|
"players": {pid: p.to_dict() for pid, p in self.players.items()},
|
|
"buildings": {f"{x},{y}": b.to_dict() for (x, y), b in self.buildings.items()}
|
|
}
|
|
|
|
def load_state(self, state: dict):
|
|
"""Load game state from database"""
|
|
if not state:
|
|
return
|
|
|
|
for player_data in state.get("players", []):
|
|
player = Player(**player_data)
|
|
player.is_online = False # All players start as offline
|
|
self.players[player.player_id] = player
|
|
|
|
for building_data in state.get("buildings", []):
|
|
building = Building(
|
|
building_type=BuildingType(building_data["type"]),
|
|
x=building_data["x"],
|
|
y=building_data["y"],
|
|
owner_id=building_data["owner_id"],
|
|
name=building_data.get("name"),
|
|
placed_at=building_data.get("placed_at", 0.0)
|
|
)
|
|
self.buildings[(building.x, building.y)] = building
|
|
|
|
if building.building_type == BuildingType.ROAD:
|
|
self.road_network.add((building.x, building.y))
|
|
|
|
self._update_connected_zones()
|
|
logger.info(f"Loaded state with {len(self.players)} players and {len(self.buildings)} buildings from database.")
|