"""HTTP client for communicating with the Village Simulation backend. Handles state including religion, factions, diplomacy, and oil economy. """ import time from dataclasses import dataclass, field from typing import Optional, Any import requests from requests.exceptions import RequestException @dataclass class SimulationState: """Parsed simulation state from the API.""" turn: int day: int step_in_day: int time_of_day: str world_width: int world_height: int agents: list[dict] market_orders: list[dict] market_prices: dict statistics: dict mode: str is_running: bool recent_logs: list[dict] # New fields for religion, factions, diplomacy oil_fields: list[dict] = field(default_factory=list) temples: list[dict] = field(default_factory=list) faction_relations: dict = field(default_factory=dict) diplomatic_events: list[dict] = field(default_factory=list) religious_events: list[dict] = field(default_factory=list) active_wars: list[dict] = field(default_factory=list) peace_treaties: list[dict] = field(default_factory=list) @classmethod def from_api_response(cls, data: dict) -> "SimulationState": """Create from API response data.""" return cls( turn=data.get("turn", 0), day=data.get("day", 1), step_in_day=data.get("step_in_day", 0), time_of_day=data.get("time_of_day", "day"), world_width=data.get("world_size", {}).get("width", 20), world_height=data.get("world_size", {}).get("height", 20), agents=data.get("agents", []), market_orders=data.get("market", {}).get("orders", []), market_prices=data.get("market", {}).get("prices", {}), statistics=data.get("statistics", {}), mode=data.get("mode", "manual"), is_running=data.get("is_running", False), recent_logs=data.get("recent_logs", []), # New fields oil_fields=data.get("oil_fields", []), temples=data.get("temples", []), faction_relations=data.get("faction_relations", {}), diplomatic_events=data.get("diplomatic_events", []), religious_events=data.get("religious_events", []), active_wars=data.get("active_wars", []), peace_treaties=data.get("peace_treaties", []), ) def get_living_agents(self) -> list[dict]: """Get only living agents.""" return [a for a in self.agents if a.get("is_alive", False)] def get_agents_by_faction(self) -> dict[str, list[dict]]: """Group living agents by faction.""" result: dict[str, list[dict]] = {} for agent in self.get_living_agents(): # Faction is under diplomacy.faction (not faction.type) diplomacy = agent.get("diplomacy", {}) faction = diplomacy.get("faction", "neutral") if faction not in result: result[faction] = [] result[faction].append(agent) return result def get_agents_by_religion(self) -> dict[str, list[dict]]: """Group living agents by religion.""" result: dict[str, list[dict]] = {} for agent in self.get_living_agents(): # Religion type is under religion.religion (not religion.type) religion_data = agent.get("religion", {}) religion = religion_data.get("religion", "atheist") if religion not in result: result[religion] = [] result[religion].append(agent) return result def get_faction_stats(self) -> dict: """Get faction statistics.""" stats = self.statistics.get("factions", {}) if not stats: # Compute from agents if not in statistics by_faction = self.get_agents_by_faction() stats = {f: len(agents) for f, agents in by_faction.items()} return stats def get_religion_stats(self) -> dict: """Get religion statistics.""" stats = self.statistics.get("religions", {}) if not stats: # Compute from agents if not in statistics by_religion = self.get_agents_by_religion() stats = {r: len(agents) for r, agents in by_religion.items()} return stats def get_avg_faith(self) -> float: """Get average faith level.""" avg = self.statistics.get("avg_faith", 0) if not avg: agents = self.get_living_agents() if agents: # Faith is under religion.faith total_faith = sum( a.get("religion", {}).get("faith", 50) for a in agents ) avg = total_faith / len(agents) return avg class SimulationClient: """HTTP client for the Village Simulation backend.""" def __init__(self, base_url: str = "http://localhost:8000"): self.base_url = base_url.rstrip("/") self.api_url = f"{self.base_url}/api" self.session = requests.Session() self.last_state: Optional[SimulationState] = None self.connected = False self._retry_count = 0 self._max_retries = 3 def _request( self, method: str, endpoint: str, json: Optional[dict] = None, timeout: float = 5.0, ) -> Optional[dict]: """Make an HTTP request to the API.""" url = f"{self.api_url}{endpoint}" try: response = self.session.request( method=method, url=url, json=json, timeout=timeout, ) response.raise_for_status() self.connected = True self._retry_count = 0 return response.json() except RequestException: self._retry_count += 1 if self._retry_count >= self._max_retries: self.connected = False return None def check_connection(self) -> bool: """Check if the backend is reachable.""" try: response = self.session.get( f"{self.base_url}/health", timeout=2.0, ) self.connected = response.status_code == 200 return self.connected except RequestException: self.connected = False return False def get_state(self) -> Optional[SimulationState]: """Fetch the current simulation state.""" data = self._request("GET", "/state") if data: self.last_state = SimulationState.from_api_response(data) return self.last_state return self.last_state def advance_turn(self) -> bool: """Advance the simulation by one step.""" result = self._request("POST", "/control/next_step") return result is not None and result.get("success", False) def set_mode(self, mode: str) -> bool: """Set the simulation mode ('manual' or 'auto').""" result = self._request("POST", "/control/mode", json={"mode": mode}) return result is not None and result.get("success", False) def initialize( self, num_agents: int = 100, world_width: int = 30, world_height: int = 30, ) -> bool: """Initialize or reset the simulation.""" result = self._request("POST", "/control/initialize", json={ "num_agents": num_agents, "world_width": world_width, "world_height": world_height, }) return result is not None and result.get("success", False) def get_status(self) -> Optional[dict]: """Get simulation status.""" return self._request("GET", "/control/status") def get_agents(self) -> Optional[list[dict]]: """Get all agents.""" result = self._request("GET", "/agents") if result: return result.get("agents", []) return None def get_market_orders(self) -> Optional[list[dict]]: """Get all market orders.""" result = self._request("GET", "/market/orders") if result: return result.get("orders", []) return None def get_market_prices(self) -> Optional[dict]: """Get market prices.""" return self._request("GET", "/market/prices") def wait_for_connection(self, timeout: float = 30.0) -> bool: """Wait for backend connection with timeout.""" start = time.time() while time.time() - start < timeout: if self.check_connection(): return True time.sleep(0.5) return False def get_config(self) -> Optional[dict]: """Get current simulation configuration.""" return self._request("GET", "/config") def update_config(self, config_data: dict) -> bool: """Update simulation configuration.""" result = self._request("POST", "/config", json=config_data) return result is not None and result.get("success", False) def reset_config(self) -> bool: """Reset configuration to defaults.""" result = self._request("POST", "/config/reset") return result is not None and result.get("success", False)