"""Game Engine for the Village Simulation. Now includes support for: - Oil industry (drill_oil, refine, burn_fuel) - Religion (pray, preach) - Diplomacy (negotiate, declare_war, make_peace) """ import random import threading import time from dataclasses import dataclass, field from enum import Enum from typing import Optional from backend.domain.agent import Agent from backend.domain.action import ActionType, ActionResult, ACTION_CONFIG from backend.domain.resources import Resource, ResourceType, get_fire_heat, get_fuel_heat from backend.domain.personality import get_action_skill_modifier from backend.domain.religion import get_religion_action_bonus from backend.domain.diplomacy import ( FactionType, get_faction_relations, reset_faction_relations ) from backend.core.world import World, WorldConfig, TimeOfDay from backend.core.market import OrderBook from backend.core.ai import get_ai_decision, AIDecision from backend.core.logger import get_simulation_logger, reset_simulation_logger from backend.config import get_config class SimulationMode(Enum): """Simulation run mode.""" MANUAL = "manual" AUTO = "auto" @dataclass class TurnLog: """Log of events that happened in a turn.""" turn: int agent_actions: list[dict] = field(default_factory=list) deaths: list[str] = field(default_factory=list) trades: list[dict] = field(default_factory=list) religious_events: list[dict] = field(default_factory=list) # NEW diplomatic_events: list[dict] = field(default_factory=list) # NEW def to_dict(self) -> dict: return { "turn": self.turn, "agent_actions": self.agent_actions, "deaths": self.deaths, "trades": self.trades, "religious_events": self.religious_events, "diplomatic_events": self.diplomatic_events, } class GameEngine: """Main game engine singleton.""" _instance: Optional["GameEngine"] = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self.world = World() self.market = OrderBook() self.mode = SimulationMode.MANUAL self.is_running = False self.auto_step_interval = get_config().auto_step_interval self._auto_thread: Optional[threading.Thread] = None self._stop_event = threading.Event() self.turn_logs: list[TurnLog] = [] self.logger = get_simulation_logger() self._initialized = True def reset(self, config: Optional[WorldConfig] = None) -> None: """Reset the simulation to initial state.""" self._stop_auto_mode() # Reset faction relations reset_faction_relations() if config: self.world = World(config=config) else: self.world = World() self.market = OrderBook() self.turn_logs = [] self.logger = reset_simulation_logger() sim_config = get_config() self.logger.start_session(sim_config.to_dict()) self.world.initialize() self.is_running = True def initialize(self, num_agents: Optional[int] = None) -> None: """Initialize the simulation with agents.""" # Reset faction relations reset_faction_relations() if num_agents is not None: self.world.config.initial_agents = num_agents self.world.initialize() self.logger = reset_simulation_logger() sim_config = get_config() self.logger.start_session(sim_config.to_dict()) self.is_running = True def next_step(self) -> TurnLog: """Advance the simulation by one step.""" if not self.is_running: return TurnLog(turn=-1) turn_log = TurnLog(turn=self.world.current_turn + 1) current_turn = self.world.current_turn + 1 self.logger.start_turn( turn=current_turn, day=self.world.current_day, step_in_day=self.world.step_in_day + 1, time_of_day=self.world.time_of_day.value, ) market_orders_before = [o.to_dict() for o in self.market.get_active_orders()] # Remove old corpses self._remove_old_corpses(current_turn) # Collect AI decisions decisions: list[tuple[Agent, AIDecision]] = [] for agent in self.world.get_living_agents(): self.logger.log_agent_before( agent_id=agent.id, agent_name=agent.name, profession=agent.profession.value, position=agent.position.to_dict(), stats=agent.stats.to_dict(), inventory=[r.to_dict() for r in agent.inventory], money=agent.money, ) if self.world.is_night(): decision = AIDecision( action=ActionType.SLEEP, reason="Night time: sleeping", ) else: decision = get_ai_decision( agent, self.market, step_in_day=self.world.step_in_day, day_steps=self.world.config.day_steps, current_turn=current_turn, world=self.world, ) decisions.append((agent, decision)) self.logger.log_agent_decision(agent.id, decision.to_dict()) # Calculate movement for agent, decision in decisions: action_name = decision.action.value agent.set_action( action_type=action_name, world_width=self.world.config.width, world_height=self.world.config.height, message=decision.reason, target_resource=decision.target_resource.value if decision.target_resource else None, target_agent=decision.target_agent_id, ) agent.update_movement() # Execute actions for agent, decision in decisions: result = self._execute_action(agent, decision, turn_log) if result: agent.complete_action(result.success, result.message) turn_log.agent_actions.append({ "agent_id": agent.id, "agent_name": agent.name, "decision": decision.to_dict(), "result": result.to_dict() if result else None, }) self.logger.log_agent_after( agent_id=agent.id, stats=agent.stats.to_dict(), inventory=[r.to_dict() for r in agent.inventory], money=agent.money, position=agent.position.to_dict(), action_result=result.to_dict() if result else {}, ) # Update market prices self.market.update_prices(current_turn) market_orders_after = [o.to_dict() for o in self.market.get_active_orders()] self.logger.log_market_state(market_orders_before, market_orders_after) # Apply passive decay for agent in self.world.get_living_agents(): agent.apply_passive_decay() # Decay resources for agent in self.world.get_living_agents(): expired = agent.decay_inventory(current_turn) # Mark dead agents newly_dead = self._mark_dead_agents(current_turn) for dead_agent in newly_dead: cause = dead_agent.death_reason self.logger.log_death(dead_agent.name, cause) self.market.cancel_seller_orders(dead_agent.id) turn_log.deaths = [a.name for a in newly_dead] self.logger.log_statistics(self.world.get_statistics()) self.logger.end_turn() # Advance time self.world.advance_time() # Check end conditions if len(self.world.get_living_agents()) == 0: self.is_running = False self.logger.close() self.turn_logs.append(turn_log) return turn_log def _mark_dead_agents(self, current_turn: int) -> list[Agent]: """Mark agents who just died as corpses.""" newly_dead = [] for agent in self.world.agents: if not agent.is_alive() and not agent.is_corpse(): cause = agent.stats.get_critical_stat() or "unknown" agent.mark_dead(current_turn, cause) agent.current_action.action_type = "dead" agent.current_action.message = f"Died: {cause}" newly_dead.append(agent) return newly_dead def _remove_old_corpses(self, current_turn: int) -> list[Agent]: """Remove corpses that have been visible for one turn.""" to_remove = [] for agent in self.world.agents: if agent.is_corpse() and agent.death_turn < current_turn: to_remove.append(agent) for agent in to_remove: self.world.agents.remove(agent) self.world.total_agents_died += 1 return to_remove def _execute_action(self, agent: Agent, decision: AIDecision, turn_log: TurnLog) -> Optional[ActionResult]: """Execute an action for an agent.""" action = decision.action config = ACTION_CONFIG[action] # Basic actions if action == ActionType.SLEEP: agent.restore_energy(config.energy_cost) return ActionResult( action_type=action, success=True, energy_spent=-config.energy_cost, message="Sleeping soundly", ) elif action == ActionType.REST: agent.restore_energy(config.energy_cost) return ActionResult( action_type=action, success=True, energy_spent=-config.energy_cost, message="Resting", ) elif action == ActionType.CONSUME: if decision.target_resource: success = agent.consume(decision.target_resource) return ActionResult( action_type=action, success=success, message=f"Consumed {decision.target_resource.value}" if success else "Nothing to consume", ) return ActionResult(action_type=action, success=False, message="No resource specified") elif action == ActionType.BUILD_FIRE: if agent.has_resource(ResourceType.WOOD): agent.remove_from_inventory(ResourceType.WOOD, 1) if not agent.spend_energy(abs(config.energy_cost)): return ActionResult(action_type=action, success=False, message="Not enough energy") fire_heat = get_fire_heat() agent.apply_heat(fire_heat) return ActionResult( action_type=action, success=True, energy_spent=abs(config.energy_cost), heat_gained=fire_heat, message="Built a warm fire", ) return ActionResult(action_type=action, success=False, message="No wood for fire") elif action == ActionType.BURN_FUEL: if agent.has_resource(ResourceType.FUEL): agent.remove_from_inventory(ResourceType.FUEL, 1) if not agent.spend_energy(abs(config.energy_cost)): return ActionResult(action_type=action, success=False, message="Not enough energy") fuel_heat = get_fuel_heat() agent.apply_heat(fuel_heat) # Fuel also provides energy from backend.config import get_config fuel_energy = get_config().resources.fuel_energy agent.restore_energy(fuel_energy) return ActionResult( action_type=action, success=True, energy_spent=abs(config.energy_cost), heat_gained=fuel_heat, message=f"Burned fuel (+{fuel_heat} heat, +{fuel_energy} energy)", ) return ActionResult(action_type=action, success=False, message="No fuel to burn") elif action == ActionType.TRADE: return self._execute_trade(agent, decision) # Religious actions elif action == ActionType.PRAY: return self._execute_pray(agent, config, turn_log) elif action == ActionType.PREACH: return self._execute_preach(agent, config, turn_log) # Diplomatic actions elif action == ActionType.NEGOTIATE: return self._execute_negotiate(agent, decision, config, turn_log) elif action == ActionType.DECLARE_WAR: return self._execute_declare_war(agent, decision, config, turn_log) elif action == ActionType.MAKE_PEACE: return self._execute_make_peace(agent, decision, config, turn_log) # Production actions elif action in [ActionType.HUNT, ActionType.GATHER, ActionType.CHOP_WOOD, ActionType.GET_WATER, ActionType.WEAVE, ActionType.DRILL_OIL, ActionType.REFINE]: return self._execute_work(agent, action, config) return ActionResult(action_type=action, success=False, message="Unknown action") def _execute_work(self, agent: Agent, action: ActionType, config) -> ActionResult: """Execute a work action (hunting, gathering, drilling, etc.).""" energy_cost = abs(config.energy_cost) if not agent.spend_energy(energy_cost): return ActionResult( action_type=action, success=False, message="Not enough energy", ) if config.requires_resource: if not agent.has_resource(config.requires_resource, config.requires_quantity): agent.restore_energy(energy_cost) return ActionResult( action_type=action, success=False, message=f"Missing required {config.requires_resource.value}", ) agent.remove_from_inventory(config.requires_resource, config.requires_quantity) # Get skill modifier skill_name = self._get_skill_for_action(action) skill_value = getattr(agent.skills, skill_name, 1.0) if skill_name else 1.0 skill_modifier = get_action_skill_modifier(skill_value) # Get religion bonus religion_bonus = get_religion_action_bonus(agent.religion.religion, action.value) # Combined modifier total_modifier = skill_modifier * religion_bonus effective_success_chance = min(0.98, config.success_chance * total_modifier) if random.random() > effective_success_chance: agent.record_action(action.value) if skill_name: agent.skills.improve(skill_name, 0.005) return ActionResult( action_type=action, success=False, energy_spent=energy_cost, message="Action failed", ) resources_gained = [] if config.output_resource: base_quantity = random.randint(config.min_output, config.max_output) quantity = max(config.min_output, int(base_quantity * total_modifier)) if quantity > 0: resource = Resource( type=config.output_resource, quantity=quantity, created_turn=self.world.current_turn, ) added = agent.add_to_inventory(resource) if added > 0: resources_gained.append(Resource( type=config.output_resource, quantity=added, created_turn=self.world.current_turn, )) if config.secondary_output: base_quantity = random.randint(config.secondary_min, config.secondary_max) quantity = max(0, int(base_quantity * total_modifier)) if quantity > 0: resource = Resource( type=config.secondary_output, quantity=quantity, created_turn=self.world.current_turn, ) added = agent.add_to_inventory(resource) if added > 0: resources_gained.append(Resource( type=config.secondary_output, quantity=added, created_turn=self.world.current_turn, )) agent.record_action(action.value) if skill_name: agent.skills.improve(skill_name, 0.015) gained_str = ", ".join(f"+{r.quantity} {r.type.value}" for r in resources_gained) message = f"{action.value}: {gained_str}" if gained_str else f"{action.value} (nothing gained)" return ActionResult( action_type=action, success=True, energy_spent=energy_cost, resources_gained=resources_gained, message=message, ) def _get_skill_for_action(self, action: ActionType) -> Optional[str]: """Get the skill name for an action.""" skill_map = { ActionType.HUNT: "hunting", ActionType.GATHER: "gathering", ActionType.CHOP_WOOD: "woodcutting", ActionType.WEAVE: "crafting", ActionType.DRILL_OIL: "gathering", # Use gathering skill for now ActionType.REFINE: "crafting", } return skill_map.get(action) def _execute_pray(self, agent: Agent, config, turn_log: TurnLog) -> ActionResult: """Execute prayer action.""" if not agent.spend_energy(abs(config.energy_cost)): return ActionResult(action_type=ActionType.PRAY, success=False, message="Not enough energy") faith_gain = config.faith_gain agent.gain_faith(faith_gain) agent.religion.record_prayer(self.world.current_turn) agent.record_action("pray") turn_log.religious_events.append({ "type": "prayer", "agent_id": agent.id, "agent_name": agent.name, "religion": agent.religion.religion.value, "faith_gained": faith_gain, "new_faith": agent.stats.faith, }) return ActionResult( action_type=ActionType.PRAY, success=True, energy_spent=abs(config.energy_cost), faith_gained=faith_gain, message=f"Prayed to {agent.religion.religion.value} (+{faith_gain} faith)", ) def _execute_preach(self, agent: Agent, config, turn_log: TurnLog) -> ActionResult: """Execute preaching action to spread religion.""" if not agent.spend_energy(abs(config.energy_cost)): return ActionResult(action_type=ActionType.PREACH, success=False, message="Not enough energy") # Find nearby agents to potentially convert nearby = self.world.get_nearby_agents(agent, radius=4.0) conversions = 0 for target in nearby: if target.religion.religion == agent.religion.religion: # Same religion - boost their faith target.gain_faith(config.faith_spread // 2) else: # Different religion - try to convert if random.random() < config.success_chance: if target.religion.convert_to(agent.religion.religion, 40): conversions += 1 agent.religion.record_conversion() self.world.total_conversions += 1 turn_log.religious_events.append({ "type": "conversion", "preacher_id": agent.id, "convert_id": target.id, "convert_name": target.name, "new_religion": agent.religion.religion.value, }) agent.religion.record_sermon() agent.record_action("preach") # Preaching also boosts own faith agent.gain_faith(config.faith_spread // 2) if conversions > 0: message = f"Converted {conversions} to {agent.religion.religion.value}!" else: message = f"Preached the word of {agent.religion.religion.value}" return ActionResult( action_type=ActionType.PREACH, success=True, energy_spent=abs(config.energy_cost), faith_gained=config.faith_spread // 2, message=message, ) def _execute_negotiate(self, agent: Agent, decision: AIDecision, config, turn_log: TurnLog) -> ActionResult: """Execute diplomatic negotiation.""" if not agent.spend_energy(abs(config.energy_cost)): return ActionResult(action_type=ActionType.NEGOTIATE, success=False, message="Not enough energy") target_faction = decision.target_faction if not target_faction: return ActionResult(action_type=ActionType.NEGOTIATE, success=False, message="No target faction") faction_relations = get_faction_relations() my_faction = agent.diplomacy.faction # Attempt negotiation if random.random() < config.success_chance * agent.diplomacy.diplomacy_skill: # Successful negotiation improves relations from backend.config import get_config boost = get_config().diplomacy.trade_relation_boost * 2 new_relation = faction_relations.modify_relation(my_faction, target_faction, int(boost)) agent.diplomacy.negotiations_conducted += 1 agent.record_action("negotiate") turn_log.diplomatic_events.append({ "type": "negotiation", "agent_id": agent.id, "agent_faction": my_faction.value, "target_faction": target_faction.value, "success": True, "new_relation": new_relation, }) return ActionResult( action_type=ActionType.NEGOTIATE, success=True, energy_spent=abs(config.energy_cost), relation_change=int(boost), target_faction=target_faction.value, diplomatic_effect="improved", message=f"Improved relations with {target_faction.value} (+{int(boost)})", ) else: agent.record_action("negotiate") return ActionResult( action_type=ActionType.NEGOTIATE, success=False, energy_spent=abs(config.energy_cost), target_faction=target_faction.value, message=f"Negotiations with {target_faction.value} failed", ) def _execute_declare_war(self, agent: Agent, decision: AIDecision, config, turn_log: TurnLog) -> ActionResult: """Execute war declaration.""" if not agent.spend_energy(abs(config.energy_cost)): return ActionResult(action_type=ActionType.DECLARE_WAR, success=False, message="Not enough energy") target_faction = decision.target_faction if not target_faction: return ActionResult(action_type=ActionType.DECLARE_WAR, success=False, message="No target faction") faction_relations = get_faction_relations() my_faction = agent.diplomacy.faction success = faction_relations.declare_war(my_faction, target_faction, self.world.current_turn) if success: self.world.total_wars += 1 agent.diplomacy.wars_declared += 1 agent.record_action("declare_war") turn_log.diplomatic_events.append({ "type": "war_declaration", "agent_id": agent.id, "aggressor_faction": my_faction.value, "defender_faction": target_faction.value, }) return ActionResult( action_type=ActionType.DECLARE_WAR, success=True, energy_spent=abs(config.energy_cost), target_faction=target_faction.value, diplomatic_effect="war", message=f"Declared WAR on {target_faction.value}!", ) else: return ActionResult( action_type=ActionType.DECLARE_WAR, success=False, energy_spent=abs(config.energy_cost), message=f"Already at war with {target_faction.value}", ) def _execute_make_peace(self, agent: Agent, decision: AIDecision, config, turn_log: TurnLog) -> ActionResult: """Execute peace treaty.""" if not agent.spend_energy(abs(config.energy_cost)): return ActionResult(action_type=ActionType.MAKE_PEACE, success=False, message="Not enough energy") target_faction = decision.target_faction if not target_faction: return ActionResult(action_type=ActionType.MAKE_PEACE, success=False, message="No target faction") faction_relations = get_faction_relations() my_faction = agent.diplomacy.faction # Peace is harder to achieve if random.random() < config.success_chance * agent.diplomacy.diplomacy_skill: success = faction_relations.make_peace(my_faction, target_faction, self.world.current_turn) if success: self.world.total_peace_treaties += 1 agent.diplomacy.peace_treaties_made += 1 agent.record_action("make_peace") turn_log.diplomatic_events.append({ "type": "peace_treaty", "agent_id": agent.id, "faction1": my_faction.value, "faction2": target_faction.value, }) return ActionResult( action_type=ActionType.MAKE_PEACE, success=True, energy_spent=abs(config.energy_cost), target_faction=target_faction.value, diplomatic_effect="peace", message=f"Peace treaty signed with {target_faction.value}!", ) return ActionResult( action_type=ActionType.MAKE_PEACE, success=False, energy_spent=abs(config.energy_cost), message=f"Peace negotiations with {target_faction.value} failed", ) def _execute_trade(self, agent: Agent, decision: AIDecision) -> ActionResult: """Execute a trade action.""" config = ACTION_CONFIG[ActionType.TRADE] if decision.adjust_order_id and decision.new_price is not None: return self._execute_price_adjustment(agent, decision) if decision.trade_items: return self._execute_multi_buy(agent, decision) if decision.order_id: result = self.market.execute_buy( buyer_id=agent.id, order_id=decision.order_id, quantity=decision.quantity, buyer_money=agent.money, ) if result.success: self.logger.log_trade(result.to_dict()) self.market._record_sale( result.resource_type, result.total_paid // result.quantity, result.quantity, self.world.current_turn, ) agent.money -= result.total_paid resource = Resource( type=result.resource_type, quantity=result.quantity, created_turn=self.world.current_turn, ) agent.add_to_inventory(resource) seller = self.world.get_agent(result.seller_id) if seller: seller.money += result.total_paid seller.record_trade(result.total_paid) seller.skills.improve("trading", 0.02) # Improve faction relations from trade faction_relations = get_faction_relations() from backend.config import get_config boost = get_config().diplomacy.trade_relation_boost faction_relations.modify_relation( agent.diplomacy.faction, seller.diplomacy.faction, boost ) agent.spend_energy(abs(config.energy_cost)) agent.record_action("trade") agent.skills.improve("trading", 0.01) return ActionResult( action_type=ActionType.TRADE, success=True, energy_spent=abs(config.energy_cost), resources_gained=[resource], message=f"Bought {result.quantity} {result.resource_type.value} for {result.total_paid}c", ) else: return ActionResult( action_type=ActionType.TRADE, success=False, message=result.message, ) elif decision.target_resource and decision.quantity > 0: if agent.has_resource(decision.target_resource, decision.quantity): agent.remove_from_inventory(decision.target_resource, decision.quantity) order = self.market.place_order( seller_id=agent.id, resource_type=decision.target_resource, quantity=decision.quantity, price_per_unit=decision.price, current_turn=self.world.current_turn, ) agent.spend_energy(abs(config.energy_cost)) agent.record_action("trade") return ActionResult( action_type=ActionType.TRADE, success=True, energy_spent=abs(config.energy_cost), message=f"Listed {decision.quantity} {decision.target_resource.value} @ {decision.price}c each", ) else: return ActionResult( action_type=ActionType.TRADE, success=False, message="Not enough resources to sell", ) return ActionResult( action_type=ActionType.TRADE, success=False, message="Invalid trade parameters", ) def _execute_price_adjustment(self, agent: Agent, decision: AIDecision) -> ActionResult: """Execute a price adjustment.""" success = self.market.adjust_order_price( order_id=decision.adjust_order_id, seller_id=agent.id, new_price=decision.new_price, current_turn=self.world.current_turn, ) if success: return ActionResult( action_type=ActionType.TRADE, success=True, energy_spent=0, message=f"Adjusted price to {decision.new_price}c", ) else: return ActionResult( action_type=ActionType.TRADE, success=False, message="Failed to adjust price", ) def _execute_multi_buy(self, agent: Agent, decision: AIDecision) -> ActionResult: """Execute a multi-item buy trade.""" config = ACTION_CONFIG[ActionType.TRADE] purchases = [(item.order_id, item.quantity) for item in decision.trade_items] results = self.market.execute_multi_buy( buyer_id=agent.id, purchases=purchases, buyer_money=agent.money, ) total_paid = 0 resources_gained = [] items_bought = [] for result in results: if result.success: self.logger.log_trade(result.to_dict()) agent.money -= result.total_paid total_paid += result.total_paid self.market._record_sale( result.resource_type, result.total_paid // result.quantity, result.quantity, self.world.current_turn ) resource = Resource( type=result.resource_type, quantity=result.quantity, created_turn=self.world.current_turn, ) agent.add_to_inventory(resource) resources_gained.append(resource) items_bought.append(f"{result.quantity} {result.resource_type.value}") seller = self.world.get_agent(result.seller_id) if seller: seller.money += result.total_paid if resources_gained: agent.spend_energy(abs(config.energy_cost)) message = f"Bought {', '.join(items_bought)} for {total_paid}c" return ActionResult( action_type=ActionType.TRADE, success=True, energy_spent=abs(config.energy_cost), resources_gained=resources_gained, message=message, ) else: return ActionResult( action_type=ActionType.TRADE, success=False, message="Failed to buy any items", ) def set_mode(self, mode: SimulationMode) -> None: """Set the simulation mode.""" if mode == self.mode: return if mode == SimulationMode.AUTO: self._start_auto_mode() else: self._stop_auto_mode() self.mode = mode def _start_auto_mode(self) -> None: """Start automatic step advancement.""" self._stop_event.clear() def auto_step(): while not self._stop_event.is_set() and self.is_running: self.next_step() time.sleep(self.auto_step_interval) self._auto_thread = threading.Thread(target=auto_step, daemon=True) self._auto_thread.start() def _stop_auto_mode(self) -> None: """Stop automatic step advancement.""" self._stop_event.set() if self._auto_thread: self._auto_thread.join(timeout=2.0) self._auto_thread = None def get_state(self) -> dict: """Get the full simulation state for API.""" return { **self.world.get_state_snapshot(), "market": self.market.get_state_snapshot(), "mode": self.mode.value, "is_running": self.is_running, "recent_logs": [ log.to_dict() for log in self.turn_logs[-5:] ], } def get_engine() -> GameEngine: """Get the global game engine instance.""" return GameEngine()