460 lines
17 KiB
Python
460 lines
17 KiB
Python
"""Agent model for the Village Simulation.
|
|
|
|
Agent stats are loaded dynamically from the global config.
|
|
"""
|
|
|
|
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
|
|
|
|
|
|
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 - kept for backwards compatibility but no longer used."""
|
|
VILLAGER = "villager"
|
|
HUNTER = "hunter"
|
|
GATHERER = "gatherer"
|
|
WOODCUTTER = "woodcutter"
|
|
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.
|
|
"""
|
|
id: str = field(default_factory=lambda: str(uuid4())[:8])
|
|
name: str = ""
|
|
profession: Profession = Profession.VILLAGER # No longer used for decision making
|
|
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
|
|
|
|
# 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
|
|
|
|
# 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
|
|
|
|
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."""
|
|
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,
|
|
}
|