"""Agent model for the Village Simulation. Agent stats are loaded dynamically from the global config. Each agent now has unique personality traits and skills that create emergent professions and behavioral diversity. """ import math import random from dataclasses import dataclass, field from enum import Enum from typing import Optional from uuid import uuid4 from .resources import Resource, ResourceType, RESOURCE_EFFECTS from .personality import ( PersonalityTraits, Skills, ProfessionType, determine_profession ) def _get_agent_stats_config(): """Get agent stats configuration from global config.""" from backend.config import get_config return get_config().agent_stats def _get_age_config(): """Get age configuration from global config.""" from backend.config import get_config return get_config().age class Profession(Enum): """Agent professions - now derived from personality and skills.""" VILLAGER = "villager" HUNTER = "hunter" GATHERER = "gatherer" WOODCUTTER = "woodcutter" TRADER = "trader" CRAFTER = "crafter" @dataclass class Position: """2D position on the map (floating point for smooth movement).""" x: float = 0.0 y: float = 0.0 def distance_to(self, other: "Position") -> float: """Calculate distance to another position.""" return math.sqrt((self.x - other.x) ** 2 + (self.y - other.y) ** 2) def move_towards(self, target: "Position", speed: float = 0.5) -> bool: """Move towards target position. Returns True if reached.""" dist = self.distance_to(target) if dist <= speed: self.x = target.x self.y = target.y return True # Calculate direction dx = target.x - self.x dy = target.y - self.y # Normalize and apply speed self.x += (dx / dist) * speed self.y += (dy / dist) * speed return False def to_dict(self) -> dict: return {"x": round(self.x, 2), "y": round(self.y, 2)} def copy(self) -> "Position": return Position(self.x, self.y) @dataclass class AgentStats: """Vital statistics for an agent. Values are loaded from config.json. Default values are used as fallback. """ # Current values - defaults will be overwritten by factory function energy: int = field(default=50) hunger: int = field(default=80) thirst: int = field(default=70) heat: int = field(default=100) # Maximum values - loaded from config MAX_ENERGY: int = field(default=50) MAX_HUNGER: int = field(default=100) MAX_THIRST: int = field(default=100) MAX_HEAT: int = field(default=100) # Passive decay rates per turn - loaded from config ENERGY_DECAY: int = field(default=1) HUNGER_DECAY: int = field(default=2) THIRST_DECAY: int = field(default=3) HEAT_DECAY: int = field(default=2) # Critical threshold - loaded from config CRITICAL_THRESHOLD: float = field(default=0.25) def apply_passive_decay(self, has_clothes: bool = False, decay_modifier: float = 1.0) -> None: """Apply passive stat decay each turn. Args: has_clothes: Whether agent has clothes (reduces heat decay) decay_modifier: Age-based modifier (old agents decay faster) """ energy_decay = int(self.ENERGY_DECAY * decay_modifier) hunger_decay = int(self.HUNGER_DECAY * decay_modifier) thirst_decay = int(self.THIRST_DECAY * decay_modifier) self.energy = max(0, self.energy - energy_decay) self.hunger = max(0, self.hunger - hunger_decay) self.thirst = max(0, self.thirst - thirst_decay) # Clothes reduce heat loss by 50% heat_decay = int(self.HEAT_DECAY * decay_modifier) heat_decay = heat_decay // 2 if has_clothes else heat_decay self.heat = max(0, self.heat - heat_decay) def is_critical(self) -> bool: """Check if any vital stat is below critical threshold.""" threshold_hunger = int(self.MAX_HUNGER * self.CRITICAL_THRESHOLD) threshold_thirst = int(self.MAX_THIRST * self.CRITICAL_THRESHOLD) threshold_heat = int(self.MAX_HEAT * self.CRITICAL_THRESHOLD) return ( self.hunger < threshold_hunger or self.thirst < threshold_thirst or self.heat < threshold_heat ) def get_critical_stat(self) -> Optional[str]: """Get the name of the most critical stat, if any.""" threshold_thirst = int(self.MAX_THIRST * self.CRITICAL_THRESHOLD) threshold_hunger = int(self.MAX_HUNGER * self.CRITICAL_THRESHOLD) threshold_heat = int(self.MAX_HEAT * self.CRITICAL_THRESHOLD) if self.thirst < threshold_thirst: return "thirst" if self.hunger < threshold_hunger: return "hunger" if self.heat < threshold_heat: return "heat" return None def can_work(self, energy_required: int) -> bool: """Check if agent has enough energy to perform an action.""" return self.energy >= abs(energy_required) def to_dict(self) -> dict: return { "energy": self.energy, "hunger": self.hunger, "thirst": self.thirst, "heat": self.heat, "max_energy": self.MAX_ENERGY, "max_hunger": self.MAX_HUNGER, "max_thirst": self.MAX_THIRST, "max_heat": self.MAX_HEAT, } def create_agent_stats() -> AgentStats: """Factory function to create AgentStats from config.""" config = _get_agent_stats_config() return AgentStats( energy=config.start_energy, hunger=config.start_hunger, thirst=config.start_thirst, heat=config.start_heat, MAX_ENERGY=config.max_energy, MAX_HUNGER=config.max_hunger, MAX_THIRST=config.max_thirst, MAX_HEAT=config.max_heat, ENERGY_DECAY=config.energy_decay, HUNGER_DECAY=config.hunger_decay, THIRST_DECAY=config.thirst_decay, HEAT_DECAY=config.heat_decay, CRITICAL_THRESHOLD=config.critical_threshold, ) @dataclass class AgentAction: """Current action being performed by an agent.""" action_type: str = "" # e.g., "hunt", "gather", "trade", "rest" target_position: Optional[Position] = None target_resource: Optional[str] = None progress: float = 0.0 # 0.0 to 1.0 is_moving: bool = False message: str = "" def to_dict(self) -> dict: return { "action_type": self.action_type, "target_position": self.target_position.to_dict() if self.target_position else None, "target_resource": self.target_resource, "progress": round(self.progress, 2), "is_moving": self.is_moving, "message": self.message, } # Action location mappings (relative positions on the map for each action type) ACTION_LOCATIONS = { "hunt": {"zone": "forest", "offset_range": (0.6, 0.9)}, # Right side (forest) "gather": {"zone": "bushes", "offset_range": (0.1, 0.4)}, # Left side (bushes) "chop_wood": {"zone": "forest", "offset_range": (0.7, 0.95)}, # Far right (forest) "get_water": {"zone": "river", "offset_range": (0.0, 0.15)}, # Far left (river) "weave": {"zone": "village", "offset_range": (0.4, 0.6)}, # Center (village) "build_fire": {"zone": "village", "offset_range": (0.45, 0.55)}, "trade": {"zone": "market", "offset_range": (0.5, 0.6)}, # Center (market) "rest": {"zone": "home", "offset_range": (0.4, 0.6)}, "sleep": {"zone": "home", "offset_range": (0.4, 0.6)}, "consume": {"zone": "current", "offset_range": (0, 0)}, # Stay in place } def _get_world_config(): """Get world configuration from global config.""" from backend.config import get_config return get_config().world @dataclass class Agent: """An agent in the village simulation. Stats, inventory slots, and starting money are loaded from config.json. Each agent now has unique personality traits and skills that create emergent behaviors and professions. Age affects skills, energy costs, and survival: - Young (< 25): Learning faster, lower skill effectiveness, less energy cost - Prime (25-45): Peak performance - Old (> 45): Higher skill effectiveness (wisdom), but higher energy costs """ id: str = field(default_factory=lambda: str(uuid4())[:8]) name: str = "" profession: Profession = Profession.VILLAGER # Now derived from personality/skills position: Position = field(default_factory=Position) stats: AgentStats = field(default_factory=create_agent_stats) inventory: list[Resource] = field(default_factory=list) money: int = field(default=-1) # -1 signals to use config value # Personality and skills - create agent diversity personality: PersonalityTraits = field(default_factory=PersonalityTraits) skills: Skills = field(default_factory=Skills) # Age system - age is in "years" where 1 year = 1 simulation day age: int = field(default=-1) # -1 signals to use random start age max_age: int = field(default=-1) # -1 signals to calculate from config birth_day: int = 0 # Day this agent was born (0 = initial spawn) last_birth_day: int = -1000 # Last day this agent gave birth (for cooldown) parent_ids: list[str] = field(default_factory=list) # IDs of parents (for lineage) children_ids: list[str] = field(default_factory=list) # IDs of children generation: int = 0 # 0 = initial spawn, 1+ = born in simulation # Movement and action tracking home_position: Position = field(default_factory=Position) current_action: AgentAction = field(default_factory=AgentAction) last_action_result: str = "" # Death tracking for corpse visualization death_turn: int = -1 # Turn when agent died, -1 if alive death_reason: str = "" # Cause of death # Statistics tracking for profession determination actions_performed: dict = field(default_factory=lambda: { "hunt": 0, "gather": 0, "chop_wood": 0, "trade": 0, "craft": 0 }) total_trades_completed: int = 0 total_money_earned: int = 0 # Personal action log (recent actions with results) action_history: list = field(default_factory=list) MAX_HISTORY_SIZE: int = 20 # Configuration - loaded from config INVENTORY_SLOTS: int = field(default=-1) # -1 signals to use config value MOVE_SPEED: float = 0.8 # Grid cells per turn def __post_init__(self): if not self.name: self.name = f"Agent_{self.id}" # Set home position to initial position self.home_position = self.position.copy() # Load config values if defaults were used config = _get_world_config() if self.money == -1: self.money = config.starting_money if self.INVENTORY_SLOTS == -1: self.INVENTORY_SLOTS = config.inventory_slots # Initialize age system age_config = _get_age_config() if self.age == -1: # Random starting age within configured range self.age = random.randint(age_config.min_start_age, age_config.max_start_age) if self.max_age == -1: # Calculate max age with variance variance = random.randint(-age_config.max_age_variance, age_config.max_age_variance) self.max_age = age_config.base_max_age + variance # Apply age-based max energy adjustment for old agents if self.get_age_category() == "old": self.stats.MAX_ENERGY = int(self.stats.MAX_ENERGY * age_config.old_max_energy_multiplier) self.stats.energy = min(self.stats.energy, self.stats.MAX_ENERGY) # Update profession based on personality and skills self._update_profession() def _update_profession(self) -> None: """Update profession based on personality and skills.""" prof_type = determine_profession(self.personality, self.skills) profession_map = { ProfessionType.HUNTER: Profession.HUNTER, ProfessionType.GATHERER: Profession.GATHERER, ProfessionType.WOODCUTTER: Profession.WOODCUTTER, ProfessionType.TRADER: Profession.TRADER, ProfessionType.GENERALIST: Profession.VILLAGER, } self.profession = profession_map.get(prof_type, Profession.VILLAGER) def get_age_category(self) -> str: """Get the agent's age category: 'young', 'prime', or 'old'.""" age_config = _get_age_config() if self.age < age_config.young_age_threshold: return "young" elif self.age <= age_config.old_age_threshold: return "prime" else: return "old" def get_skill_modifier(self) -> float: """Get skill effectiveness modifier based on age. Young agents are less effective but learn faster. Old agents are more effective (wisdom) but learn slower. """ age_config = _get_age_config() category = self.get_age_category() if category == "young": return age_config.young_skill_multiplier elif category == "prime": return age_config.prime_skill_multiplier else: return age_config.old_skill_multiplier def get_learning_modifier(self) -> float: """Get learning rate modifier based on age.""" age_config = _get_age_config() category = self.get_age_category() if category == "young": return age_config.young_learning_multiplier elif category == "prime": return age_config.prime_learning_multiplier else: return age_config.old_learning_multiplier def get_energy_cost_modifier(self) -> float: """Get energy cost modifier based on age. Young agents use less energy. Old agents use more energy. """ age_config = _get_age_config() category = self.get_age_category() if category == "young": return age_config.young_energy_cost_multiplier elif category == "prime": return age_config.prime_energy_cost_multiplier else: return age_config.old_energy_cost_multiplier def get_decay_modifier(self) -> float: """Get stat decay modifier based on age. Old agents decay faster (frailer). """ age_config = _get_age_config() if self.get_age_category() == "old": return age_config.old_decay_multiplier return 1.0 def age_one_day(self) -> None: """Age the agent by one day (called at day transition).""" age_config = _get_age_config() self.age += age_config.age_per_day # Check if agent just became old - reduce max energy if self.age == age_config.old_age_threshold + 1: self.stats.MAX_ENERGY = int(self.stats.MAX_ENERGY * age_config.old_max_energy_multiplier) self.stats.energy = min(self.stats.energy, self.stats.MAX_ENERGY) def is_too_old(self) -> bool: """Check if agent has exceeded their maximum age.""" return self.age >= self.max_age def can_give_birth(self, current_day: int) -> bool: """Check if agent is eligible to give birth.""" age_config = _get_age_config() # Age check if self.age < age_config.min_birth_age or self.age > age_config.max_birth_age: return False # Cooldown check days_since_birth = current_day - self.last_birth_day if days_since_birth < age_config.birth_cooldown_days: return False # Resource check if self.stats.hunger < age_config.birth_food_requirement: return False if self.stats.energy < age_config.birth_energy_requirement: return False return True def record_birth(self, current_day: int, child_id: str) -> None: """Record that this agent gave birth.""" self.last_birth_day = current_day self.children_ids.append(child_id) # Birth is exhausting - reduce stats self.stats.energy = max(0, self.stats.energy - 20) self.stats.hunger = max(0, self.stats.hunger - 30) def record_action(self, action_type: str) -> None: """Record an action for profession tracking.""" if action_type in self.actions_performed: self.actions_performed[action_type] += 1 def log_action(self, turn: int, action_type: str, result: str, success: bool = True) -> None: """Add an action to the agent's personal history log.""" entry = { "turn": turn, "action": action_type, "result": result, "success": success, } self.action_history.append(entry) # Keep only recent history if len(self.action_history) > self.MAX_HISTORY_SIZE: self.action_history = self.action_history[-self.MAX_HISTORY_SIZE:] def record_trade(self, money_earned: int) -> None: """Record a completed trade for statistics.""" self.total_trades_completed += 1 if money_earned > 0: self.total_money_earned += money_earned def is_alive(self) -> bool: """Check if the agent is still alive.""" # Death by needs if self.stats.hunger <= 0 or self.stats.thirst <= 0 or self.stats.heat <= 0: return False # Death by old age if self.is_too_old(): return False return True def is_corpse(self) -> bool: """Check if this agent is a corpse (died but still visible).""" return self.death_turn >= 0 def can_act(self) -> bool: """Check if agent can perform active actions.""" return self.is_alive() and self.stats.energy > 0 def has_clothes(self) -> bool: """Check if agent has clothes equipped.""" return any(r.type == ResourceType.CLOTHES for r in self.inventory) def inventory_space(self) -> int: """Get remaining inventory slots.""" total_items = sum(r.quantity for r in self.inventory) return max(0, self.INVENTORY_SLOTS - total_items) def inventory_full(self) -> bool: """Check if inventory is full.""" return self.inventory_space() <= 0 def set_action( self, action_type: str, world_width: int, world_height: int, message: str = "", target_resource: Optional[str] = None, ) -> None: """Set the current action and calculate target position.""" location = ACTION_LOCATIONS.get(action_type, {"zone": "current", "offset_range": (0, 0)}) if location["zone"] == "current": # Stay in place target = self.position.copy() is_moving = False else: # Calculate target position based on action zone offset_range = location["offset_range"] offset_min = float(offset_range[0]) offset_max = float(offset_range[1]) target_x = world_width * random.uniform(offset_min, offset_max) # Keep y position somewhat consistent but allow some variation target_y = self.home_position.y + random.uniform(-2, 2) target_y = max(0.5, min(world_height - 0.5, target_y)) target = Position(target_x, target_y) is_moving = self.position.distance_to(target) > 0.5 self.current_action = AgentAction( action_type=action_type, target_position=target, target_resource=target_resource, progress=0.0, is_moving=is_moving, message=message, ) def update_movement(self) -> None: """Update agent position moving towards target.""" if self.current_action.target_position and self.current_action.is_moving: reached = self.position.move_towards( self.current_action.target_position, self.MOVE_SPEED ) if reached: self.current_action.is_moving = False self.current_action.progress = 0.5 # At location, doing action def complete_action(self, success: bool, message: str) -> None: """Mark current action as complete.""" self.current_action.progress = 1.0 self.current_action.is_moving = False self.last_action_result = message self.current_action.message = message if success else f"Failed: {message}" def add_to_inventory(self, resource: Resource) -> int: """Add resource to inventory, returns quantity actually added.""" space = self.inventory_space() if space <= 0: return 0 quantity_to_add = min(resource.quantity, space) # Try to stack with existing resource of same type for existing in self.inventory: if existing.type == resource.type: existing.quantity += quantity_to_add return quantity_to_add # Add as new stack new_resource = Resource( type=resource.type, quantity=quantity_to_add, created_turn=resource.created_turn, ) self.inventory.append(new_resource) return quantity_to_add def remove_from_inventory(self, resource_type: ResourceType, quantity: int = 1) -> int: """Remove resource from inventory, returns quantity actually removed.""" removed = 0 for resource in self.inventory[:]: if resource.type == resource_type: take = min(resource.quantity, quantity - removed) resource.quantity -= take removed += take if resource.quantity <= 0: self.inventory.remove(resource) if removed >= quantity: break return removed def get_resource_count(self, resource_type: ResourceType) -> int: """Get total count of a resource type in inventory.""" return sum( r.quantity for r in self.inventory if r.type == resource_type ) def has_resource(self, resource_type: ResourceType, quantity: int = 1) -> bool: """Check if agent has at least the specified quantity of a resource.""" return self.get_resource_count(resource_type) >= quantity def consume(self, resource_type: ResourceType) -> bool: """Consume a resource from inventory and apply its effects.""" if not self.has_resource(resource_type, 1): return False effect = RESOURCE_EFFECTS[resource_type] self.stats.hunger = min( self.stats.MAX_HUNGER, self.stats.hunger + effect.hunger ) self.stats.thirst = min( self.stats.MAX_THIRST, self.stats.thirst + effect.thirst ) self.stats.heat = min( self.stats.MAX_HEAT, self.stats.heat + effect.heat ) self.stats.energy = min( self.stats.MAX_ENERGY, self.stats.energy + effect.energy ) self.remove_from_inventory(resource_type, 1) return True def apply_heat(self, amount: int) -> None: """Apply heat from a fire.""" self.stats.heat = min(self.stats.MAX_HEAT, self.stats.heat + amount) def restore_energy(self, amount: int) -> None: """Restore energy (from sleep/rest).""" self.stats.energy = min(self.stats.MAX_ENERGY, self.stats.energy + amount) def spend_energy(self, amount: int) -> bool: """Spend energy on an action. Returns False if not enough energy.""" if self.stats.energy < amount: return False self.stats.energy -= amount return True def decay_inventory(self, current_turn: int) -> list[Resource]: """Remove expired resources from inventory. Returns list of removed resources.""" expired = [] for resource in self.inventory[:]: if resource.is_expired(current_turn): expired.append(resource) self.inventory.remove(resource) return expired def apply_passive_decay(self) -> None: """Apply passive stat decay for this turn, modified by age.""" decay_modifier = self.get_decay_modifier() self.stats.apply_passive_decay(has_clothes=self.has_clothes(), decay_modifier=decay_modifier) def mark_dead(self, turn: int, reason: str) -> None: """Mark this agent as dead.""" self.death_turn = turn self.death_reason = reason def to_dict(self) -> dict: """Convert to dictionary for API serialization.""" # Update profession before serializing self._update_profession() return { "id": self.id, "name": self.name, "profession": self.profession.value, "position": self.position.to_dict(), "home_position": self.home_position.to_dict(), "stats": self.stats.to_dict(), "inventory": [r.to_dict() for r in self.inventory], "money": self.money, "is_alive": self.is_alive(), "is_corpse": self.is_corpse(), "can_act": self.can_act(), "current_action": self.current_action.to_dict(), "last_action_result": self.last_action_result, "death_turn": self.death_turn, "death_reason": self.death_reason, # Age system "age": self.age, "max_age": self.max_age, "age_category": self.get_age_category(), "birth_day": self.birth_day, "generation": self.generation, "parent_ids": self.parent_ids.copy(), "children_count": len(self.children_ids), # Age modifiers (for UI display) "skill_modifier": round(self.get_skill_modifier(), 2), "energy_cost_modifier": round(self.get_energy_cost_modifier(), 2), "learning_modifier": round(self.get_learning_modifier(), 2), # New fields for agent diversity "personality": self.personality.to_dict(), "skills": self.skills.to_dict(), "actions_performed": self.actions_performed.copy(), "total_trades": self.total_trades_completed, "total_money_earned": self.total_money_earned, # Personal action history "action_history": self.action_history.copy(), }