"""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 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) -> None: """Apply passive stat decay each turn.""" self.energy = max(0, self.energy - self.ENERGY_DECAY) self.hunger = max(0, self.hunger - self.HUNGER_DECAY) self.thirst = max(0, self.thirst - self.THIRST_DECAY) # Clothes reduce heat loss by 50% heat_decay = self.HEAT_DECAY // 2 if has_clothes else self.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. """ 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) # 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 # 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 # 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 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 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.""" return ( self.stats.hunger > 0 and self.stats.thirst > 0 and self.stats.heat > 0 ) 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.""" self.stats.apply_passive_decay(has_clothes=self.has_clothes()) 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, # 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, }