""" behaviors.consume ================= Generic resource consumption and collection behaviors for WorldSmithAI. This module implements domain-agnostic resource interaction behaviors: - ConsumeBehavior - EatBehavior - HarvestBehavior - MineBehavior - CollectBehavior The behaviors can be attached to any agent type and can target any resource type. No species, economy, ecosystem, or fantasy-specific assumptions are hardcoded. Example DSL-inspired behavior parameters ---------------------------------------- Consume a nearby resource and increase an agent state key: { "id": "consume_food", "type": "consume", "parameters": { "selector": {"type": "food"}, "amount": 1.0, "radius": 2.0, "state_gain_key": "energy", "state_gain_multiplier": 0.5, "allow_partial": true } } Harvest a resource into inventory: { "id": "harvest_grain", "type": "harvest", "parameters": { "selector": {"type": "grain"}, "amount": 3.0, "radius": 1.5, "inventory_key": "inventory", "inventory_item_key": "grain" } } Collect knowledge without depleting the world resource: { "id": "collect_knowledge", "type": "collect", "parameters": { "selector": {"type": "knowledge"}, "amount": 1.0, "deplete_resource": false, "state_gain_key": "knowledge", "state_gain_multiplier": 1.0 } } Minimal usage example --------------------- from core.agent import Agent from core.resource import Resource from core.world import World from behaviors.consume import ConsumeBehavior world = World() world.add_resource( Resource( id="food_1", type="food", amount=10.0, position=[1.0, 0.0], ) ) agent = Agent( id="forager_1", type="forager", position=[0.0, 0.0], state={"energy": 0.2}, behaviors={ "consume_food": ConsumeBehavior( id="consume_food", parameters={ "selector": {"type": "food"}, "amount": 2.0, "radius": 2.0, "state_gain_key": "energy", "state_gain_multiplier": 0.25, }, ) }, ) world.add_agent(agent) result = world.step() """ from __future__ import annotations import logging from collections.abc import Mapping, Sequence from copy import deepcopy from dataclasses import dataclass, field from numbers import Real from typing import Any, Literal, TypeAlias import numpy as np from numpy.typing import NDArray from core.agent import Agent, WorldProtocol from core.behavior import Behavior, BehaviorResult from core.resource import Resource, ResourceOperationResult logger = logging.getLogger(__name__) PositionInput: TypeAlias = Sequence[float] | NDArray[np.float64] ResourceSelectionStrategy: TypeAlias = Literal["nearest", "farthest", "largest", "smallest", "random"] DEFAULT_INTERACTION_AMOUNT: float = 1.0 ZERO_TOLERANCE: float = 1.0e-12 @dataclass(frozen=True, slots=True) class ResourceCandidate: """ Candidate resource selected from the world. Attributes ---------- id: Resource id. type: Resource type. amount: Current resource amount. position: Resource position. distance: Distance from the acting agent. metadata: Additional resource metadata. """ id: str type: str amount: float position: NDArray[np.float64] distance: float metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize resource candidate fields. Raises ------ ValueError If id, type, amount, position, or distance are invalid. TypeError If metadata is invalid. """ if not isinstance(self.id, str) or not self.id.strip(): raise ValueError("ResourceCandidate.id must be a non-empty string.") if not isinstance(self.type, str) or not self.type.strip(): raise ValueError("ResourceCandidate.type must be a non-empty string.") if isinstance(self.amount, bool) or not isinstance(self.amount, Real): raise TypeError("ResourceCandidate.amount must be numeric.") amount = float(self.amount) if not np.isfinite(amount): raise ValueError("ResourceCandidate.amount must be finite.") if amount < 0.0: raise ValueError("ResourceCandidate.amount cannot be negative.") position = self._normalize_position(self.position, "position") if not np.isfinite(self.distance): raise ValueError("ResourceCandidate.distance must be finite.") if self.distance < 0.0: raise ValueError("ResourceCandidate.distance cannot be negative.") if not isinstance(self.metadata, Mapping): raise TypeError("ResourceCandidate.metadata must be a mapping.") object.__setattr__(self, "id", self.id.strip()) object.__setattr__(self, "type", self.type.strip()) object.__setattr__(self, "amount", amount) object.__setattr__(self, "position", position) object.__setattr__(self, "distance", float(self.distance)) object.__setattr__(self, "metadata", deepcopy(dict(self.metadata))) def to_metadata(self) -> dict[str, Any]: """ Convert the candidate to JSON-friendly metadata. Returns ------- dict[str, Any] Serializable candidate data. """ return { "id": self.id, "type": self.type, "amount": self.amount, "position": self.position.tolist(), "distance": self.distance, "metadata": deepcopy(dict(self.metadata)), } @staticmethod def _normalize_position( value: PositionInput, label: str, ) -> NDArray[np.float64]: """ Normalize a position into a finite one-dimensional vector. Parameters ---------- value: Candidate position. label: Human-readable label. Returns ------- NDArray[np.float64] Normalized position. """ position = np.asarray(value, dtype=np.float64) if position.ndim != 1: raise ValueError(f"{label} must be one-dimensional.") if position.size == 0: raise ValueError(f"{label} cannot be empty.") if not np.all(np.isfinite(position)): raise ValueError(f"{label} must contain only finite values.") return position @dataclass(frozen=True, slots=True) class ResourceInteractionComputation: """ Structured summary of one resource interaction. Attributes ---------- interaction_kind: Human-readable interaction kind, such as ``"consume"``, ``"eat"``, ``"harvest"``, ``"mine"``, or ``"collect"``. resource_id: Resource id. resource_type: Resource type. requested_amount: Amount requested by the behavior. actual_amount: Amount actually obtained or consumed. resource_depleted: Whether the resource was depleted. partial: Whether less than the requested amount was obtained. removed_empty_resource: Whether the resource was removed from the world after interaction. operation_result: Optional ResourceOperationResult produced by Resource.deplete(...). candidate: Target candidate metadata. state_updates: Agent state updates generated by this interaction. """ interaction_kind: str resource_id: str resource_type: str requested_amount: float actual_amount: float resource_depleted: bool partial: bool removed_empty_resource: bool operation_result: ResourceOperationResult | None = None candidate: ResourceCandidate | None = None state_updates: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize interaction computation fields. Raises ------ ValueError If identifiers or numeric values are invalid. TypeError If boolean flags or state updates are invalid. """ if not isinstance(self.interaction_kind, str) or not self.interaction_kind.strip(): raise ValueError("interaction_kind must be a non-empty string.") if not isinstance(self.resource_id, str) or not self.resource_id.strip(): raise ValueError("resource_id must be a non-empty string.") if not isinstance(self.resource_type, str) or not self.resource_type.strip(): raise ValueError("resource_type must be a non-empty string.") for label, value in { "requested_amount": self.requested_amount, "actual_amount": self.actual_amount, }.items(): if not np.isfinite(value): raise ValueError(f"{label} must be finite.") if value < 0.0: raise ValueError(f"{label} cannot be negative.") for label, value in { "resource_depleted": self.resource_depleted, "partial": self.partial, "removed_empty_resource": self.removed_empty_resource, }.items(): if not isinstance(value, bool): raise TypeError(f"{label} must be a boolean.") if self.operation_result is not None and not isinstance( self.operation_result, ResourceOperationResult, ): raise TypeError("operation_result must be a ResourceOperationResult or None.") if self.candidate is not None and not isinstance(self.candidate, ResourceCandidate): raise TypeError("candidate must be a ResourceCandidate or None.") if not isinstance(self.state_updates, Mapping): raise TypeError("state_updates must be a mapping.") object.__setattr__(self, "interaction_kind", self.interaction_kind.strip()) object.__setattr__(self, "resource_id", self.resource_id.strip()) object.__setattr__(self, "resource_type", self.resource_type.strip()) object.__setattr__(self, "requested_amount", float(self.requested_amount)) object.__setattr__(self, "actual_amount", float(self.actual_amount)) object.__setattr__(self, "state_updates", deepcopy(dict(self.state_updates))) def to_metadata(self) -> dict[str, Any]: """ Convert the computation into JSON-friendly metadata. Returns ------- dict[str, Any] Serializable interaction metadata. """ return { "interaction_kind": self.interaction_kind, "resource_id": self.resource_id, "resource_type": self.resource_type, "requested_amount": self.requested_amount, "actual_amount": self.actual_amount, "resource_depleted": self.resource_depleted, "partial": self.partial, "removed_empty_resource": self.removed_empty_resource, "operation_result": ( None if self.operation_result is None else self.operation_result.to_dict() ), "candidate": None if self.candidate is None else self.candidate.to_metadata(), "state_updates": deepcopy(dict(self.state_updates)), } @dataclass(slots=True) class ResourceInteractionBehavior(Behavior): """ Base class for resource interaction behaviors. This class implements resource selection, amount resolution, optional depletion, optional inventory updates, optional state gains, optional action cost, and reward calculation. Concrete subclasses configure semantic defaults such as interaction kind and whether gathered amounts should be stored in inventory by default. """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check generic resource interaction preconditions. Parameters ---------- agent: Agent attempting the behavior. world: World in which the behavior would execute. Returns ------- bool True if a required resource can be found and action cost can be paid. """ try: if not self._has_sufficient_action_budget(agent, DEFAULT_INTERACTION_AMOUNT): return False if not self._requires_resource(): return True selected = self._select_resource(agent=agent, world=world) return selected is not None except Exception: logger.exception( "Resource interaction precondition failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return False def _execute_resource_interaction( self, *, agent: Agent, world: WorldProtocol, interaction_kind: str, success_message: str, failure_message: str, store_in_inventory_default: bool, deplete_resource_default: bool = True, ) -> BehaviorResult: """ Execute a generic resource interaction. Parameters ---------- agent: Agent executing the behavior. world: World in which the behavior executes. interaction_kind: Interaction label. success_message: Message used when interaction succeeds. failure_message: Message used when no matching resource or usable amount is found. store_in_inventory_default: Whether to store obtained amount in inventory by default. deplete_resource_default: Whether to deplete the resource by default. Returns ------- BehaviorResult Behavior execution result. """ try: selected = self._select_resource(agent=agent, world=world) if selected is None: if self._requires_resource(): return self.failure( reward=self._failure_reward(), message=failure_message, metadata={ "selector": self._selector(), "radius": self._radius(), "selection_strategy": self._selection_strategy(), }, ) return self.success( reward=self._success_reward(actual_amount=0.0), message="Resource interaction completed with no required resource.", metadata={ "interaction_kind": interaction_kind, "resource_required": False, }, ) resource, candidate = selected requested_amount = self._requested_amount(resource) deplete_resource = self._depletes_resource(default=deplete_resource_default) if not self._has_sufficient_action_budget(agent, requested_amount): return self.failure( reward=self._failure_reward(), message="Agent does not have sufficient state budget for interaction.", metadata={ "interaction_kind": interaction_kind, "required_cost": self._action_cost(requested_amount), "cost_state_key": self._cost_state_key(), }, ) operation_result: ResourceOperationResult | None = None if deplete_resource: operation_result = resource.deplete( requested_amount, allow_partial=self._allow_partial(), metadata={ "source": self.__class__.__name__, "behavior_id": self.id, "agent_id": agent.id, "interaction_kind": interaction_kind, }, ) actual_amount = abs(operation_result.actual_delta) partial = operation_result.partial operation_success = operation_result.success else: actual_amount = self._non_depleting_actual_amount( resource=resource, requested_amount=requested_amount, ) partial = actual_amount < requested_amount operation_success = actual_amount > 0.0 or requested_amount == 0.0 removed_empty_resource = self._remove_resource_if_empty( resource=resource, world=world, ) if not operation_success and actual_amount <= ZERO_TOLERANCE: computation = ResourceInteractionComputation( interaction_kind=interaction_kind, resource_id=resource.id, resource_type=resource.type, requested_amount=requested_amount, actual_amount=actual_amount, resource_depleted=deplete_resource, partial=partial, removed_empty_resource=removed_empty_resource, operation_result=operation_result, candidate=candidate, state_updates={}, ) return self.failure( reward=self._failure_reward(), message="Resource interaction did not obtain any amount.", metadata=computation.to_metadata(), ) state_updates = self._build_state_updates( agent=agent, resource=resource, actual_amount=actual_amount, store_in_inventory_default=store_in_inventory_default, ) cost_updates = self._action_cost_state_updates( agent=agent, actual_amount=actual_amount, existing_updates=state_updates, ) state_updates.update(cost_updates) computation = ResourceInteractionComputation( interaction_kind=interaction_kind, resource_id=resource.id, resource_type=resource.type, requested_amount=requested_amount, actual_amount=actual_amount, resource_depleted=deplete_resource, partial=partial, removed_empty_resource=removed_empty_resource, operation_result=operation_result, candidate=candidate, state_updates=state_updates, ) metadata = computation.to_metadata() metadata.update( { "allow_partial": self._allow_partial(), "deplete_resource": deplete_resource, "store_in_inventory": self._stores_in_inventory( default=store_in_inventory_default ), } ) return self.success( reward=self._success_reward(actual_amount=actual_amount), state_updates=state_updates, memory_updates={ "resource_interaction": metadata, }, message=success_message, metadata=metadata, ) except Exception as exc: logger.exception( "%s failed for agent_id=%s behavior_id=%s", self.__class__.__name__, agent.id, self.id, ) return self.failure( reward=self._failure_reward(), message=f"{self.__class__.__name__} raised an exception.", metadata={"error": repr(exc)}, ) def _select_resource( self, *, agent: Agent, world: WorldProtocol, ) -> tuple[Resource, ResourceCandidate] | None: """ Select a resource from the world. Parameters ---------- agent: Acting agent. world: World containing resources. Returns ------- tuple[Resource, ResourceCandidate] | None Selected resource and candidate metadata, or None. """ resources = self._resources_from_world(world) selector = self._selector() radius = self._radius() candidates: list[tuple[Resource, ResourceCandidate]] = [] for resource in resources: if not isinstance(resource, Resource): continue if not self._matches_resource_selector(resource, selector): continue if self._ignore_empty_resources() and resource.amount <= ZERO_TOLERANCE: continue if resource.position.shape != agent.position.shape: continue distance = float(np.linalg.norm(resource.position - agent.position)) if radius is not None and distance > radius: continue candidate = ResourceCandidate( id=resource.id, type=resource.type, amount=resource.amount, position=resource.position.copy(), distance=distance, metadata=deepcopy(resource.metadata), ) candidates.append((resource, candidate)) if not candidates: return None strategy = self._selection_strategy() if strategy == "nearest": return min(candidates, key=lambda item: (item[1].distance, item[1].id)) if strategy == "farthest": return max(candidates, key=lambda item: (item[1].distance, item[1].id)) if strategy == "largest": return max(candidates, key=lambda item: (item[1].amount, item[1].id)) if strategy == "smallest": return min(candidates, key=lambda item: (item[1].amount, item[1].id)) if strategy == "random": rng = self._rng(world) index = int(rng.integers(0, len(candidates))) return candidates[index] raise ValueError(f"Unsupported resource selection strategy: {strategy}") def _resources_from_world(self, world: WorldProtocol) -> list[Any]: """ Retrieve candidate resources from a world-like object. Parameters ---------- world: World-like object. Returns ------- list[Any] Candidate resources. """ selector = self._selector() selector_method = getattr(world, "select_resources", None) if callable(selector_method): try: selected = selector_method(selector) except TypeError: selected = selector_method(selector=selector) return list(selected) resources = getattr(world, "resources", {}) if isinstance(resources, Mapping): return list(resources.values()) if isinstance(resources, Sequence) and not isinstance(resources, (str, bytes)): return list(resources) return [] def _matches_resource_selector( self, resource: Resource, selector: Mapping[str, Any], ) -> bool: """ Return whether a resource matches the behavior selector. Parameters ---------- resource: Candidate resource. selector: Selector mapping. Returns ------- bool True if resource matches. """ if "id" in selector and not self._value_matches(resource.id, selector["id"]): return False if "ids" in selector and not self._value_matches(resource.id, selector["ids"]): return False if "type" in selector and not self._value_matches(resource.type, selector["type"]): return False if "types" in selector and not self._value_matches(resource.type, selector["types"]): return False if "metadata" in selector and not self._mapping_contains( resource.metadata, selector["metadata"], ): return False min_amount = selector.get("min_amount", selector.get("amount_min")) max_amount = selector.get("max_amount", selector.get("amount_max")) if min_amount is not None: if resource.amount < self._normalize_finite_float(min_amount, "min_amount"): return False if max_amount is not None: if resource.amount > self._normalize_finite_float(max_amount, "max_amount"): return False return True def _requested_amount(self, resource: Resource) -> float: """ Resolve requested interaction amount. Parameters ---------- resource: Selected resource. Returns ------- float Non-negative requested amount. """ for key in ( "amount", "consume_amount", "eat_amount", "harvest_amount", "mine_amount", "collect_amount", "quantity", ): if key in self.parameters and self.parameters[key] is not None: value = self.parameters[key] if isinstance(value, str): normalized = value.strip().lower() if normalized in {"all", "available", "max", "maximum"}: return resource.amount return self._normalize_non_negative_float(value, key) return DEFAULT_INTERACTION_AMOUNT def _non_depleting_actual_amount( self, *, resource: Resource, requested_amount: float, ) -> float: """ Compute obtained amount for non-depleting interactions. Parameters ---------- resource: Selected resource. requested_amount: Requested amount. Returns ------- float Actual amount obtained. """ if self._requires_available_amount_for_non_depleting(): return min(requested_amount, resource.amount) return requested_amount def _build_state_updates( self, *, agent: Agent, resource: Resource, actual_amount: float, store_in_inventory_default: bool, ) -> dict[str, Any]: """ Build state updates caused by the interaction. Parameters ---------- agent: Acting agent. resource: Selected resource. actual_amount: Amount obtained or consumed. store_in_inventory_default: Whether inventory storage is enabled by default. Returns ------- dict[str, Any] Agent state updates. """ updates: dict[str, Any] = {} updates.update( self._state_gain_updates( agent=agent, resource=resource, actual_amount=actual_amount, ) ) if self._stores_in_inventory(default=store_in_inventory_default): updates.update( self._inventory_state_update( agent=agent, resource=resource, actual_amount=actual_amount, ) ) explicit_updates = self.parameters.get("state_updates") if explicit_updates is not None: if not isinstance(explicit_updates, Mapping): raise TypeError("state_updates must be a mapping.") self._validate_string_keys(explicit_updates, "state_updates") updates.update(deepcopy(dict(explicit_updates))) return updates def _state_gain_updates( self, *, agent: Agent, resource: Resource, actual_amount: float, ) -> dict[str, Any]: """ Build numeric state-gain updates. Supported parameters -------------------- state_gain_key: Single state key to increment. state_gain_multiplier: Multiplier applied to actual_amount. state_gain_updates: Mapping of state key to multiplier. max_state_value: Optional upper clamp for state_gain_key only. min_state_value: Optional lower clamp for state_gain_key only. Parameters ---------- agent: Acting agent. resource: Selected resource. actual_amount: Amount obtained or consumed. Returns ------- dict[str, Any] State gain updates. """ updates: dict[str, Any] = {} gain_key = self._optional_string_parameter( "state_gain_key", aliases=("gain_state_key", "output_state_key", "collection_state_key"), ) if gain_key is not None: multiplier = self._optional_finite_float_parameter( "state_gain_multiplier", aliases=("gain_multiplier", "output_multiplier"), default=1.0, ) current = self._numeric_agent_state(agent, gain_key, default=0.0) new_value = current + actual_amount * multiplier new_value = self._apply_optional_state_clamps(new_value) updates[gain_key] = new_value gain_updates = self.parameters.get("state_gain_updates") if gain_updates is not None: if not isinstance(gain_updates, Mapping): raise TypeError("state_gain_updates must be a mapping.") for key, multiplier_value in gain_updates.items(): if not isinstance(key, str) or not key.strip(): raise TypeError("state_gain_updates keys must be non-empty strings.") multiplier = self._normalize_finite_float( multiplier_value, f"state_gain_updates['{key}']", ) current = self._numeric_agent_state(agent, key, default=0.0) updates[key] = current + actual_amount * multiplier resource_type_key = self._optional_string_parameter( "resource_type_state_key", aliases=("resource_counter_key",), ) if resource_type_key is not None: current = self._numeric_agent_state(agent, resource_type_key, default=0.0) updates[resource_type_key] = current + actual_amount return updates def _inventory_state_update( self, *, agent: Agent, resource: Resource, actual_amount: float, ) -> dict[str, Any]: """ Build an inventory update for the obtained resource amount. Parameters ---------- agent: Acting agent. resource: Selected resource. actual_amount: Amount obtained. Returns ------- dict[str, Any] Inventory state update. """ inventory_key = self._inventory_key() item_key = self._inventory_item_key(resource) current_inventory = agent.state.get(inventory_key, {}) if current_inventory is None: current_inventory = {} if not isinstance(current_inventory, Mapping): raise TypeError( f"Agent state key '{inventory_key}' must be a mapping to store inventory." ) inventory = deepcopy(dict(current_inventory)) current_amount = inventory.get(item_key, 0.0) if isinstance(current_amount, bool) or not isinstance(current_amount, Real): raise TypeError( f"Inventory item '{item_key}' under '{inventory_key}' must be numeric." ) multiplier = self._optional_finite_float_parameter( "inventory_multiplier", aliases=("collection_multiplier", "storage_multiplier"), default=1.0, ) inventory[item_key] = float(current_amount) + actual_amount * multiplier return {inventory_key: inventory} def _action_cost_state_updates( self, *, agent: Agent, actual_amount: float, existing_updates: Mapping[str, Any], ) -> dict[str, Any]: """ Build state updates for optional action cost. Parameters ---------- agent: Acting agent. actual_amount: Actual interaction amount. existing_updates: State updates already planned. Returns ------- dict[str, Any] Cost-related state updates. """ key = self._cost_state_key() if key is None: return {} cost = self._action_cost(actual_amount) if cost <= ZERO_TOLERANCE: return {} if key in existing_updates: base_value = existing_updates[key] if isinstance(base_value, bool) or not isinstance(base_value, Real): raise TypeError( f"Cannot apply action cost to non-numeric pending state key '{key}'." ) current = float(base_value) else: current = self._numeric_agent_state(agent, key, default=0.0) return {key: current - cost} def _remove_resource_if_empty( self, *, resource: Resource, world: WorldProtocol, ) -> bool: """ Remove resource from world if configured and empty. Parameters ---------- resource: Resource that may be removed. world: World-like object. Returns ------- bool True if resource was removed. """ if not self._remove_empty_resource(): return False threshold = self._empty_threshold() if resource.amount > threshold: return False remover = getattr(world, "remove_resource", None) if not callable(remover): return False try: remover(resource.id, missing_ok=True) return True except TypeError: remover(resource.id) return True def _has_sufficient_action_budget( self, agent: Agent, amount: float, ) -> bool: """ Return whether the agent can pay optional action cost. Parameters ---------- agent: Acting agent. amount: Estimated interaction amount. Returns ------- bool True if no cost exists or the agent has enough budget. """ key = self._cost_state_key() if key is None: return True cost = self._action_cost(amount) if cost <= ZERO_TOLERANCE: return True current = self._numeric_agent_state(agent, key, default=0.0) return current >= cost def _action_cost(self, amount: float) -> float: """ Compute optional action cost. Parameters ---------- amount: Interaction amount. Returns ------- float Non-negative cost. """ normalized_amount = self._normalize_non_negative_float(amount, "amount") base_cost = self._optional_non_negative_float_parameter( "cost", aliases=("action_cost", "base_cost"), default=0.0, ) cost_per_unit = self._optional_non_negative_float_parameter( "cost_per_unit", aliases=("cost_per_amount",), default=0.0, ) return base_cost + cost_per_unit * normalized_amount def _success_reward(self, *, actual_amount: float) -> float: """ Compute success reward. Parameters ---------- actual_amount: Actual amount obtained or consumed. Returns ------- float Reward value. """ base_reward = self._optional_finite_float_parameter( "success_reward", aliases=("reward",), default=0.0, ) reward_per_unit = self._optional_finite_float_parameter( "reward_per_unit", aliases=("reward_per_amount",), default=0.0, ) return base_reward + reward_per_unit * actual_amount def _failure_reward(self) -> float: """ Return failure reward or penalty. Returns ------- float Failure reward value. """ return self._optional_finite_float_parameter( "failure_reward", aliases=("failure_penalty",), default=0.0, ) def _selector(self) -> dict[str, Any]: """ Build a normalized resource selector from parameters. Returns ------- dict[str, Any] Selector mapping. """ selector = self.parameters.get("selector", {}) if selector is None: selector = {} if not isinstance(selector, Mapping): raise TypeError("Resource selector must be a mapping.") normalized = deepcopy(dict(selector)) self._validate_string_keys(normalized, "selector") for parameter_key, selector_key in ( ("resource_id", "id"), ("target_resource_id", "id"), ("resource_type", "type"), ("target_resource_type", "type"), ): if parameter_key in self.parameters and self.parameters[parameter_key] is not None: normalized.setdefault(selector_key, self.parameters[parameter_key]) return normalized def _radius(self) -> float | None: """ Return optional interaction radius. Returns ------- float | None Non-negative radius or None. """ return self._optional_non_negative_float_parameter( "radius", aliases=("range", "interaction_radius"), default=None, ) def _selection_strategy(self) -> ResourceSelectionStrategy: """ Return resource selection strategy. Returns ------- ResourceSelectionStrategy Normalized selection strategy. """ raw_value = self.parameters.get( "selection_strategy", self.parameters.get("target_strategy", "nearest"), ) value = str(raw_value).strip().lower() if value in {"nearest", "closest"}: return "nearest" if value in {"farthest", "furthest"}: return "farthest" if value in {"largest", "most"}: return "largest" if value in {"smallest", "least"}: return "smallest" if value == "random": return "random" raise ValueError( "selection_strategy must be one of: nearest, farthest, largest, " "smallest, random." ) def _allow_partial(self) -> bool: """ Return whether partial interaction is allowed. Returns ------- bool True if partial depletion is allowed. """ return self._optional_bool_parameter("allow_partial", default=True) def _requires_resource(self) -> bool: """ Return whether a matching resource is required. Returns ------- bool True if behavior should fail when no resource is found. """ return self._optional_bool_parameter("require_resource", default=True) def _ignore_empty_resources(self) -> bool: """ Return whether empty resources should be ignored. Returns ------- bool True if empty resources are skipped. """ return self._optional_bool_parameter("ignore_empty_resources", default=True) def _depletes_resource(self, *, default: bool) -> bool: """ Return whether the behavior depletes the selected resource. Parameters ---------- default: Default depletion behavior. Returns ------- bool True if resource should be depleted. """ return self._optional_bool_parameter("deplete_resource", default=default) def _requires_available_amount_for_non_depleting(self) -> bool: """ Return whether non-depleting interactions require available resource amount. Returns ------- bool True if actual amount is capped by resource.amount. """ return self._optional_bool_parameter( "require_available_amount", default=True, ) def _stores_in_inventory(self, *, default: bool) -> bool: """ Return whether obtained amount should be stored in inventory. Parameters ---------- default: Default storage behavior. Returns ------- bool True if inventory should be updated. """ return self._optional_bool_parameter("store_in_inventory", default=default) def _inventory_key(self) -> str: """ Return the inventory state key. Returns ------- str Inventory key. """ return self._optional_string_parameter( "inventory_key", aliases=("storage_key",), default="inventory", ) def _inventory_item_key(self, resource: Resource) -> str: """ Return the inventory item key for a resource. Parameters ---------- resource: Selected resource. Returns ------- str Inventory item key. """ return self._optional_string_parameter( "inventory_item_key", aliases=("item_key", "stored_as"), default=resource.type, ) def _cost_state_key(self) -> str | None: """ Return optional state key used to pay action cost. Returns ------- str | None Cost state key or None. """ return self._optional_string_parameter( "cost_state_key", aliases=("state_cost_key", "budget_state_key"), default=None, ) def _remove_empty_resource(self) -> bool: """ Return whether empty resources should be removed from the world. Returns ------- bool True if empty resources should be removed. """ return self._optional_bool_parameter( "remove_empty_resource", default=False, ) def _empty_threshold(self) -> float: """ Return amount threshold considered empty. Returns ------- float Non-negative threshold. """ return self._optional_non_negative_float_parameter( "empty_threshold", aliases=("remove_threshold",), default=0.0, ) def _apply_optional_state_clamps(self, value: float) -> float: """ Apply optional min/max clamps to a numeric state value. Parameters ---------- value: Candidate state value. Returns ------- float Clamped value. """ new_value = value max_value = self._optional_finite_float_parameter( "max_state_value", aliases=("state_gain_max",), default=None, ) min_value = self._optional_finite_float_parameter( "min_state_value", aliases=("state_gain_min",), default=None, ) if max_value is not None: new_value = min(new_value, max_value) if min_value is not None: new_value = max(new_value, min_value) return new_value @staticmethod def _numeric_agent_state( agent: Agent, key: str, *, default: float, ) -> float: """ Read a numeric value from agent state. Parameters ---------- agent: Agent whose state should be read. key: State key. default: Default value when key is absent. Returns ------- float Numeric state value. """ value = agent.state.get(key, default) if isinstance(value, bool) or not isinstance(value, Real): raise TypeError( f"Agent '{agent.id}' state key '{key}' must be numeric." ) normalized = float(value) if not np.isfinite(normalized): raise ValueError( f"Agent '{agent.id}' state key '{key}' must be finite." ) return normalized def _optional_string_parameter( self, key: str, *, aliases: Sequence[str] = (), default: str | None = None, ) -> str | None: """ Return an optional string parameter. Parameters ---------- key: Primary parameter key. aliases: Alternative parameter keys. default: Default value. Returns ------- str | None Normalized string or None. """ for candidate_key in (key, *aliases): if candidate_key in self.parameters and self.parameters[candidate_key] is not None: value = self.parameters[candidate_key] if not isinstance(value, str) or not value.strip(): raise ValueError(f"Parameter '{candidate_key}' must be a non-empty string.") return value.strip() return default def _optional_bool_parameter(self, key: str, *, default: bool) -> bool: """ Return an optional boolean parameter. Parameters ---------- key: Parameter key. default: Default value. Returns ------- bool Boolean parameter. """ if key not in self.parameters or self.parameters[key] is None: return default value = self.parameters[key] if not isinstance(value, bool): raise TypeError(f"Parameter '{key}' must be a boolean.") return value def _optional_finite_float_parameter( self, key: str, *, aliases: Sequence[str] = (), default: float | None, ) -> float | None: """ Return an optional finite float parameter. Parameters ---------- key: Primary parameter key. aliases: Alternative parameter keys. default: Default value. Returns ------- float | None Normalized finite float or None. """ for candidate_key in (key, *aliases): if candidate_key in self.parameters and self.parameters[candidate_key] is not None: return self._normalize_finite_float( self.parameters[candidate_key], candidate_key, ) return default def _optional_non_negative_float_parameter( self, key: str, *, aliases: Sequence[str] = (), default: float | None, ) -> float | None: """ Return an optional non-negative float parameter. Parameters ---------- key: Primary parameter key. aliases: Alternative parameter keys. default: Default value. Returns ------- float | None Normalized non-negative float or None. """ value = self._optional_finite_float_parameter( key, aliases=aliases, default=default, ) if value is not None and value < 0.0: raise ValueError(f"Parameter '{key}' cannot be negative.") return value @staticmethod def _normalize_finite_float(value: Any, label: str) -> float: """ Normalize a finite float. Parameters ---------- value: Candidate value. label: Human-readable label. Returns ------- float Normalized finite float. """ if isinstance(value, bool) or not isinstance(value, Real): raise TypeError(f"{label} must be a real number.") normalized = float(value) if not np.isfinite(normalized): raise ValueError(f"{label} must be finite.") return normalized def _normalize_non_negative_float(self, value: Any, label: str) -> float: """ Normalize a non-negative finite float. Parameters ---------- value: Candidate value. label: Human-readable label. Returns ------- float Non-negative finite float. """ normalized = self._normalize_finite_float(value, label) if normalized < 0.0: raise ValueError(f"{label} cannot be negative.") return normalized @staticmethod def _value_matches(value: Any, expected: Any) -> bool: """ Return whether a value matches a scalar or collection filter. Parameters ---------- value: Candidate value. expected: Scalar or collection of allowed values. Returns ------- bool True if matched. """ if isinstance(expected, (str, bytes)): return value == expected if isinstance(expected, set): return value in expected if isinstance(expected, frozenset): return value in expected if isinstance(expected, Sequence): return value in expected return value == expected @staticmethod def _mapping_contains(actual: Any, expected: Any) -> bool: """ Return whether a mapping contains expected key/value pairs. Parameters ---------- actual: Candidate mapping. expected: Expected mapping. Returns ------- bool True if all expected pairs match. """ if not isinstance(actual, Mapping) or not isinstance(expected, Mapping): return False for key, expected_value in expected.items(): if key not in actual: return False if actual[key] != expected_value: return False return True @staticmethod def _rng(world: WorldProtocol) -> np.random.Generator: """ Return world RNG or fallback generator. Parameters ---------- world: World-like object. Returns ------- np.random.Generator Random generator. """ rng = getattr(world, "rng", None) if isinstance(rng, np.random.Generator): return rng logger.warning( "World does not expose a NumPy Generator as rng; using fallback RNG." ) return np.random.default_rng() @staticmethod def _validate_string_keys(mapping: Mapping[str, Any], label: str) -> None: """ Validate mapping keys are strings. Parameters ---------- mapping: Mapping to validate. label: Human-readable label. Raises ------ TypeError If any key is not a string. """ for key in mapping: if not isinstance(key, str): raise TypeError(f"{label} keys must be strings.") @dataclass(slots=True) class ConsumeBehavior(ResourceInteractionBehavior): """ Consume a resource. This behavior depletes a selected resource and can optionally increase agent state through ``state_gain_key`` or ``state_gain_updates``. Common parameters ----------------- selector: Resource selector. amount: Amount to consume. May be a number or ``"all"``. radius: Optional maximum distance. allow_partial: Whether partial consumption is allowed. state_gain_key: Optional state key to increment. state_gain_multiplier: Multiplier applied to actual consumed amount. deplete_resource: Whether the resource should be depleted. Defaults to True. """ def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute resource consumption. Parameters ---------- agent: Agent consuming the resource. world: World containing resources. Returns ------- BehaviorResult Consumption result. """ return self._execute_resource_interaction( agent=agent, world=world, interaction_kind="consume", success_message="Agent consumed resource.", failure_message="ConsumeBehavior found no matching resource.", store_in_inventory_default=False, deplete_resource_default=True, ) @dataclass(slots=True) class EatBehavior(ConsumeBehavior): """ Eat a resource. EatBehavior is a semantic alias around ConsumeBehavior. It does not assume the resource is food and does not assume the agent has an ``energy`` state. DSL parameters decide which resource is selected and which state keys change. """ def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute eating behavior. Parameters ---------- agent: Agent eating a resource. world: World containing resources. Returns ------- BehaviorResult Eating result. """ return self._execute_resource_interaction( agent=agent, world=world, interaction_kind="eat", success_message="Agent ate resource.", failure_message="EatBehavior found no matching resource.", store_in_inventory_default=False, deplete_resource_default=True, ) @dataclass(slots=True) class HarvestBehavior(ResourceInteractionBehavior): """ Harvest a resource into agent state or inventory. Harvesting defaults to storing the obtained amount in an inventory-like mapping under ``agent.state["inventory"]``. This can be disabled with: {"store_in_inventory": false} Common parameters ----------------- selector: Resource selector. amount: Amount to harvest. radius: Optional maximum distance. inventory_key: Agent state key containing inventory mapping. Defaults to ``"inventory"``. inventory_item_key: Item key inside inventory. Defaults to selected resource type. state_gain_key: Optional numeric state key to increment. """ def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute harvesting behavior. Parameters ---------- agent: Agent harvesting the resource. world: World containing resources. Returns ------- BehaviorResult Harvest result. """ return self._execute_resource_interaction( agent=agent, world=world, interaction_kind="harvest", success_message="Agent harvested resource.", failure_message="HarvestBehavior found no matching resource.", store_in_inventory_default=True, deplete_resource_default=True, ) @dataclass(slots=True) class MineBehavior(ResourceInteractionBehavior): """ Mine or extract a resource into agent state or inventory. MineBehavior is domain-agnostic. It can extract ore, fuel, knowledge, compute, mana, or any other resource type defined by the DSL. By default, mined amount is stored in inventory. """ def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute mining behavior. Parameters ---------- agent: Agent mining the resource. world: World containing resources. Returns ------- BehaviorResult Mine result. """ return self._execute_resource_interaction( agent=agent, world=world, interaction_kind="mine", success_message="Agent mined resource.", failure_message="MineBehavior found no matching resource.", store_in_inventory_default=True, deplete_resource_default=True, ) @dataclass(slots=True) class CollectBehavior(ResourceInteractionBehavior): """ Collect a resource into agent state or inventory. CollectBehavior defaults to storing obtained amount in inventory. It can also support non-depleting collection by setting: {"deplete_resource": false} This is useful for resources like public knowledge, signals, influence, or observations where collection should not necessarily reduce the world resource. """ def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute collection behavior. Parameters ---------- agent: Agent collecting the resource. world: World containing resources. Returns ------- BehaviorResult Collection result. """ return self._execute_resource_interaction( agent=agent, world=world, interaction_kind="collect", success_message="Agent collected resource.", failure_message="CollectBehavior found no matching resource.", store_in_inventory_default=True, deplete_resource_default=True, ) Consume = ConsumeBehavior Eat = EatBehavior Harvest = HarvestBehavior Mine = MineBehavior Collect = CollectBehavior __all__ = [ "Collect", "CollectBehavior", "Consume", "ConsumeBehavior", "DEFAULT_INTERACTION_AMOUNT", "Eat", "EatBehavior", "Harvest", "HarvestBehavior", "Mine", "MineBehavior", "ResourceCandidate", "ResourceInteractionBehavior", "ResourceInteractionComputation", "ResourceSelectionStrategy", ]