255 lines
9.2 KiB
Python
255 lines
9.2 KiB
Python
"""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)
|