""" behaviors.movement ================== Generic movement behaviors for WorldSmithAI. This module implements domain-agnostic movement primitives: - MoveBehavior - WanderBehavior - SeekBehavior - FleeBehavior These behaviors can be attached to any Agent regardless of whether the agent represents an animal, merchant, scientist, robot, spaceship, villager, company, or fantasy creature. The behaviors operate on numeric position vectors and do not assume a grid, specific terrain type, species, biome, civilization, or world theme. Example DSL-inspired behavior parameters ---------------------------------------- Move in a fixed direction: { "id": "move_east", "type": "move", "parameters": { "direction": [1.0, 0.0], "step_size": 1.0 } } Wander randomly: { "id": "wander", "type": "wander", "parameters": { "step_size": 0.5 } } Seek nearest food resource: { "id": "seek_food", "type": "seek", "parameters": { "target": "resource", "selector": {"type": "food"}, "step_size": 1.0, "radius": 10.0 } } Flee nearest predator agent: { "id": "flee_predator", "type": "flee", "parameters": { "target": "agent", "selector": {"type": "predator"}, "step_size": 1.5, "radius": 5.0 } } Minimal usage example --------------------- from core.agent import Agent from behaviors.movement import WanderBehavior agent = Agent( id="dragon_1", type="dragon", position=[0.0, 0.0], state={"fuel": 10.0}, behaviors={ "wander": WanderBehavior( id="wander", parameters={ "step_size": 1.0, "cost_state_key": "fuel", "cost_per_unit": 0.1, }, ) }, ) result = agent.step(world) print(agent.position) """ 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 logger = logging.getLogger(__name__) PositionInput: TypeAlias = Sequence[float] | NDArray[np.float64] TargetKind: TypeAlias = Literal["agent", "resource"] TargetStrategy: TypeAlias = Literal["nearest", "farthest", "random"] DEFAULT_STEP_SIZE: float = 1.0 ZERO_TOLERANCE: float = 1.0e-12 @dataclass(frozen=True, slots=True) class MovementComputation: """ Structured description of one movement computation. Attributes ---------- movement_kind: Human-readable movement category, such as ``"move"``, ``"wander"``, ``"seek"``, or ``"flee"``. previous_position: Agent position before movement. new_position: Agent position after movement. requested_delta: Delta requested before optional bounds are applied. actual_delta: Delta actually applied after optional bounds are applied. requested_distance: Length of the requested movement delta. actual_distance: Length of the applied movement delta. target_position: Optional target or threat position used by seek/flee/move-to-target. reached_target: Whether the movement reached its target position. bounded: Whether bounds changed the requested new position. bounds_mode: Optional bounds mode, such as ``"clip"`` or ``"wrap"``. """ movement_kind: str previous_position: NDArray[np.float64] new_position: NDArray[np.float64] requested_delta: NDArray[np.float64] actual_delta: NDArray[np.float64] requested_distance: float actual_distance: float target_position: NDArray[np.float64] | None = None reached_target: bool = False bounded: bool = False bounds_mode: str | None = None def __post_init__(self) -> None: """ Validate and normalize movement computation fields. Raises ------ ValueError If vectors are invalid or distances are non-finite. TypeError If boolean flags are invalid. """ previous_position = self._normalize_vector( self.previous_position, "previous_position", ) new_position = self._normalize_vector(self.new_position, "new_position") requested_delta = self._normalize_vector( self.requested_delta, "requested_delta", ) actual_delta = self._normalize_vector(self.actual_delta, "actual_delta") if previous_position.shape != new_position.shape: raise ValueError("previous_position and new_position shapes must match.") if previous_position.shape != requested_delta.shape: raise ValueError("requested_delta shape must match position shape.") if previous_position.shape != actual_delta.shape: raise ValueError("actual_delta shape must match position shape.") target_position = None if self.target_position is not None: target_position = self._normalize_vector( self.target_position, "target_position", ) if target_position.shape != previous_position.shape: raise ValueError("target_position shape must match position shape.") if not np.isfinite(self.requested_distance): raise ValueError("requested_distance must be finite.") if not np.isfinite(self.actual_distance): raise ValueError("actual_distance must be finite.") if self.requested_distance < 0.0: raise ValueError("requested_distance cannot be negative.") if self.actual_distance < 0.0: raise ValueError("actual_distance cannot be negative.") if not isinstance(self.reached_target, bool): raise TypeError("reached_target must be a boolean.") if not isinstance(self.bounded, bool): raise TypeError("bounded must be a boolean.") object.__setattr__(self, "movement_kind", str(self.movement_kind)) object.__setattr__(self, "previous_position", previous_position) object.__setattr__(self, "new_position", new_position) object.__setattr__(self, "requested_delta", requested_delta) object.__setattr__(self, "actual_delta", actual_delta) object.__setattr__(self, "requested_distance", float(self.requested_distance)) object.__setattr__(self, "actual_distance", float(self.actual_distance)) object.__setattr__(self, "target_position", target_position) def to_metadata(self) -> dict[str, Any]: """ Convert the movement computation into JSON-friendly metadata. Returns ------- dict[str, Any] Serializable movement metadata. """ return { "movement_kind": self.movement_kind, "previous_position": self.previous_position.tolist(), "new_position": self.new_position.tolist(), "requested_delta": self.requested_delta.tolist(), "actual_delta": self.actual_delta.tolist(), "requested_distance": self.requested_distance, "actual_distance": self.actual_distance, "target_position": ( None if self.target_position is None else self.target_position.tolist() ), "reached_target": self.reached_target, "bounded": self.bounded, "bounds_mode": self.bounds_mode, } @staticmethod def _normalize_vector( value: PositionInput, label: str, ) -> NDArray[np.float64]: """ Normalize a vector into a finite one-dimensional NumPy array. Parameters ---------- value: Candidate vector. label: Human-readable label for error messages. Returns ------- NDArray[np.float64] Normalized vector. """ vector = np.asarray(value, dtype=np.float64) if vector.ndim != 1: raise ValueError(f"{label} must be one-dimensional.") if vector.size == 0: raise ValueError(f"{label} cannot be empty.") if not np.all(np.isfinite(vector)): raise ValueError(f"{label} must contain only finite values.") return vector @dataclass(frozen=True, slots=True) class TargetCandidate: """ Candidate target selected from the world. Attributes ---------- kind: Target category: ``"agent"`` or ``"resource"``. id: Candidate object id. type: Candidate object type. position: Candidate position. distance: Distance from the acting agent. metadata: Optional candidate metadata for debugging, metrics, or narration. """ kind: TargetKind id: str type: str position: NDArray[np.float64] distance: float metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize target candidate fields. Raises ------ ValueError If id, type, position, or distance are invalid. TypeError If metadata is invalid. """ if self.kind not in {"agent", "resource"}: raise ValueError("TargetCandidate.kind must be 'agent' or 'resource'.") if not isinstance(self.id, str) or not self.id.strip(): raise ValueError("TargetCandidate.id must be a non-empty string.") if not isinstance(self.type, str) or not self.type.strip(): raise ValueError("TargetCandidate.type must be a non-empty string.") position = MovementComputation._normalize_vector( self.position, "TargetCandidate.position", ) if not np.isfinite(self.distance): raise ValueError("TargetCandidate.distance must be finite.") if self.distance < 0.0: raise ValueError("TargetCandidate.distance cannot be negative.") if not isinstance(self.metadata, Mapping): raise TypeError("TargetCandidate.metadata must be a mapping.") object.__setattr__(self, "id", self.id.strip()) object.__setattr__(self, "type", self.type.strip()) object.__setattr__(self, "position", position) object.__setattr__(self, "distance", float(self.distance)) object.__setattr__(self, "metadata", dict(self.metadata)) def to_metadata(self) -> dict[str, Any]: """ Convert the target candidate into JSON-friendly metadata. Returns ------- dict[str, Any] Serializable target metadata. """ return { "kind": self.kind, "id": self.id, "type": self.type, "position": self.position.tolist(), "distance": self.distance, "metadata": deepcopy(dict(self.metadata)), } @dataclass(slots=True) class MovementBehavior(Behavior): """ Base class for concrete movement behaviors. This class provides reusable vector math, bounds handling, target resolution, random direction generation, and optional movement cost support. It remains abstract because concrete subclasses must implement ``execute(...)``. """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check generic movement preconditions. Parameters ---------- agent: Agent attempting to move. world: World in which movement would occur. Returns ------- bool True when the agent can pay the estimated movement cost. """ estimated_distance = self._step_size() return self._has_sufficient_state_budget(agent, estimated_distance) def _attempt_movement( self, *, agent: Agent, world: WorldProtocol, delta: NDArray[np.float64], movement_kind: str, message: str, target_position: NDArray[np.float64] | None = None, reached_target: bool = False, extra_metadata: Mapping[str, Any] | None = None, ) -> BehaviorResult: """ Apply a movement delta to an agent. Parameters ---------- agent: Agent being moved. world: World in which movement occurs. delta: Requested movement delta. movement_kind: Movement kind label. message: Human-readable success message. target_position: Optional target position. reached_target: Whether this movement reached its target. extra_metadata: Additional result metadata. Returns ------- BehaviorResult Behavior execution result. """ requested_delta = self._normalize_vector(delta, "delta") if requested_delta.shape != agent.position.shape: return self.failure( reward=self._failure_reward(), message=( "Movement delta dimensionality does not match agent position." ), metadata={ "agent_position_dimension": int(agent.position.size), "delta_dimension": int(requested_delta.size), }, ) requested_distance = float(np.linalg.norm(requested_delta)) if not self._has_sufficient_state_budget(agent, requested_distance): return self.failure( reward=self._failure_reward(), message="Agent does not have sufficient state budget for movement.", metadata={ "required_cost": self._movement_cost(requested_distance), "cost_state_key": self._cost_state_key(), }, ) previous_position = agent.position.copy() requested_new_position = previous_position + requested_delta bounded_position, bounded, bounds_mode = self._apply_bounds( requested_new_position, world, ) actual_delta = bounded_position - previous_position actual_distance = float(np.linalg.norm(actual_delta)) agent.set_position(bounded_position) computation = MovementComputation( movement_kind=movement_kind, previous_position=previous_position, new_position=bounded_position, requested_delta=requested_delta, actual_delta=actual_delta, requested_distance=requested_distance, actual_distance=actual_distance, target_position=target_position, reached_target=reached_target, bounded=bounded, bounds_mode=bounds_mode, ) state_updates = self._movement_cost_state_updates( agent, actual_distance=actual_distance, ) metadata = computation.to_metadata() if extra_metadata is not None: metadata.update(deepcopy(dict(extra_metadata))) return self.success( reward=self._success_reward(), state_updates=state_updates, memory_updates={ "movement": metadata, }, message=message, metadata=metadata, ) def _delta_toward( self, *, current_position: NDArray[np.float64], target_position: NDArray[np.float64], step_size: float, ) -> tuple[NDArray[np.float64], bool]: """ Compute a movement delta toward a target. Parameters ---------- current_position: Current agent position. target_position: Desired target position. step_size: Maximum movement distance. Returns ------- tuple[NDArray[np.float64], bool] Movement delta and whether the target will be reached. """ if current_position.shape != target_position.shape: raise ValueError("target_position dimensionality must match agent position.") offset = target_position - current_position distance = float(np.linalg.norm(offset)) if distance <= ZERO_TOLERANCE: return np.zeros_like(current_position), True if distance <= step_size: return offset, True return offset / distance * step_size, False def _delta_away( self, *, current_position: NDArray[np.float64], threat_position: NDArray[np.float64], step_size: float, world: WorldProtocol, ) -> NDArray[np.float64]: """ Compute a movement delta away from a threat. Parameters ---------- current_position: Current agent position. threat_position: Threat position. step_size: Movement distance. world: World used for deterministic random fallback when positions overlap. Returns ------- NDArray[np.float64] Movement delta away from the threat. """ if current_position.shape != threat_position.shape: raise ValueError("threat_position dimensionality must match agent position.") offset = current_position - threat_position distance = float(np.linalg.norm(offset)) if distance <= ZERO_TOLERANCE: direction = self._random_unit_vector(world, current_position.size) return direction * step_size return offset / distance * step_size def _resolve_target( self, *, agent: Agent, world: WorldProtocol, default_kind: TargetKind, position_keys: Sequence[str], ) -> tuple[NDArray[np.float64], TargetCandidate | None] | None: """ Resolve an explicit position or world-selected target. Parameters ---------- agent: Acting agent. world: World in which the agent acts. default_kind: Default target kind when no target category is specified. position_keys: Parameter keys that may contain explicit target positions. Returns ------- tuple[NDArray[np.float64], TargetCandidate | None] | None Target position and optional candidate metadata, or None. """ for key in position_keys: if key in self.parameters and self.parameters[key] is not None: return self._normalize_vector(self.parameters[key], key), None target_kind = self._target_kind(default_kind) selector = self._selector() radius = self._optional_non_negative_float_parameter("radius") candidate = self._select_target_candidate( agent=agent, world=world, kind=target_kind, selector=selector, radius=radius, ) if candidate is None: return None return candidate.position, candidate def _select_target_candidate( self, *, agent: Agent, world: WorldProtocol, kind: TargetKind, selector: Mapping[str, Any], radius: float | None, ) -> TargetCandidate | None: """ Select a target candidate from the world. Parameters ---------- agent: Acting agent. world: World to inspect. kind: Target kind. selector: Generic selector mapping. radius: Optional maximum distance. Returns ------- TargetCandidate | None Selected candidate, if any. """ objects = self._objects_from_world(world=world, kind=kind, selector=selector) candidates: list[TargetCandidate] = [] for candidate_object in objects: if kind == "agent" and getattr(candidate_object, "id", None) == agent.id: continue if not self._matches_simple_selector(candidate_object, selector): continue position = getattr(candidate_object, "position", None) if position is None: continue try: candidate_position = self._normalize_vector(position, "candidate_position") except ValueError: logger.debug("Skipping target candidate with invalid position.") continue if candidate_position.shape != agent.position.shape: continue distance = float(np.linalg.norm(candidate_position - agent.position)) if radius is not None and distance > radius: continue candidate_id = str(getattr(candidate_object, "id", "unknown")) candidate_type = str(getattr(candidate_object, "type", "unknown")) candidates.append( TargetCandidate( kind=kind, id=candidate_id, type=candidate_type, position=candidate_position, distance=distance, metadata={ "class": candidate_object.__class__.__name__, }, ) ) if not candidates: return None strategy = self._target_strategy() if strategy == "nearest": return min(candidates, key=lambda item: (item.distance, item.id)) if strategy == "farthest": return max(candidates, key=lambda item: (item.distance, item.id)) if strategy == "random": rng = self._rng(world) index = int(rng.integers(0, len(candidates))) return candidates[index] raise ValueError(f"Unsupported target strategy: {strategy}") def _objects_from_world( self, *, world: WorldProtocol, kind: TargetKind, selector: Mapping[str, Any], ) -> list[Any]: """ Retrieve candidate objects from the world. Parameters ---------- world: World to inspect. kind: Target kind. selector: Generic selector mapping. Returns ------- list[Any] Candidate world objects. """ if kind == "resource": selector_method = getattr(world, "select_resources", None) collection = getattr(world, "resources", {}) else: selector_method = getattr(world, "select_agents", None) collection = getattr(world, "agents", {}) if callable(selector_method): try: selected = selector_method(selector) except TypeError: selected = selector_method(selector=selector) return list(selected) if isinstance(collection, Mapping): return list(collection.values()) if isinstance(collection, Sequence) and not isinstance(collection, (str, bytes)): return list(collection) return [] def _matches_simple_selector( self, candidate: Any, selector: Mapping[str, Any], ) -> bool: """ Apply a simple selector to a candidate object. This fallback is used when a world does not provide its own ``select_agents`` or ``select_resources`` implementation. Parameters ---------- candidate: Candidate object. selector: Selector mapping. Returns ------- bool True if the candidate matches. """ if not selector: return True candidate_id = getattr(candidate, "id", None) candidate_type = getattr(candidate, "type", None) if "id" in selector and candidate_id != selector["id"]: return False if "ids" in selector and candidate_id not in self._as_set(selector["ids"]): return False if "type" in selector and candidate_type != selector["type"]: return False if "types" in selector and candidate_type not in self._as_set(selector["types"]): return False if "alive" in selector and hasattr(candidate, "alive"): if bool(getattr(candidate, "alive")) is not bool(selector["alive"]): return False if "state" in selector: state = getattr(candidate, "state", {}) if not self._mapping_contains(state, selector["state"]): return False if "metadata" in selector: metadata = getattr(candidate, "metadata", {}) if not self._mapping_contains(metadata, selector["metadata"]): return False return True def _apply_bounds( self, position: NDArray[np.float64], world: WorldProtocol, ) -> tuple[NDArray[np.float64], bool, str | None]: """ Apply optional bounds to a position. Bounds may be supplied in behavior parameters or in ``world.metadata``: {"min": [0, 0], "max": [10, 10]} or: [[0, 10], [0, 10]] Parameters ---------- position: Requested new position. world: World that may provide metadata bounds. Returns ------- tuple[NDArray[np.float64], bool, str | None] Bounded position, whether bounds changed it, and bounds mode. """ raw_bounds = self.parameters.get("bounds") if raw_bounds is None: world_metadata = getattr(world, "metadata", {}) if isinstance(world_metadata, Mapping): raw_bounds = world_metadata.get("bounds") if raw_bounds is None: return position, False, None lower, upper = self._normalize_bounds(raw_bounds, position.size) mode = str(self.parameters.get("bounds_mode", "clip")).strip().lower() if mode == "none": return position, False, "none" if mode == "clip": bounded = np.minimum(upper, np.maximum(lower, position)) elif mode == "wrap": widths = upper - lower bounded = position.copy() for index, width in enumerate(widths): if width <= ZERO_TOLERANCE: bounded[index] = lower[index] else: bounded[index] = lower[index] + np.mod( bounded[index] - lower[index], width, ) else: raise ValueError("bounds_mode must be one of: 'clip', 'wrap', 'none'.") changed = not bool(np.allclose(position, bounded)) return bounded, changed, mode def _normalize_bounds( self, raw_bounds: Any, dimension: int, ) -> tuple[NDArray[np.float64], NDArray[np.float64]]: """ Normalize bounds into lower and upper vectors. Parameters ---------- raw_bounds: Bounds mapping or sequence. dimension: Expected dimensionality. Returns ------- tuple[NDArray[np.float64], NDArray[np.float64]] Lower and upper bound vectors. """ if isinstance(raw_bounds, Mapping): if "min" not in raw_bounds or "max" not in raw_bounds: raise KeyError("Bounds mapping must contain 'min' and 'max'.") lower = self._normalize_vector(raw_bounds["min"], "bounds.min") upper = self._normalize_vector(raw_bounds["max"], "bounds.max") elif isinstance(raw_bounds, Sequence) and not isinstance( raw_bounds, (str, bytes), ): if len(raw_bounds) != dimension: raise ValueError( "Bounds sequence length must match position dimensionality." ) pairs = list(raw_bounds) lower = np.asarray([pair[0] for pair in pairs], dtype=np.float64) upper = np.asarray([pair[1] for pair in pairs], dtype=np.float64) else: raise TypeError("bounds must be a mapping or sequence of [min, max] pairs.") if lower.shape != (dimension,) or upper.shape != (dimension,): raise ValueError("Bounds dimensionality must match position dimensionality.") if not np.all(np.isfinite(lower)) or not np.all(np.isfinite(upper)): raise ValueError("Bounds must contain only finite values.") if np.any(upper < lower): raise ValueError("Each upper bound must be greater than or equal to lower.") return lower, upper def _movement_cost_state_updates( self, agent: Agent, *, actual_distance: float, ) -> dict[str, Any]: """ Build state updates for optional movement cost. Parameters ---------- agent: Agent being moved. actual_distance: Actual distance moved. Returns ------- dict[str, Any] State updates to apply after movement. """ key = self._cost_state_key() if key is None: return {} total_cost = self._movement_cost(actual_distance) if total_cost <= 0.0: return {} current = self._numeric_agent_state(agent, key, default=0.0) return {key: current - total_cost} def _has_sufficient_state_budget( self, agent: Agent, estimated_distance: float, ) -> bool: """ Return whether an agent can pay optional movement cost. Parameters ---------- agent: Candidate moving agent. estimated_distance: Estimated movement distance. Returns ------- bool True if the agent has sufficient state budget. """ key = self._cost_state_key() if key is None: return True required_cost = self._movement_cost(estimated_distance) if required_cost <= 0.0: return True current = self._numeric_agent_state(agent, key, default=0.0) return current >= required_cost def _movement_cost(self, distance: float) -> float: """ Compute movement cost for a distance. Parameters ---------- distance: Movement distance. Returns ------- float Non-negative total movement cost. """ normalized_distance = self._normalize_non_negative_float(distance, "distance") base_cost = self._optional_non_negative_float_parameter("cost", default=0.0) cost_amount = self._optional_non_negative_float_parameter( "cost_amount", default=0.0, ) cost_per_unit = self._optional_non_negative_float_parameter( "cost_per_unit", default=0.0, ) charge_on_noop = self._optional_bool_parameter( "charge_cost_on_noop", default=False, ) if normalized_distance <= ZERO_TOLERANCE and not charge_on_noop: return 0.0 return base_cost + cost_amount + cost_per_unit * normalized_distance def _cost_state_key(self) -> str | None: """ Return the optional state key used to pay movement cost. Returns ------- str | None State key or None when movement has no state cost. """ for key in ("cost_state_key", "state_cost_key", "budget_state_key"): value = self.parameters.get(key) if value is not None: if not isinstance(value, str) or not value.strip(): raise ValueError(f"{key} must be a non-empty string.") return value.strip() return None def _step_size(self) -> float: """ Return the movement step size. Returns ------- float Non-negative movement step size. """ for key in ("step_size", "speed", "distance"): if key in self.parameters and self.parameters[key] is not None: return self._normalize_non_negative_float(self.parameters[key], key) return DEFAULT_STEP_SIZE def _selector(self) -> dict[str, Any]: """ Return the target selector. Returns ------- dict[str, Any] Normalized selector mapping. """ selector = self.parameters.get("selector", {}) if selector is None: return {} if not isinstance(selector, Mapping): raise TypeError("Movement selector must be a mapping.") normalized = dict(selector) for key in normalized: if not isinstance(key, str): raise TypeError("Movement selector keys must be strings.") if "target_id" in self.parameters: normalized.setdefault("id", self.parameters["target_id"]) if "target_type" in self.parameters: normalized.setdefault("type", self.parameters["target_type"]) return normalized def _target_kind(self, default: TargetKind) -> TargetKind: """ Return the configured target kind. Parameters ---------- default: Default target kind. Returns ------- TargetKind Normalized target kind. """ raw_value = self.parameters.get( "target", self.parameters.get( "target_kind", self.parameters.get("target_category", default), ), ) value = str(raw_value).strip().lower() if value in {"agent", "agents"}: return "agent" if value in {"resource", "resources"}: return "resource" raise ValueError( "Movement target kind must be one of: 'agent', 'agents', " "'resource', 'resources'." ) def _target_strategy(self) -> TargetStrategy: """ Return target selection strategy. Returns ------- TargetStrategy Normalized strategy. """ raw_value = 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 == "random": return "random" raise ValueError( "target_strategy must be one of: 'nearest', 'farthest', 'random'." ) def _success_reward(self) -> float: """ Return success reward. Returns ------- float Reward for successful movement. """ if "success_reward" in self.parameters: return self._normalize_finite_float( self.parameters["success_reward"], "success_reward", ) if "reward" in self.parameters: return self._normalize_finite_float(self.parameters["reward"], "reward") return 0.0 def _failure_reward(self) -> float: """ Return failure reward. Returns ------- float Reward or penalty for failed movement. """ if "failure_reward" in self.parameters: return self._normalize_finite_float( self.parameters["failure_reward"], "failure_reward", ) return 0.0 def _random_unit_vector( self, world: WorldProtocol, dimension: int, ) -> NDArray[np.float64]: """ Generate a random unit vector. Parameters ---------- world: World providing optional deterministic RNG. dimension: Desired vector dimensionality. Returns ------- NDArray[np.float64] Random unit vector. """ if dimension <= 0: raise ValueError("dimension must be positive.") rng = self._rng(world) for _ in range(16): vector = rng.normal(size=dimension) norm = float(np.linalg.norm(vector)) if norm > ZERO_TOLERANCE: return np.asarray(vector / norm, dtype=np.float64) fallback = np.zeros(dimension, dtype=np.float64) fallback[0] = 1.0 return fallback @staticmethod def _rng(world: WorldProtocol) -> np.random.Generator: """ Return the world's RNG or a 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 an unseeded fallback generator." ) return np.random.default_rng() @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 when actual contains all expected pairs. """ if not isinstance(actual, Mapping) or not isinstance(expected, Mapping): return False for key, value in expected.items(): if key not in actual: return False if actual[key] != value: return False return True @staticmethod def _as_set(value: Any) -> set[Any]: """ Normalize a scalar or sequence into a set. Parameters ---------- value: Scalar or sequence value. Returns ------- set[Any] Set of values. """ if isinstance(value, (str, bytes)): return {value} if isinstance(value, Sequence): return set(value) if isinstance(value, set): return value return {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 if the 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_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_non_negative_float_parameter( self, key: str, *, default: float | None = None, ) -> float | None: """ Return an optional non-negative float parameter. Parameters ---------- key: Parameter key. default: Default value. Returns ------- float | None Normalized value. """ if key not in self.parameters or self.parameters[key] is None: return default return self._normalize_non_negative_float(self.parameters[key], key) @staticmethod def _normalize_vector(value: PositionInput, label: str) -> NDArray[np.float64]: """ Normalize a vector into a finite one-dimensional NumPy array. Parameters ---------- value: Candidate vector. label: Human-readable label. Returns ------- NDArray[np.float64] Normalized vector. """ return MovementComputation._normalize_vector(value, label) @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 Normalized non-negative finite float. """ normalized = self._normalize_finite_float(value, label) if normalized < 0.0: raise ValueError(f"{label} cannot be negative.") return normalized @dataclass(slots=True) class MoveBehavior(MovementBehavior): """ Move an agent by a delta, direction, or target position. Parameter precedence -------------------- 1. ``target_position`` / ``destination`` / ``position`` 2. ``direction`` 3. ``delta`` Common parameters ----------------- step_size: Maximum movement distance when using direction or target movement. direction: Direction vector. delta: Direct displacement vector. target_position: Destination position. Movement will not overshoot the target. normalize_direction: Whether direction should be normalized before multiplying by step size. Defaults to True. limit_delta: Whether direct delta movement should be capped by step_size. Defaults to False. """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check whether the behavior has a movement specification and budget. Parameters ---------- agent: Agent attempting to move. world: World in which movement would occur. Returns ------- bool True if movement can be attempted. """ if not self._has_movement_specification(): return False estimated_distance = self._estimated_requested_distance(agent) return self._has_sufficient_state_budget(agent, estimated_distance) def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute direct movement. Parameters ---------- agent: Agent executing the movement. world: World in which movement occurs. Returns ------- BehaviorResult Movement execution result. """ try: step_size = self._step_size() target_position = self._target_position_from_parameters() if target_position is not None: delta, reached_target = self._delta_toward( current_position=agent.position, target_position=target_position, step_size=step_size, ) return self._attempt_movement( agent=agent, world=world, delta=delta, movement_kind="move", message=( "Agent moved toward target position." if not reached_target else "Agent reached target position." ), target_position=target_position, reached_target=reached_target, ) if "direction" in self.parameters and self.parameters["direction"] is not None: direction = self._normalize_vector( self.parameters["direction"], "direction", ) normalize_direction = self._optional_bool_parameter( "normalize_direction", default=True, ) if direction.shape != agent.position.shape: return self.failure( reward=self._failure_reward(), message="Direction dimensionality does not match position.", ) if normalize_direction: norm = float(np.linalg.norm(direction)) if norm <= ZERO_TOLERANCE: return self.failure( reward=self._failure_reward(), message="Direction vector cannot be zero.", ) delta = direction / norm * step_size else: delta = direction * step_size return self._attempt_movement( agent=agent, world=world, delta=delta, movement_kind="move", message="Agent moved in configured direction.", ) if "delta" in self.parameters and self.parameters["delta"] is not None: delta = self._normalize_vector(self.parameters["delta"], "delta") if self._optional_bool_parameter("limit_delta", default=False): distance = float(np.linalg.norm(delta)) if distance > step_size and distance > ZERO_TOLERANCE: delta = delta / distance * step_size return self._attempt_movement( agent=agent, world=world, delta=delta, movement_kind="move", message="Agent moved by configured delta.", ) return self.failure( reward=self._failure_reward(), message="MoveBehavior requires target_position, direction, or delta.", ) except Exception as exc: logger.exception("MoveBehavior failed for agent_id=%s", agent.id) return self.failure( reward=self._failure_reward(), message="MoveBehavior raised an exception.", metadata={"error": repr(exc)}, ) def _has_movement_specification(self) -> bool: """ Return whether parameters define a movement instruction. Returns ------- bool True if target, direction, or delta is configured. """ if self._target_position_from_parameters() is not None: return True return any( key in self.parameters and self.parameters[key] is not None for key in ("direction", "delta") ) def _estimated_requested_distance(self, agent: Agent) -> float: """ Estimate movement distance before execution. Parameters ---------- agent: Agent attempting to move. Returns ------- float Estimated requested movement distance. """ target_position = self._target_position_from_parameters() step_size = self._step_size() if target_position is not None: if target_position.shape != agent.position.shape: return step_size return min(float(np.linalg.norm(target_position - agent.position)), step_size) if "delta" in self.parameters and self.parameters["delta"] is not None: delta = self._normalize_vector(self.parameters["delta"], "delta") distance = float(np.linalg.norm(delta)) if self._optional_bool_parameter("limit_delta", default=False): return min(distance, step_size) return distance return step_size def _target_position_from_parameters(self) -> NDArray[np.float64] | None: """ Read an optional configured target position. Returns ------- NDArray[np.float64] | None Target position, if configured. """ for key in ("target_position", "destination", "position"): if key in self.parameters and self.parameters[key] is not None: return self._normalize_vector(self.parameters[key], key) return None @dataclass(slots=True) class WanderBehavior(MovementBehavior): """ Move an agent in a random direction. Wander uses ``world.rng`` when available, making stochastic movement reproducible when the world has a fixed seed. Common parameters ----------------- step_size: Movement distance. cost_state_key: Optional numeric state key charged by movement. cost_per_unit: Optional cost per distance unit. bounds: Optional movement bounds. """ def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute random movement. Parameters ---------- agent: Agent executing the behavior. world: World in which movement occurs. Returns ------- BehaviorResult Wander execution result. """ try: step_size = self._step_size() direction = self._random_unit_vector(world, int(agent.position.size)) delta = direction * step_size return self._attempt_movement( agent=agent, world=world, delta=delta, movement_kind="wander", message="Agent wandered randomly.", extra_metadata={ "direction": direction.tolist(), }, ) except Exception as exc: logger.exception("WanderBehavior failed for agent_id=%s", agent.id) return self.failure( reward=self._failure_reward(), message="WanderBehavior raised an exception.", metadata={"error": repr(exc)}, ) @dataclass(slots=True) class SeekBehavior(MovementBehavior): """ Move an agent toward a target position, agent, or resource. Common parameters ----------------- target_position: Explicit target position. target: ``"resource"`` or ``"agent"``. Defaults to ``"resource"``. selector: Generic selector for target objects. radius: Optional maximum search radius. target_strategy: ``"nearest"``, ``"farthest"``, or ``"random"``. step_size: Maximum movement distance. """ def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute seek movement. Parameters ---------- agent: Agent executing seek. world: World in which seek occurs. Returns ------- BehaviorResult Seek execution result. """ try: resolved_target = self._resolve_target( agent=agent, world=world, default_kind="resource", position_keys=("target_position", "destination", "position"), ) if resolved_target is None: return self.failure( reward=self._failure_reward(), message="SeekBehavior found no matching target.", metadata={ "selector": self._selector(), "target": self.parameters.get("target", "resource"), }, ) target_position, candidate = resolved_target step_size = self._step_size() delta, reached_target = self._delta_toward( current_position=agent.position, target_position=target_position, step_size=step_size, ) metadata: dict[str, Any] = {} if candidate is not None: metadata["target_candidate"] = candidate.to_metadata() return self._attempt_movement( agent=agent, world=world, delta=delta, movement_kind="seek", message=( "Agent moved toward target." if not reached_target else "Agent reached target." ), target_position=target_position, reached_target=reached_target, extra_metadata=metadata, ) except Exception as exc: logger.exception("SeekBehavior failed for agent_id=%s", agent.id) return self.failure( reward=self._failure_reward(), message="SeekBehavior raised an exception.", metadata={"error": repr(exc)}, ) @dataclass(slots=True) class FleeBehavior(MovementBehavior): """ Move an agent away from a threat position, agent, or resource. Common parameters ----------------- threat_position: Explicit threat position. target: ``"agent"`` or ``"resource"``. Defaults to ``"agent"``. selector: Generic selector for threat objects. radius: Optional maximum search radius. target_strategy: ``"nearest"``, ``"farthest"``, or ``"random"``. step_size: Movement distance. success_when_no_target: Whether no threat should count as a successful no-op. Defaults to True. """ def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute flee movement. Parameters ---------- agent: Agent executing flee. world: World in which flee occurs. Returns ------- BehaviorResult Flee execution result. """ try: resolved_threat = self._resolve_target( agent=agent, world=world, default_kind="agent", position_keys=("threat_position", "source_position", "target_position"), ) if resolved_threat is None: success_when_no_target = self._optional_bool_parameter( "success_when_no_target", default=True, ) if success_when_no_target: return self.success( reward=self._success_reward(), message="FleeBehavior found no threat; agent stayed in place.", metadata={ "movement_kind": "flee", "no_target": True, "selector": self._selector(), }, ) return self.failure( reward=self._failure_reward(), message="FleeBehavior found no matching threat.", metadata={ "selector": self._selector(), "target": self.parameters.get("target", "agent"), }, ) threat_position, candidate = resolved_threat step_size = self._step_size() delta = self._delta_away( current_position=agent.position, threat_position=threat_position, step_size=step_size, world=world, ) metadata: dict[str, Any] = {} if candidate is not None: metadata["threat_candidate"] = candidate.to_metadata() return self._attempt_movement( agent=agent, world=world, delta=delta, movement_kind="flee", message="Agent moved away from threat.", target_position=threat_position, reached_target=False, extra_metadata=metadata, ) except Exception as exc: logger.exception("FleeBehavior failed for agent_id=%s", agent.id) return self.failure( reward=self._failure_reward(), message="FleeBehavior raised an exception.", metadata={"error": repr(exc)}, ) Move = MoveBehavior Wander = WanderBehavior Seek = SeekBehavior Flee = FleeBehavior __all__ = [ "DEFAULT_STEP_SIZE", "Flee", "FleeBehavior", "Move", "MoveBehavior", "MovementBehavior", "MovementComputation", "PositionInput", "Seek", "SeekBehavior", "TargetCandidate", "TargetKind", "TargetStrategy", "Wander", "WanderBehavior", ]