""" core.world ========== Central simulation container for WorldSmithAI. This module defines the domain-agnostic World runtime object. A World owns agents, resources, events, metrics, metadata, simulation history, and an optional scheduler. It coordinates simulation steps without hardcoding any specific ecosystem, economy, civilization, or fantasy setting. The World does not know what "wolf", "scientist", "dragon", "mana", "food", "war", or "breakthrough" means. Those meanings are supplied by the DSL, behaviors, policies, events, and future factory modules. Minimal usage example --------------------- from core.agent import Agent from core.resource import Resource from core.event import Event from core.world import World world = World() world.add_agent( Agent( id="agent_1", type="researcher", position=[0.0, 0.0], state={"energy": 1.0}, ) ) world.add_resource( Resource( id="knowledge_pool", type="knowledge", amount=10.0, position=[1.0, 1.0], metadata={"regeneration_rate": 0.1}, ) ) world.add_event( Event( name="funding_boom", trigger_step=0, payload={ "effects": [ { "target": "resources", "operation": "regenerate", "selector": {"type": "knowledge"}, "parameters": {"amount": 5.0}, } ] }, ) ) result = world.step() print(result.to_dict()) """ from __future__ import annotations import logging from collections import Counter from collections.abc import Mapping, Sequence as AbcSequence from copy import deepcopy from dataclasses import dataclass, field from numbers import Integral, Real from typing import Any, Protocol, TypeAlias, cast import numpy as np from numpy.typing import NDArray from core.agent import Agent, AgentStepResult from core.event import Event, EventEffect, EventEffectResult, EventExecutionResult from core.resource import Resource, ResourceOperationResult logger = logging.getLogger(__name__) PositionInput: TypeAlias = AbcSequence[float] | NDArray[np.float64] Selector: TypeAlias = Mapping[str, Any] @dataclass(frozen=True, slots=True) class WorldStepResult: """ Structured result of one world simulation step. Attributes ---------- started_step: Step count at the beginning of the world step. ended_step: Step count after the world step completes. success: Whether the world step completed successfully. agent_results: Per-agent step results. event_results: Per-event execution results. resource_results: Per-resource update results. metrics: Metrics collected during this step. message: Human-readable step summary. metadata: Additional structured step metadata. errors: Step-level error messages, if any. """ started_step: int ended_step: int success: bool agent_results: tuple[AgentStepResult, ...] = field(default_factory=tuple) event_results: tuple[EventExecutionResult, ...] = field(default_factory=tuple) resource_results: tuple[ResourceOperationResult, ...] = field(default_factory=tuple) metrics: Mapping[str, Any] = field(default_factory=dict) message: str = "" metadata: Mapping[str, Any] = field(default_factory=dict) errors: tuple[str, ...] = field(default_factory=tuple) def __post_init__(self) -> None: """ Validate and normalize step result fields. Raises ------ TypeError If fields have invalid runtime types. ValueError If step counts are invalid. """ if isinstance(self.started_step, bool) or not isinstance( self.started_step, Integral, ): raise TypeError("WorldStepResult.started_step must be an integer.") if isinstance(self.ended_step, bool) or not isinstance( self.ended_step, Integral, ): raise TypeError("WorldStepResult.ended_step must be an integer.") if self.started_step < 0: raise ValueError("WorldStepResult.started_step cannot be negative.") if self.ended_step < self.started_step: raise ValueError("WorldStepResult.ended_step cannot precede started_step.") if not isinstance(self.success, bool): raise TypeError("WorldStepResult.success must be a boolean.") object.__setattr__(self, "started_step", int(self.started_step)) object.__setattr__(self, "ended_step", int(self.ended_step)) if not isinstance(self.agent_results, tuple): object.__setattr__(self, "agent_results", tuple(self.agent_results)) if not isinstance(self.event_results, tuple): object.__setattr__(self, "event_results", tuple(self.event_results)) if not isinstance(self.resource_results, tuple): object.__setattr__(self, "resource_results", tuple(self.resource_results)) for result in self.agent_results: if not isinstance(result, AgentStepResult): raise TypeError( "WorldStepResult.agent_results must contain AgentStepResult " "objects." ) for result in self.event_results: if not isinstance(result, EventExecutionResult): raise TypeError( "WorldStepResult.event_results must contain " "EventExecutionResult objects." ) for result in self.resource_results: if not isinstance(result, ResourceOperationResult): raise TypeError( "WorldStepResult.resource_results must contain " "ResourceOperationResult objects." ) if not isinstance(self.metrics, Mapping): raise TypeError("WorldStepResult.metrics must be a mapping.") if not isinstance(self.metadata, Mapping): raise TypeError("WorldStepResult.metadata must be a mapping.") if not isinstance(self.errors, tuple): object.__setattr__(self, "errors", tuple(self.errors)) for error in self.errors: if not isinstance(error, str): raise TypeError("WorldStepResult.errors must contain strings.") object.__setattr__(self, "metrics", deepcopy(dict(self.metrics))) object.__setattr__(self, "metadata", deepcopy(dict(self.metadata))) def to_dict(self) -> dict[str, Any]: """ Convert the step result into a JSON-friendly dictionary. Returns ------- dict[str, Any] Serializable world step result. """ return { "started_step": self.started_step, "ended_step": self.ended_step, "success": self.success, "agent_results": [result.to_dict() for result in self.agent_results], "event_results": [result.to_dict() for result in self.event_results], "resource_results": [ result.to_dict() for result in self.resource_results ], "metrics": deepcopy(dict(self.metrics)), "message": self.message, "metadata": deepcopy(dict(self.metadata)), "errors": list(self.errors), } class SchedulerProtocol(Protocol): """ Structural protocol for future scheduler implementations. The concrete scheduler will be implemented in ``core/scheduler.py``. This protocol lets World accept an injected scheduler without importing a concrete scheduler class. """ def step(self, world: World) -> WorldStepResult | Mapping[str, Any] | None: """ Execute one simulation step. Schedulers should execute events, agents, and resources for the current world step. The World remains responsible for incrementing ``world.step_count``. Parameters ---------- world: World instance to advance. Returns ------- WorldStepResult | Mapping[str, Any] | None Structured step result, compatible mapping, or None. """ ... @dataclass(slots=True) class World: """ Central domain-agnostic simulation container. The World owns all runtime objects and provides generic coordination APIs used by agents, events, schedulers, metrics, visualization, and the future Gradio app. Attributes ---------- agents: Mapping from agent id to Agent. resources: Mapping from resource id to Resource. events: Mapping from event id to Event. metrics: Mutable metrics storage. Built-in step metrics are appended under the ``"history"`` key. step_count: Current simulation step. metadata: Generic world-level metadata. scheduler: Optional scheduler object implementing SchedulerProtocol. seed: Optional random seed. Defaults to 0 for deterministic behavior. history: Step history as JSON-friendly dictionaries. event_history: Event execution history as JSON-friendly dictionaries. history_limit: Optional maximum number of step history entries. """ agents: dict[str, Agent] = field(default_factory=dict) resources: dict[str, Resource] = field(default_factory=dict) events: dict[str, Event] = field(default_factory=dict) metrics: dict[str, Any] = field(default_factory=dict) step_count: int = 0 metadata: dict[str, Any] = field(default_factory=dict) scheduler: SchedulerProtocol | None = None seed: int | None = 0 history: list[dict[str, Any]] = field(default_factory=list) event_history: list[dict[str, Any]] = field(default_factory=list) history_limit: int | None = None rng: np.random.Generator = field(init=False, repr=False) def __post_init__(self) -> None: """ Validate and normalize world fields. Raises ------ TypeError If fields have invalid runtime types. ValueError If identifiers, step count, seed, or spatial dimensions are invalid. """ self.step_count = self._normalize_step(self.step_count, "step_count") if not isinstance(self.agents, Mapping): raise TypeError("World.agents must be a mapping.") if not isinstance(self.resources, Mapping): raise TypeError("World.resources must be a mapping.") if not isinstance(self.events, Mapping): raise TypeError("World.events must be a mapping.") if not isinstance(self.metrics, Mapping): raise TypeError("World.metrics must be a mapping.") if not isinstance(self.metadata, Mapping): raise TypeError("World.metadata must be a mapping.") if not isinstance(self.history, list): raise TypeError("World.history must be a list.") if not isinstance(self.event_history, list): raise TypeError("World.event_history must be a list.") self.agents = dict(self.agents) self.resources = dict(self.resources) self.events = dict(self.events) self.metrics = dict(self.metrics) self.metadata = dict(self.metadata) self._validate_string_keys(self.metrics, "metrics") self._validate_string_keys(self.metadata, "metadata") for key, agent in self.agents.items(): if not isinstance(key, str): raise TypeError("World.agents keys must be strings.") if not isinstance(agent, Agent): raise TypeError("World.agents values must be Agent objects.") if key != agent.id: raise ValueError( f"World.agents key '{key}' does not match Agent.id " f"'{agent.id}'." ) for key, resource in self.resources.items(): if not isinstance(key, str): raise TypeError("World.resources keys must be strings.") if not isinstance(resource, Resource): raise TypeError("World.resources values must be Resource objects.") if key != resource.id: raise ValueError( f"World.resources key '{key}' does not match Resource.id " f"'{resource.id}'." ) for key, event in self.events.items(): if not isinstance(key, str): raise TypeError("World.events keys must be strings.") if not isinstance(event, Event): raise TypeError("World.events values must be Event objects.") if event.id is None: raise ValueError("World.events values must have non-null ids.") if key != event.id: raise ValueError( f"World.events key '{key}' does not match Event.id " f"'{event.id}'." ) if self.scheduler is not None and not callable( getattr(self.scheduler, "step", None) ): raise TypeError("World.scheduler must implement a callable step(world).") if self.history_limit is not None: self.history_limit = self._normalize_positive_int( self.history_limit, "history_limit", ) normalized_seed = self._normalize_seed(self.seed) self.seed = normalized_seed self.rng = np.random.default_rng(normalized_seed) self._validate_spatial_dimensions() @property def dimension(self) -> int | None: """ Return the spatial dimensionality of this world. Returns ------- int | None Number of position dimensions, or None when the world has no positioned agents/resources. """ dimensions: set[int] = set() for agent in self.agents.values(): dimensions.add(int(agent.position.size)) for resource in self.resources.values(): dimensions.add(int(resource.position.size)) if not dimensions: return None if len(dimensions) > 1: raise ValueError( f"World contains inconsistent spatial dimensions: {dimensions}." ) return next(iter(dimensions)) def add_agent(self, agent: Agent, *, replace: bool = False) -> None: """ Add an agent to the world. Parameters ---------- agent: Agent to add. replace: Whether to replace an existing agent with the same id. Raises ------ TypeError If agent or replace is invalid. ValueError If the id already exists or the position dimensionality is invalid. """ if not isinstance(agent, Agent): raise TypeError("agent must be an Agent.") if not isinstance(replace, bool): raise TypeError("replace must be a boolean.") if agent.id in self.agents and not replace: raise ValueError(f"Agent with id '{agent.id}' already exists.") self._validate_position_dimension(agent.position) self.agents[agent.id] = agent def remove_agent(self, agent_id: str, *, missing_ok: bool = False) -> Agent | None: """ Remove an agent from the world. Parameters ---------- agent_id: Id of the agent to remove. missing_ok: Whether missing ids should be ignored. Returns ------- Agent | None Removed agent, or None when missing_ok is True and the id is absent. Raises ------ KeyError If the agent id is missing and missing_ok is False. """ normalized_id = self._normalize_identifier(agent_id, "agent_id") if normalized_id not in self.agents: if missing_ok: return None raise KeyError(f"Unknown agent id: {normalized_id}") return self.agents.pop(normalized_id) def add_resource(self, resource: Resource, *, replace: bool = False) -> None: """ Add a resource to the world. Parameters ---------- resource: Resource to add. replace: Whether to replace an existing resource with the same id. Raises ------ TypeError If resource or replace is invalid. ValueError If the id already exists or the position dimensionality is invalid. """ if not isinstance(resource, Resource): raise TypeError("resource must be a Resource.") if not isinstance(replace, bool): raise TypeError("replace must be a boolean.") if resource.id in self.resources and not replace: raise ValueError(f"Resource with id '{resource.id}' already exists.") self._validate_position_dimension(resource.position) self.resources[resource.id] = resource def remove_resource( self, resource_id: str, *, missing_ok: bool = False, ) -> Resource | None: """ Remove a resource from the world. Parameters ---------- resource_id: Id of the resource to remove. missing_ok: Whether missing ids should be ignored. Returns ------- Resource | None Removed resource, or None when missing_ok is True and the id is absent. Raises ------ KeyError If the resource id is missing and missing_ok is False. """ normalized_id = self._normalize_identifier(resource_id, "resource_id") if normalized_id not in self.resources: if missing_ok: return None raise KeyError(f"Unknown resource id: {normalized_id}") return self.resources.pop(normalized_id) def add_event(self, event: Event, *, replace: bool = False) -> None: """ Add an event to the world. Parameters ---------- event: Event to add. replace: Whether to replace an existing event with the same id. Raises ------ TypeError If event or replace is invalid. ValueError If the event id already exists. """ if not isinstance(event, Event): raise TypeError("event must be an Event.") if not isinstance(replace, bool): raise TypeError("replace must be a boolean.") if event.id is None: raise ValueError("event.id cannot be None.") if event.id in self.events and not replace: raise ValueError(f"Event with id '{event.id}' already exists.") self.events[event.id] = event def remove_event(self, event_id: str, *, missing_ok: bool = False) -> Event | None: """ Remove an event from the world. Parameters ---------- event_id: Id of the event to remove. missing_ok: Whether missing ids should be ignored. Returns ------- Event | None Removed event, or None when missing_ok is True and the id is absent. Raises ------ KeyError If the event id is missing and missing_ok is False. """ normalized_id = self._normalize_identifier(event_id, "event_id") if normalized_id not in self.events: if missing_ok: return None raise KeyError(f"Unknown event id: {normalized_id}") return self.events.pop(normalized_id) def step(self) -> WorldStepResult: """ Advance the world by one simulation step. If a scheduler is attached, the scheduler executes the step. Otherwise, World uses its built-in deterministic default sequence: 1. Process due events. 2. Execute agents in insertion order. 3. Update regenerating resources. 4. Collect metrics. Returns ------- WorldStepResult Structured result of the world step. """ started_step = self.step_count try: if self.scheduler is not None: result = self._step_with_scheduler(started_step) else: result = self._default_step(started_step) self.record_step_result(result) self.step_count = started_step + 1 return result except Exception as exc: logger.exception("World step failed at step_count=%s", started_step) result = WorldStepResult( started_step=started_step, ended_step=started_step, success=False, message="World step raised an exception.", errors=(repr(exc),), metadata={"failed": True}, ) self.record_step_result(result) return result def run(self, steps: int) -> tuple[WorldStepResult, ...]: """ Run the world for multiple simulation steps. Parameters ---------- steps: Number of steps to execute. Returns ------- tuple[WorldStepResult, ...] Step results in execution order. Raises ------ TypeError If steps is not an integer. ValueError If steps is negative. """ normalized_steps = self._normalize_step(steps, "steps") results: list[WorldStepResult] = [] for _ in range(normalized_steps): results.append(self.step()) return tuple(results) def process_events(self) -> tuple[EventExecutionResult, ...]: """ Execute all events due at the current world step. Returns ------- tuple[EventExecutionResult, ...] Event execution results. """ results: list[EventExecutionResult] = [] due_events = sorted( ( event for event in self.events.values() if event.is_due(self.step_count) ), key=lambda event: (event.trigger_step, event.id or event.name), ) for event in due_events: try: results.append(event.execute(self)) except Exception as exc: logger.exception( "Event execution failed: event_id=%s event_name=%s", event.id, event.name, ) results.append( EventExecutionResult( event_id=event.id or event.name, event_name=event.name, trigger_step=event.trigger_step, executed_step=self.step_count, success=False, skipped=False, message="Event execution raised an exception.", metadata={"error": repr(exc)}, ) ) return tuple(results) def execute_agents(self) -> tuple[AgentStepResult, ...]: """ Execute one step for each alive agent. Returns ------- tuple[AgentStepResult, ...] Per-agent step results. """ results: list[AgentStepResult] = [] for agent in list(self.agents.values()): if not agent.alive: continue try: results.append(agent.step(self)) except Exception as exc: logger.exception("Agent step failed: agent_id=%s", agent.id) results.append( AgentStepResult( agent_id=agent.id, agent_type=agent.type, step_count=self.step_count, behavior_id=None, success=False, reward=0.0, message="Agent step raised an exception.", metadata={"error": repr(exc)}, ) ) return tuple(results) def update_resources(self) -> tuple[ResourceOperationResult, ...]: """ Apply default resource regeneration updates. Resources regenerate only when their metadata includes a positive ``regeneration_rate``. Returns ------- tuple[ResourceOperationResult, ...] Per-resource operation results. """ results: list[ResourceOperationResult] = [] for resource in self.resources.values(): try: if resource.regeneration_rate <= 0.0: continue results.append( resource.regenerate( metadata={"source": "world.update_resources"}, ) ) except Exception: logger.exception( "Resource update failed: resource_id=%s resource_type=%s", resource.id, resource.type, ) return tuple(results) def query( self, *, agent: Agent | None = None, position: PositionInput | None = None, radius: float | None = None, include_self: bool = False, agent_types: str | AbcSequence[str] | None = None, resource_types: str | AbcSequence[str] | None = None, only_alive: bool = True, limit: int | None = None, ) -> dict[str, Any]: """ Return a spatial observation around a position. This method is used by ``Agent.observe(world)``. It returns nearby agents and resources as JSON-friendly snapshots with distances. Parameters ---------- agent: Optional observing agent. position: Query position. If omitted, ``agent.position`` is used. radius: Optional non-negative maximum distance. include_self: Whether to include the observing agent in nearby agents. agent_types: Optional agent type or sequence of types to include. resource_types: Optional resource type or sequence of types to include. only_alive: Whether dead/inactive agents should be excluded. limit: Optional maximum number of agents and resources returned per group. Returns ------- dict[str, Any] Observation containing nearby agents, nearby resources, step count, query position, and world metadata. """ if agent is not None and not isinstance(agent, Agent): raise TypeError("agent must be an Agent when provided.") query_position = self._resolve_query_position(agent=agent, position=position) normalized_radius = self._normalize_optional_non_negative_float( radius, "radius", ) if not isinstance(include_self, bool): raise TypeError("include_self must be a boolean.") if not isinstance(only_alive, bool): raise TypeError("only_alive must be a boolean.") normalized_limit = ( None if limit is None else self._normalize_positive_int(limit, "limit") ) allowed_agent_types = self._normalize_optional_string_filter(agent_types) allowed_resource_types = self._normalize_optional_string_filter(resource_types) nearby_agents: list[dict[str, Any]] = [] nearby_resources: list[dict[str, Any]] = [] for candidate in self.agents.values(): if agent is not None and candidate.id == agent.id and not include_self: continue if only_alive and not candidate.alive: continue if allowed_agent_types is not None and candidate.type not in allowed_agent_types: continue distance = self._distance_or_none(candidate.position, query_position) if distance is None: continue if normalized_radius is not None and distance > normalized_radius: continue snapshot = candidate.snapshot() snapshot["distance"] = distance nearby_agents.append(snapshot) for resource in self.resources.values(): if allowed_resource_types is not None and resource.type not in allowed_resource_types: continue distance = self._distance_or_none(resource.position, query_position) if distance is None: continue if normalized_radius is not None and distance > normalized_radius: continue snapshot = resource.snapshot() snapshot["distance"] = distance nearby_resources.append(snapshot) nearby_agents.sort(key=lambda item: (item["distance"], item["id"])) nearby_resources.sort(key=lambda item: (item["distance"], item["id"])) if normalized_limit is not None: nearby_agents = nearby_agents[:normalized_limit] nearby_resources = nearby_resources[:normalized_limit] return { "step_count": self.step_count, "position": query_position.tolist(), "radius": normalized_radius, "nearby_agents": nearby_agents, "nearby_resources": nearby_resources, "world_metadata": deepcopy(self.metadata), } def select_agents(self, selector: Selector | None = None) -> list[Agent]: """ Select agents using a generic selector. Supported selector keys include: ``id``, ``ids``, ``type``, ``types``, ``alive``, ``state``, ``state_min``, ``state_max``, ``near``, ``position``, and ``radius``. Parameters ---------- selector: Generic selector mapping. Returns ------- list[Agent] Matching agents. """ normalized_selector = self._normalize_selector(selector) return [ agent for agent in self.agents.values() if self._matches_agent_selector(agent, normalized_selector) ] def select_resources(self, selector: Selector | None = None) -> list[Resource]: """ Select resources using a generic selector. Supported selector keys include: ``id``, ``ids``, ``type``, ``types``, ``metadata``, ``min_amount``, ``max_amount``, ``amount_min``, ``amount_max``, ``near``, ``position``, and ``radius``. Parameters ---------- selector: Generic selector mapping. Returns ------- list[Resource] Matching resources. """ normalized_selector = self._normalize_selector(selector) return [ resource for resource in self.resources.values() if self._matches_resource_selector(resource, normalized_selector) ] def select_events(self, selector: Selector | None = None) -> list[Event]: """ Select events using a generic selector. Supported selector keys include: ``id``, ``ids``, ``name``, ``names``, ``trigger_step``, and ``enabled``. Parameters ---------- selector: Generic selector mapping. Returns ------- list[Event] Matching events. """ normalized_selector = self._normalize_selector(selector) return [ event for event in self.events.values() if self._matches_event_selector(event, normalized_selector) ] def apply_event_effect( self, *, event: Event, effect: EventEffect, ) -> EventEffectResult: """ Apply one structured event effect. This method is the safe bridge between the generic event system and world mutation. It supports generic operations for agents, resources, events, metrics, and world metadata. It never executes arbitrary Python code. Parameters ---------- event: Event that produced the effect. effect: Structured event effect. Returns ------- EventEffectResult Structured effect application result. """ if not isinstance(event, Event): raise TypeError("event must be an Event.") if not isinstance(effect, EventEffect): raise TypeError("effect must be an EventEffect.") target = effect.target.strip().lower() if target in {"agent", "agents"}: return self._apply_agent_effect(event=event, effect=effect) if target in {"resource", "resources"}: return self._apply_resource_effect(event=event, effect=effect) if target in {"event", "events"}: return self._apply_event_collection_effect(event=event, effect=effect) if target in {"world", "metadata", "world_metadata"}: return self._apply_mapping_effect( effect=effect, target_mapping=self.metadata, mapping_label="metadata", ) if target in {"metric", "metrics"}: return self._apply_mapping_effect( effect=effect, target_mapping=self.metrics, mapping_label="metrics", ) return EventEffectResult( target=effect.target, operation=effect.operation, success=False, affected_count=0, message=f"Unsupported event effect target: {effect.target}", metadata={"event_id": event.id, "event_name": event.name}, ) def record_event(self, *, event: Event, result: EventExecutionResult) -> None: """ Record an event execution result. Parameters ---------- event: Event that executed. result: Structured event execution result. """ if not isinstance(event, Event): raise TypeError("event must be an Event.") if not isinstance(result, EventExecutionResult): raise TypeError("result must be an EventExecutionResult.") self.event_history.append(result.to_dict()) def collect_metrics(self) -> dict[str, Any]: """ Collect generic built-in world metrics. These metrics are intentionally domain-agnostic. Specialized metrics modules will later compute diversity, entropy, stability, and interestingness. Returns ------- dict[str, Any] JSON-friendly metrics snapshot. """ agents_by_type = Counter(agent.type for agent in self.agents.values()) alive_agents_by_type = Counter( agent.type for agent in self.agents.values() if agent.alive ) resources_by_type = Counter( resource.type for resource in self.resources.values() ) resource_amounts_by_type: dict[str, float] = {} for resource in self.resources.values(): resource_amounts_by_type[resource.type] = ( resource_amounts_by_type.get(resource.type, 0.0) + resource.amount ) enabled_events = sum(1 for event in self.events.values() if event.enabled) executed_events = sum(event.execution_count for event in self.events.values()) return { "step_count": self.step_count, "agents": { "total": len(self.agents), "alive": sum(1 for agent in self.agents.values() if agent.alive), "by_type": dict(agents_by_type), "alive_by_type": dict(alive_agents_by_type), }, "resources": { "total": len(self.resources), "by_type": dict(resources_by_type), "amount_by_type": resource_amounts_by_type, }, "events": { "total": len(self.events), "enabled": enabled_events, "executions": executed_events, }, } def record_step_result(self, result: WorldStepResult) -> None: """ Record a world step result in history and metrics storage. Parameters ---------- result: Step result to record. """ if not isinstance(result, WorldStepResult): raise TypeError("result must be a WorldStepResult.") self.history.append(result.to_dict()) if self.history_limit is not None: overflow = len(self.history) - self.history_limit if overflow > 0: del self.history[:overflow] metric_history = self.metrics.setdefault("history", []) if isinstance(metric_history, list): metric_history.append(deepcopy(dict(result.metrics))) else: logger.warning( "World.metrics['history'] exists but is not a list; " "step metrics were not appended." ) def snapshot(self, *, include_history: bool = False) -> dict[str, Any]: """ Return a JSON-friendly snapshot of the world. Parameters ---------- include_history: Whether to include step and event history. Returns ------- dict[str, Any] Serializable world snapshot. """ if not isinstance(include_history, bool): raise TypeError("include_history must be a boolean.") snapshot = { "step_count": self.step_count, "dimension": self.dimension, "metadata": deepcopy(self.metadata), "agents": { agent_id: agent.snapshot() for agent_id, agent in self.agents.items() }, "resources": { resource_id: resource.snapshot() for resource_id, resource in self.resources.items() }, "events": { event_id: event.snapshot() for event_id, event in self.events.items() }, "metrics": deepcopy(self.metrics), } if include_history: snapshot["history"] = deepcopy(self.history) snapshot["event_history"] = deepcopy(self.event_history) return snapshot def _default_step(self, started_step: int) -> WorldStepResult: """ Execute the built-in deterministic step sequence. Parameters ---------- started_step: Step count at the beginning of the step. Returns ------- WorldStepResult Structured step result. """ event_results = self.process_events() agent_results = self.execute_agents() resource_results = self.update_resources() metrics = self.collect_metrics() success = ( all(result.success for result in event_results) and all(result.success for result in agent_results) and all(result.success for result in resource_results) ) return WorldStepResult( started_step=started_step, ended_step=started_step + 1, success=success, agent_results=agent_results, event_results=event_results, resource_results=resource_results, metrics=metrics, message=( "World step completed successfully." if success else "World step completed with one or more failed operations." ), metadata={ "scheduler": None, "agent_count": len(agent_results), "event_count": len(event_results), "resource_update_count": len(resource_results), }, ) def _step_with_scheduler(self, started_step: int) -> WorldStepResult: """ Execute one step through the injected scheduler. Parameters ---------- started_step: Step count at the beginning of the step. Returns ------- WorldStepResult Structured step result. """ if self.scheduler is None: raise RuntimeError("Cannot step with scheduler because scheduler is None.") raw_result = self.scheduler.step(self) return self._normalize_scheduler_result( raw_result=raw_result, started_step=started_step, ) def _normalize_scheduler_result( self, *, raw_result: WorldStepResult | Mapping[str, Any] | None, started_step: int, ) -> WorldStepResult: """ Normalize a scheduler return value into WorldStepResult. Parameters ---------- raw_result: Value returned by the scheduler. started_step: Step count at the beginning of the step. Returns ------- WorldStepResult Normalized scheduler result. """ if raw_result is None: return WorldStepResult( started_step=started_step, ended_step=started_step + 1, success=True, metrics=self.collect_metrics(), message="Scheduler completed without returning a result.", metadata={"scheduler": self.scheduler.__class__.__name__}, ) if isinstance(raw_result, WorldStepResult): return raw_result if isinstance(raw_result, Mapping): return WorldStepResult( started_step=int(raw_result.get("started_step", started_step)), ended_step=int(raw_result.get("ended_step", started_step + 1)), success=bool(raw_result.get("success", True)), agent_results=tuple(raw_result.get("agent_results", ())), event_results=tuple(raw_result.get("event_results", ())), resource_results=tuple(raw_result.get("resource_results", ())), metrics=dict(raw_result.get("metrics", self.collect_metrics())), message=str(raw_result.get("message", "")), metadata=dict(raw_result.get("metadata", {})), errors=tuple(raw_result.get("errors", ())), ) raise TypeError( "Scheduler.step(world) must return WorldStepResult, mapping, or None." ) def _apply_agent_effect( self, *, event: Event, effect: EventEffect, ) -> EventEffectResult: """ Apply an event effect targeting agents. Parameters ---------- event: Source event. effect: Agent-targeting effect. Returns ------- EventEffectResult Structured effect result. """ operation = self._normalize_operation(effect.operation) parameters = dict(effect.parameters) if operation in {"spawn", "spawn_agent", "create", "create_agent"}: agent = self._agent_from_effect_parameters(parameters) self.add_agent(agent) return self._effect_result( effect=effect, success=True, affected_count=1, message="Agent spawned.", metadata={"event_id": event.id, "created_agent_id": agent.id}, ) selected_agents = self.select_agents(effect.selector) if operation in {"set_state", "update_state", "set_agent_state"}: updates = self._state_updates_from_parameters(parameters) for agent in selected_agents: agent.update_state(deepcopy(updates)) elif operation in {"increment_state", "add_state"}: key = self._require_string_parameter(parameters, "key") amount = self._require_finite_float_parameter(parameters, "amount") for agent in selected_agents: current = self._numeric_state_value(agent, key, default=0.0) agent.update_state({key: current + amount}) elif operation in {"multiply_state", "scale_state"}: key = self._require_string_parameter(parameters, "key") factor = self._require_finite_float_parameter(parameters, "factor") for agent in selected_agents: current = self._numeric_state_value(agent, key, default=0.0) agent.update_state({key: current * factor}) elif operation in {"mark_dead", "kill", "deactivate"}: reason = parameters.get("reason") for agent in selected_agents: agent.mark_dead(reason=None if reason is None else str(reason)) elif operation in {"revive", "activate"}: reason = parameters.get("reason") for agent in selected_agents: agent.revive(reason=None if reason is None else str(reason)) elif operation in {"remove", "remove_agent", "delete"}: for agent in selected_agents: self.remove_agent(agent.id, missing_ok=True) elif operation in {"set_position", "move_to"}: position = self._require_position_parameter(parameters, "position") self._validate_position_dimension(position) for agent in selected_agents: agent.set_position(position) else: return self._effect_result( effect=effect, success=False, affected_count=0, message=f"Unsupported agent operation: {effect.operation}", metadata={"event_id": event.id}, ) return self._effect_result( effect=effect, success=True, affected_count=len(selected_agents), message=f"Agent operation '{operation}' applied.", metadata={ "event_id": event.id, "selected_agent_ids": [agent.id for agent in selected_agents], }, ) def _apply_resource_effect( self, *, event: Event, effect: EventEffect, ) -> EventEffectResult: """ Apply an event effect targeting resources. Parameters ---------- event: Source event. effect: Resource-targeting effect. Returns ------- EventEffectResult Structured effect result. """ operation = self._normalize_operation(effect.operation) parameters = dict(effect.parameters) if operation in {"spawn", "spawn_resource", "create", "create_resource"}: resource = self._resource_from_effect_parameters(parameters) self.add_resource(resource) return self._effect_result( effect=effect, success=True, affected_count=1, message="Resource spawned.", metadata={"event_id": event.id, "created_resource_id": resource.id}, ) selected_resources = self.select_resources(effect.selector) operation_results: list[dict[str, Any]] = [] if operation in {"set_amount", "set_resource_amount"}: amount = self._require_non_negative_float_parameter(parameters, "amount") for resource in selected_resources: operation_results.append( resource.set_amount( amount, metadata={"event_id": event.id, "operation": operation}, ).to_dict() ) elif operation in {"increment_amount", "add_amount"}: amount = self._require_finite_float_parameter(parameters, "amount") for resource in selected_resources: if amount >= 0.0: result = resource.regenerate( amount, metadata={"event_id": event.id, "operation": operation}, ) else: result = resource.deplete( abs(amount), allow_partial=True, metadata={"event_id": event.id, "operation": operation}, ) operation_results.append(result.to_dict()) elif operation in {"multiply_amount", "scale_amount"}: factor = self._require_non_negative_float_parameter(parameters, "factor") for resource in selected_resources: operation_results.append( resource.set_amount( resource.amount * factor, metadata={"event_id": event.id, "operation": operation}, ).to_dict() ) elif operation in {"deplete", "deplete_amount"}: amount = self._require_non_negative_float_parameter(parameters, "amount") allow_partial = self._optional_bool_parameter( parameters, "allow_partial", default=True, ) for resource in selected_resources: operation_results.append( resource.deplete( amount, allow_partial=allow_partial, metadata={"event_id": event.id, "operation": operation}, ).to_dict() ) elif operation in {"regenerate", "regenerate_amount"}: amount = parameters.get("amount") capacity = parameters.get("capacity") normalized_amount = ( None if amount is None else self._normalize_non_negative_float(amount, "amount") ) normalized_capacity = ( None if capacity is None else self._normalize_non_negative_float(capacity, "capacity") ) for resource in selected_resources: operation_results.append( resource.regenerate( normalized_amount, capacity=normalized_capacity, metadata={"event_id": event.id, "operation": operation}, ).to_dict() ) elif operation in {"set_position", "move_to"}: position = self._require_position_parameter(parameters, "position") self._validate_position_dimension(position) for resource in selected_resources: resource.set_position(position) elif operation in {"remove", "remove_resource", "delete"}: for resource in selected_resources: self.remove_resource(resource.id, missing_ok=True) else: return self._effect_result( effect=effect, success=False, affected_count=0, message=f"Unsupported resource operation: {effect.operation}", metadata={"event_id": event.id}, ) operation_success = all( bool(result.get("success", True)) for result in operation_results ) return self._effect_result( effect=effect, success=operation_success, affected_count=len(selected_resources), message=f"Resource operation '{operation}' applied.", metadata={ "event_id": event.id, "selected_resource_ids": [ resource.id for resource in selected_resources ], "operation_results": operation_results, }, ) def _apply_event_collection_effect( self, *, event: Event, effect: EventEffect, ) -> EventEffectResult: """ Apply an event effect targeting the world's event collection. Parameters ---------- event: Source event. effect: Event-targeting effect. Returns ------- EventEffectResult Structured effect result. """ operation = self._normalize_operation(effect.operation) parameters = dict(effect.parameters) if operation in {"spawn", "spawn_event", "create", "create_event"}: created_event = self._event_from_effect_parameters(parameters) self.add_event(created_event) return self._effect_result( effect=effect, success=True, affected_count=1, message="Event spawned.", metadata={"event_id": event.id, "created_event_id": created_event.id}, ) selected_events = self.select_events(effect.selector) if operation in {"enable", "enable_event"}: for selected_event in selected_events: selected_event.enable() elif operation in {"disable", "disable_event"}: for selected_event in selected_events: selected_event.disable() elif operation in {"reset", "reset_event"}: for selected_event in selected_events: selected_event.reset() elif operation in {"remove", "remove_event", "delete"}: for selected_event in selected_events: if selected_event.id is not None: self.remove_event(selected_event.id, missing_ok=True) else: return self._effect_result( effect=effect, success=False, affected_count=0, message=f"Unsupported event collection operation: {effect.operation}", metadata={"event_id": event.id}, ) return self._effect_result( effect=effect, success=True, affected_count=len(selected_events), message=f"Event collection operation '{operation}' applied.", metadata={ "event_id": event.id, "selected_event_ids": [ selected_event.id for selected_event in selected_events ], }, ) def _apply_mapping_effect( self, *, effect: EventEffect, target_mapping: dict[str, Any], mapping_label: str, ) -> EventEffectResult: """ Apply a generic mapping mutation effect. Parameters ---------- effect: Mapping-targeting effect. target_mapping: Mutable mapping to mutate. mapping_label: Human-readable mapping label. Returns ------- EventEffectResult Structured effect result. """ operation = self._normalize_operation(effect.operation) parameters = dict(effect.parameters) if operation in { "set", "set_metadata", "set_metric", "update", "update_metadata", "merge", "merge_metadata", }: updates = self._mapping_updates_from_parameters(parameters) target_mapping.update(deepcopy(updates)) elif operation in {"increment", "increment_metadata", "increment_metric"}: key = self._require_string_parameter(parameters, "key") amount = self._require_finite_float_parameter(parameters, "amount") current = target_mapping.get(key, 0.0) if isinstance(current, bool) or not isinstance(current, Real): return self._effect_result( effect=effect, success=False, affected_count=0, message=f"Cannot increment non-numeric {mapping_label} key '{key}'.", ) target_mapping[key] = float(current) + amount elif operation in {"multiply", "multiply_metadata", "multiply_metric"}: key = self._require_string_parameter(parameters, "key") factor = self._require_finite_float_parameter(parameters, "factor") current = target_mapping.get(key, 0.0) if isinstance(current, bool) or not isinstance(current, Real): return self._effect_result( effect=effect, success=False, affected_count=0, message=f"Cannot multiply non-numeric {mapping_label} key '{key}'.", ) target_mapping[key] = float(current) * factor elif operation in {"append", "append_metadata", "append_metric"}: key = self._require_string_parameter(parameters, "key") value = parameters.get("value") if key not in target_mapping or target_mapping[key] is None: target_mapping[key] = [] if not isinstance(target_mapping[key], list): return self._effect_result( effect=effect, success=False, affected_count=0, message=f"Cannot append to non-list {mapping_label} key '{key}'.", ) target_mapping[key].append(deepcopy(value)) else: return self._effect_result( effect=effect, success=False, affected_count=0, message=f"Unsupported {mapping_label} operation: {effect.operation}", ) return self._effect_result( effect=effect, success=True, affected_count=1, message=f"{mapping_label.capitalize()} operation '{operation}' applied.", metadata={"mapping_label": mapping_label}, ) def _agent_from_effect_parameters(self, parameters: Mapping[str, Any]) -> Agent: """ Create an Agent from event effect parameters. Parameters ---------- parameters: Effect parameters containing agent fields. Returns ------- Agent Newly constructed agent. """ agent_id = self._require_string_parameter(parameters, "id") agent_type = self._require_string_parameter(parameters, "type") position = self._require_position_parameter(parameters, "position") state = parameters.get("state", {}) memory = parameters.get("memory", []) goals = parameters.get("goals", []) alive = self._optional_bool_parameter(parameters, "alive", default=True) if not isinstance(state, Mapping): raise TypeError("Spawned agent state must be a mapping.") if not isinstance(memory, list): raise TypeError("Spawned agent memory must be a list.") if not isinstance(goals, list): raise TypeError("Spawned agent goals must be a list.") memory_limit = parameters.get("memory_limit") normalized_memory_limit = ( None if memory_limit is None else self._normalize_positive_int(memory_limit, "memory_limit") ) return Agent( id=agent_id, type=agent_type, position=position, state=deepcopy(dict(state)), memory=deepcopy(memory), goals=[str(goal) for goal in goals], behaviors={}, policy=None, alive=alive, memory_limit=normalized_memory_limit, ) def _resource_from_effect_parameters( self, parameters: Mapping[str, Any], ) -> Resource: """ Create a Resource from event effect parameters. Parameters ---------- parameters: Effect parameters containing resource fields. Returns ------- Resource Newly constructed resource. """ resource_id = self._require_string_parameter(parameters, "id") resource_type = self._require_string_parameter(parameters, "type") amount = self._require_non_negative_float_parameter(parameters, "amount") position = self._require_position_parameter(parameters, "position") metadata = parameters.get("metadata", {}) if not isinstance(metadata, Mapping): raise TypeError("Spawned resource metadata must be a mapping.") return Resource( id=resource_id, type=resource_type, amount=amount, position=position, metadata=deepcopy(dict(metadata)), ) def _event_from_effect_parameters(self, parameters: Mapping[str, Any]) -> Event: """ Create an Event from event effect parameters. Parameters ---------- parameters: Effect parameters containing event fields. Returns ------- Event Newly constructed event. """ name = self._require_string_parameter(parameters, "name") trigger_step = self._normalize_step( self._require_parameter(parameters, "trigger_step"), "trigger_step", ) payload = parameters.get("payload", {}) event_id = parameters.get("id") repeat_interval = parameters.get("repeat_interval") max_executions = parameters.get("max_executions") if not isinstance(payload, Mapping): raise TypeError("Spawned event payload must be a mapping.") return Event( name=name, trigger_step=trigger_step, payload=deepcopy(dict(payload)), id=None if event_id is None else str(event_id), repeat_interval=( None if repeat_interval is None else self._normalize_positive_int( repeat_interval, "repeat_interval", ) ), max_executions=( None if max_executions is None else self._normalize_positive_int(max_executions, "max_executions") ), ) def _matches_agent_selector( self, agent: Agent, selector: Mapping[str, Any], ) -> bool: """ Return whether an agent matches a selector. Parameters ---------- agent: Candidate agent. selector: Normalized selector mapping. Returns ------- bool True when the agent matches. """ if not self._matches_id_and_type(agent.id, agent.type, selector): return False if "alive" in selector and agent.alive is not bool(selector["alive"]): return False if "state" in selector and not self._matches_mapping_filter( agent.state, selector["state"], label="state", ): return False if "state_min" in selector and not self._matches_numeric_thresholds( agent.state, selector["state_min"], comparison="min", ): return False if "state_max" in selector and not self._matches_numeric_thresholds( agent.state, selector["state_max"], comparison="max", ): return False return self._matches_spatial_selector(agent.position, selector) def _matches_resource_selector( self, resource: Resource, selector: Mapping[str, Any], ) -> bool: """ Return whether a resource matches a selector. Parameters ---------- resource: Candidate resource. selector: Normalized selector mapping. Returns ------- bool True when the resource matches. """ if not self._matches_id_and_type(resource.id, resource.type, selector): return False if "metadata" in selector and not self._matches_mapping_filter( resource.metadata, selector["metadata"], label="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 and resource.amount < self._normalize_finite_float( min_amount, "min_amount", ): return False if max_amount is not None and resource.amount > self._normalize_finite_float( max_amount, "max_amount", ): return False return self._matches_spatial_selector(resource.position, selector) def _matches_event_selector( self, event: Event, selector: Mapping[str, Any], ) -> bool: """ Return whether an event matches a selector. Parameters ---------- event: Candidate event. selector: Normalized selector mapping. Returns ------- bool True when the event matches. """ event_id = event.id or "" if "id" in selector and not self._value_matches_filter( event_id, selector["id"], ): return False if "ids" in selector and not self._value_matches_filter( event_id, selector["ids"], ): return False if "name" in selector and not self._value_matches_filter( event.name, selector["name"], ): return False if "names" in selector and not self._value_matches_filter( event.name, selector["names"], ): return False if "trigger_step" in selector and event.trigger_step != self._normalize_step( selector["trigger_step"], "trigger_step", ): return False if "enabled" in selector and event.enabled is not bool(selector["enabled"]): return False return True def _matches_id_and_type( self, object_id: str, object_type: str, selector: Mapping[str, Any], ) -> bool: """ Return whether an id/type pair matches a selector. Parameters ---------- object_id: Candidate id. object_type: Candidate type. selector: Selector mapping. Returns ------- bool True when id and type filters match. """ if "id" in selector and not self._value_matches_filter( object_id, selector["id"], ): return False if "ids" in selector and not self._value_matches_filter( object_id, selector["ids"], ): return False if "type" in selector and not self._value_matches_filter( object_type, selector["type"], ): return False if "types" in selector and not self._value_matches_filter( object_type, selector["types"], ): return False return True def _matches_mapping_filter( self, mapping: Mapping[str, Any], expected: Any, *, label: str, ) -> bool: """ Return whether a mapping matches exact key/value filters. Parameters ---------- mapping: Candidate mapping. expected: Expected mapping filter. label: Human-readable label for errors. Returns ------- bool True when all expected key/value filters match. """ if not isinstance(expected, Mapping): raise TypeError(f"Selector {label} filter must be a mapping.") for key, expected_value in expected.items(): if not isinstance(key, str): raise TypeError(f"Selector {label} keys must be strings.") if key not in mapping: return False if not self._value_matches_filter(mapping[key], expected_value): return False return True def _matches_numeric_thresholds( self, mapping: Mapping[str, Any], thresholds: Any, *, comparison: str, ) -> bool: """ Return whether numeric mapping values satisfy thresholds. Parameters ---------- mapping: Candidate mapping. thresholds: Threshold mapping. comparison: Either ``"min"`` or ``"max"``. Returns ------- bool True when all thresholds are satisfied. """ if not isinstance(thresholds, Mapping): raise TypeError("Numeric threshold selector must be a mapping.") for key, threshold in thresholds.items(): if not isinstance(key, str): raise TypeError("Numeric threshold selector keys must be strings.") if key not in mapping: return False value = mapping[key] if isinstance(value, bool) or not isinstance(value, Real): return False normalized_threshold = self._normalize_finite_float( threshold, f"{comparison}_threshold", ) if comparison == "min" and float(value) < normalized_threshold: return False if comparison == "max" and float(value) > normalized_threshold: return False return True def _matches_spatial_selector( self, position: NDArray[np.float64], selector: Mapping[str, Any], ) -> bool: """ Return whether a position matches spatial selector constraints. Parameters ---------- position: Candidate position. selector: Selector mapping. Returns ------- bool True when spatial filters match. """ if "near" in selector and selector["near"] is not None: near = selector["near"] if not isinstance(near, Mapping): raise TypeError("Selector 'near' must be a mapping.") target_position = self._normalize_position( self._require_parameter(near, "position") ) radius = self._normalize_non_negative_float( self._require_parameter(near, "radius"), "near.radius", ) distance = self._distance_or_none(position, target_position) return distance is not None and distance <= radius if "position" in selector and "radius" in selector: target_position = self._normalize_position(selector["position"]) radius = self._normalize_non_negative_float(selector["radius"], "radius") distance = self._distance_or_none(position, target_position) return distance is not None and distance <= radius if "position" in selector: target_position = self._normalize_position(selector["position"]) return position.shape == target_position.shape and bool( np.allclose(position, target_position) ) return True @staticmethod def _value_matches_filter(value: Any, expected: Any) -> bool: """ Return whether a value matches a scalar or collection filter. Parameters ---------- value: Candidate value. expected: Scalar value or collection of allowed values. Returns ------- bool True when the value matches. """ if isinstance(expected, (str, bytes)): return value == expected if isinstance(expected, AbcSequence) or isinstance(expected, set | frozenset): return value in expected return value == expected def _effect_result( self, *, effect: EventEffect, success: bool, affected_count: int, message: str, metadata: Mapping[str, Any] | None = None, ) -> EventEffectResult: """ Build an EventEffectResult. Parameters ---------- effect: Source effect. success: Whether the operation succeeded. affected_count: Number of affected objects. message: Human-readable result message. metadata: Optional structured result metadata. Returns ------- EventEffectResult Structured effect result. """ return EventEffectResult( target=effect.target, operation=effect.operation, success=success, affected_count=affected_count, message=message, metadata=dict(metadata or {}), ) def _state_updates_from_parameters( self, parameters: Mapping[str, Any], ) -> dict[str, Any]: """ Extract agent state updates from effect parameters. Parameters ---------- parameters: Effect parameters. Returns ------- dict[str, Any] State updates. """ if "updates" in parameters: updates = parameters["updates"] if not isinstance(updates, Mapping): raise TypeError("State updates must be a mapping.") self._validate_string_keys(updates, "state_updates") return deepcopy(dict(updates)) key = self._require_string_parameter(parameters, "key") if "value" not in parameters: raise KeyError("State update requires either 'updates' or 'value'.") return {key: deepcopy(parameters["value"])} def _mapping_updates_from_parameters( self, parameters: Mapping[str, Any], ) -> dict[str, Any]: """ Extract generic mapping updates from effect parameters. Parameters ---------- parameters: Effect parameters. Returns ------- dict[str, Any] Mapping updates. """ if "updates" in parameters: updates = parameters["updates"] if not isinstance(updates, Mapping): raise TypeError("Mapping updates must be a mapping.") self._validate_string_keys(updates, "mapping_updates") return deepcopy(dict(updates)) key = self._require_string_parameter(parameters, "key") if "value" not in parameters: raise KeyError("Mapping update requires either 'updates' or 'value'.") return {key: deepcopy(parameters["value"])} def _numeric_state_value( self, 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 the key is missing. 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." ) return float(value) def _resolve_query_position( self, *, agent: Agent | None, position: PositionInput | None, ) -> NDArray[np.float64]: """ Resolve the query position from an explicit position or agent. Parameters ---------- agent: Optional observing agent. position: Optional explicit position. Returns ------- NDArray[np.float64] Normalized query position. """ if position is not None: return self._normalize_position(position) if agent is not None: return agent.position.copy() raise ValueError("query requires either position or agent.") def _normalize_selector(self, selector: Selector | None) -> dict[str, Any]: """ Normalize a selector mapping. Parameters ---------- selector: Optional selector mapping. Returns ------- dict[str, Any] Normalized selector. """ if selector is None: return {} if not isinstance(selector, Mapping): raise TypeError("selector must be a mapping.") normalized = dict(selector) self._validate_string_keys(normalized, "selector") return normalized @staticmethod def _normalize_operation(operation: str) -> str: """ Normalize an operation name. Parameters ---------- operation: Operation name. Returns ------- str Normalized operation name. """ if not isinstance(operation, str) or not operation.strip(): raise ValueError("operation must be a non-empty string.") return operation.strip().lower().replace("-", "_") @staticmethod def _normalize_identifier(value: Any, label: str) -> str: """ Normalize an identifier. Parameters ---------- value: Candidate identifier. label: Human-readable label. Returns ------- str Normalized identifier. """ if not isinstance(value, str) or not value.strip(): raise ValueError(f"{label} must be a non-empty string.") return value.strip() @staticmethod def _normalize_step(value: Any, label: str) -> int: """ Normalize a non-negative integer step value. Parameters ---------- value: Candidate step value. label: Human-readable label. Returns ------- int Normalized step. """ if isinstance(value, bool) or not isinstance(value, Integral): raise TypeError(f"{label} must be an integer.") normalized = int(value) if normalized < 0: raise ValueError(f"{label} cannot be negative.") return normalized @staticmethod def _normalize_positive_int(value: Any, label: str) -> int: """ Normalize a positive integer. Parameters ---------- value: Candidate integer. label: Human-readable label. Returns ------- int Positive integer. """ if isinstance(value, bool) or not isinstance(value, Integral): raise TypeError(f"{label} must be an integer.") normalized = int(value) if normalized <= 0: raise ValueError(f"{label} must be positive.") return normalized @staticmethod def _normalize_seed(value: Any) -> int | None: """ Normalize an optional random seed. Parameters ---------- value: Candidate seed. Returns ------- int | None Normalized seed. """ if value is None: return None if isinstance(value, bool) or not isinstance(value, Integral): raise TypeError("seed must be an integer or None.") normalized = int(value) if normalized < 0: raise ValueError("seed cannot be negative.") return normalized @staticmethod def _normalize_position(position: PositionInput) -> NDArray[np.float64]: """ Normalize a position into a one-dimensional NumPy vector. Parameters ---------- position: Candidate position. Returns ------- NDArray[np.float64] Normalized position. """ normalized = np.asarray(position, dtype=np.float64) if normalized.ndim != 1: raise ValueError("position must be one-dimensional.") if normalized.size == 0: raise ValueError("position cannot be empty.") if not np.all(np.isfinite(normalized)): raise ValueError("position must contain only finite values.") return normalized @staticmethod def _normalize_finite_float(value: Any, label: str) -> float: """ Normalize a finite float. Parameters ---------- value: Candidate numeric 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 numeric 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 def _normalize_optional_non_negative_float( self, value: Any, label: str, ) -> float | None: """ Normalize an optional non-negative finite float. Parameters ---------- value: Candidate numeric value or None. label: Human-readable label. Returns ------- float | None Normalized value or None. """ if value is None: return None return self._normalize_non_negative_float(value, label) def _require_parameter(self, parameters: Mapping[str, Any], key: str) -> Any: """ Return a required parameter value. Parameters ---------- parameters: Parameter mapping. key: Required key. Returns ------- Any Parameter value. """ if key not in parameters: raise KeyError(f"Missing required parameter '{key}'.") return parameters[key] def _require_string_parameter( self, parameters: Mapping[str, Any], key: str, ) -> str: """ Return a required string parameter. Parameters ---------- parameters: Parameter mapping. key: Required key. Returns ------- str Normalized string parameter. """ value = self._require_parameter(parameters, key) if not isinstance(value, str) or not value.strip(): raise ValueError(f"Parameter '{key}' must be a non-empty string.") return value.strip() def _require_position_parameter( self, parameters: Mapping[str, Any], key: str, ) -> NDArray[np.float64]: """ Return a required position parameter. Parameters ---------- parameters: Parameter mapping. key: Required key. Returns ------- NDArray[np.float64] Normalized position. """ return self._normalize_position(self._require_parameter(parameters, key)) def _require_finite_float_parameter( self, parameters: Mapping[str, Any], key: str, ) -> float: """ Return a required finite float parameter. Parameters ---------- parameters: Parameter mapping. key: Required key. Returns ------- float Normalized finite float. """ return self._normalize_finite_float( self._require_parameter(parameters, key), key, ) def _require_non_negative_float_parameter( self, parameters: Mapping[str, Any], key: str, ) -> float: """ Return a required non-negative float parameter. Parameters ---------- parameters: Parameter mapping. key: Required key. Returns ------- float Normalized non-negative float. """ return self._normalize_non_negative_float( self._require_parameter(parameters, key), key, ) @staticmethod def _optional_bool_parameter( parameters: Mapping[str, Any], key: str, *, default: bool, ) -> bool: """ Return an optional boolean parameter. Parameters ---------- parameters: Parameter mapping. key: Optional key. default: Default boolean value. Returns ------- bool Boolean parameter. """ if key not in parameters: return default value = parameters[key] if not isinstance(value, bool): raise TypeError(f"Parameter '{key}' must be a boolean.") return value def _validate_spatial_dimensions(self) -> None: """ Validate that all positioned objects share one dimensionality. Raises ------ ValueError If multiple spatial dimensionalities are present. """ _ = self.dimension def _validate_position_dimension(self, position: NDArray[np.float64]) -> None: """ Validate a position against the world's existing dimensionality. Parameters ---------- position: Position to validate. Raises ------ ValueError If the position dimensionality does not match the world. """ current_dimension = self.dimension if current_dimension is None: return if int(position.size) != current_dimension: raise ValueError( f"Position dimension {position.size} does not match world " f"dimension {current_dimension}." ) @staticmethod def _distance_or_none( left: NDArray[np.float64], right: NDArray[np.float64], ) -> float | None: """ Compute distance between positions, returning None on mismatch. Parameters ---------- left: First position. right: Second position. Returns ------- float | None Euclidean distance, or None when dimensions differ. """ if left.shape != right.shape: return None return float(np.linalg.norm(left - right)) @staticmethod def _normalize_optional_string_filter( value: str | AbcSequence[str] | None, ) -> set[str] | None: """ Normalize an optional string or sequence-of-strings filter. Parameters ---------- value: Filter value. Returns ------- set[str] | None Set of allowed strings, or None. """ if value is None: return None if isinstance(value, str): return {value} if not isinstance(value, AbcSequence): raise TypeError("String filter must be a string or sequence of strings.") normalized: set[str] = set() for item in value: if not isinstance(item, str): raise TypeError("String filter sequences must contain strings.") normalized.add(item) return normalized @staticmethod def _validate_string_keys(mapping: Mapping[str, Any], label: str) -> None: """ Validate that all mapping keys are strings. Parameters ---------- mapping: Mapping to validate. label: Human-readable mapping label. Raises ------ TypeError If any key is not a string. """ for key in mapping: if not isinstance(key, str): raise TypeError(f"World.{label} keys must be strings.") __all__ = [ "PositionInput", "SchedulerProtocol", "Selector", "World", "WorldStepResult", ]