2026-01-19 22:55:26 +03:00

699 lines
26 KiB
Python

"""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(),
}