unienv_interface.world.nodes¶
Stateful node implementations and combinators.
CombinedWorldNode(name: str, nodes: Iterable[WorldNode[Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]], direct_return: bool = True, render_mode: Optional[str] = 'auto', root_node: Optional[WorldNode[Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]] = None)
Bases: WorldNode[Optional[CombinedDataT], CombinedDataT, CombinedDataT, BArrayType, BDeviceType, BDtypeType, BRNGType], Generic[BArrayType, BDeviceType, BDtypeType, BRNGType]
A WorldNode that combines multiple WorldNodes into one node, using a dictionary to store the data from each node.
The observation, reward, termination, truncation, and info are combined from all child nodes.
The keys in the dictionary are the names of the child nodes.
If there is only one child node that supports value and direct_return is set to True, the value is returned directly instead of a dictionary.
Root node¶
An optional root_node may be passed next to nodes. It is a regular
node (sharing the same :class:~unienv_interface.world.world.World) that is
NOT part of the public nodes list and whose data merges at the TOP LEVEL
of the aggregated dictionaries instead of nesting under its own name. It
rides the same machinery as a child node -- priorities, control/update rate
ratios, lifecycle dispatch, rewards/signals, info, render and close -- under
the reserved internal key :data:ROOT_NODE_KEY (""), which cannot
collide with a child node name (child names are validated non-empty).
- Spaces / data (context, observation, action -- independently). When the
root node is the sole contributor of a channel, the existing
direct_returnrule applies verbatim:direct_return=Truepasses the root's own space/data through unwrapped (anySpacetype, e.g. a plainBoxSpace), whiledirect_return=Falsewraps it under the reserved key (DictSpace({ROOT_NODE_KEY: space})/{ROOT_NODE_KEY: data}). When the root and at least one named child contribute, the root MUST expose aDictSpacefor that channel and its keys are merged next to the child entries:{child_name: child_space, ..., **root_space.spaces}(resp.{child_name: child_data, ..., **root_data}). A root key equal to a child node name raisesValueError. - Actions. For a sole-root action channel the aggregate action space is the root's action space and the whole action value is forwarded to the root at its own control-rate tick. In the mixed case the incoming mapping is routed by key: keys matching action-child names update those children's caches (exactly as before), and every remaining key accumulates into the root's action entry as a dict. Each consumer is dispatched at its own ratio tick with hold-last semantics identical to children.
- Info / render. These channels nest by node name for children, so the
root's entries are keyed by the root's own
name(the reserved key is only used for internal state/routing and aggregation). get_nodenever resolves the reserved key (it is not a user-facing path);get_nodes_by_fndoes traverse the root node.
class-attribute
instance-attribute
¶
context_space: Optional[Space[ContextType, BDeviceType, BDtypeType, BRNGType]] = None
class-attribute
instance-attribute
¶
observation_space: Optional[Space[ObsType, BDeviceType, BDtypeType, BRNGType]] = None
class-attribute
instance-attribute
¶
action_space: Optional[Space[ActType, BDeviceType, BDtypeType, BRNGType]] = None
property
¶
backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType]
Backend provided by the attached world.
class-attribute
instance-attribute
¶
supported_render_modes = ('dict', 'auto')
instance-attribute
¶
has_termination_signal = any(node.has_termination_signal for node in self._all_nodes)
instance-attribute
¶
has_truncation_signal = any(node.has_truncation_signal for node in self._all_nodes)
get_nodes_by_type(node_type: Type[WorldNode]) -> list[WorldNode]
Return nodes in this subtree that are instances of node_type.
has_wrapper_attr(name: str) -> bool
Checks if the attribute name exists in the environment.
set_wrapper_attr(name: str, value: Any)
Sets the attribute name on the environment with value.
staticmethod
¶
aggregate_spaces(spaces: Dict[str, Optional[Space[Any, BDeviceType, BDtypeType, BRNGType]]], direct_return: bool = True) -> Tuple[Optional[str], Optional[DictSpace[BDeviceType, BDtypeType, BRNGType]]]
Aggregate child spaces into either a passthrough or a DictSpace.
staticmethod
¶
aggregate_data(data: Dict[str, Any], direct_return: bool = True) -> Optional[Union[Dict[str, Any], Any]]
Aggregate child outputs using the same direct-return convention as spaces.
get_priority_order(phase: str) -> List[int]
Return the cached descending priority order for a named lifecycle phase.
Supported phase names: 'reload', 'after_reload', 'reset',
'after_reset', 'pre_step', 'post_step'.
get_node(nested_keys: Union[str, Sequence[str]]) -> Optional[WorldNode]
Resolve a child node by dotted-path-like key traversal.
get_nodes_by_fn(fn: Callable[[WorldNode], bool]) -> list[WorldNode]
Return every node in the subtree that satisfies fn (root included).
pre_environment_step(dt, *, priority: int = 0)
Dispatch pre-step callbacks to the nodes (children and the optional root node) at the matching frequency.
set_next_action(action)
Route combined actions to child nodes (and the optional root node), respecting per-node control rates.
The incoming action is split by :meth:_split_child_actions (children
only). When a root_node owns an action space and no single node is
the direct-action passthrough provider, the keys that do not belong to
any named child are the root's action keys: they accumulate into the
root's cache entry under the reserved key as a dict. Every consumer
(children and root) is then dispatched at its own ratio tick, and only
once it has received at least one action (hold-last semantics, so a
child-only step re-delivers the root's previous action at its tick and
vice versa).
post_environment_step(dt, *, priority: int = 0)
Dispatch post-step callbacks to the nodes (children and the optional root node) at the matching frequency.
reset(*, priority: int = 0, seed=None, mask=None, pernode_kwargs: Dict[str, Any] = {})
Forward reset calls to the nodes (children and the optional root node)
that participate at priority.
reload(*, priority: int = 0, seed=None, mask=None, pernode_kwargs: Dict[str, Any] = {})
Forward reload calls to the nodes (children and the optional root node)
that participate at priority.
after_reset(*, priority: int = 0, mask=None)
Forward post-reset hooks and reset internal routing counters.
after_reload(*, priority: int = 0, mask=None)
Call after_reload on the nodes (children and the optional root node). Similar to after_reset but for the reload flow.
FlatCombinedWorldNode(name: str, nodes: Iterable[WorldNode[Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]], render_mode: Optional[str] = 'auto', root_node: Optional[WorldNode[Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]] = None)
Bases: CombinedWorldNode[BArrayType, BDeviceType, BDtypeType, BRNGType]
A WorldNode that combines multiple WorldNodes and flattens their data.
Unlike CombinedWorldNode which stores data as {node_name: {key: value}}, FlatCombinedWorldNode merges dictionaries directly as {key1: value1, key2: value2}.
This requires: - All nodes with observation/action/context spaces must use DictSpace - Keys across all nodes must be unique (no overlaps)
The node names are only used for identification, not for nesting data.
Root node¶
An optional root_node (passed separately, sharing the same world and not
listed in nodes) is handled as one more flat contributor:
- Its context / observation spaces and data are appended to the flatten list,
so its keys appear directly at the top level next to the children's keys
(an overlapping key raises
ValueError). When the root is the only contributor its space/data is returned unwrapped, mirroring the existing single-child flat behavior. - For actions, every key not claimed by any child's action space is routed to
the root under the reserved internal key, while claimed keys are sliced to
their children exactly as before. Without a
root_nodesuch unclaimed keys are rejected with aValueError. - Rewards / signals / info / render and the lifecycle hooks include the root
like any other node (info and render use the root's own
name).
Initialize a FlatCombinedWorldNode.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Name of this combined node |
required |
nodes
|
Iterable[WorldNode[Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]]
|
Iterable of nodes to combine |
required |
render_mode
|
Optional[str]
|
Render mode ('dict', 'auto', or specific mode) |
'auto'
|
root_node
|
Optional[WorldNode[Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]]
|
Optional node merged at the top level of the flattened data instead of nesting under its own name |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If node spaces have overlapping keys or non-DictSpace types |
class-attribute
instance-attribute
¶
context_space: Optional[Space[ContextType, BDeviceType, BDtypeType, BRNGType]] = None
class-attribute
instance-attribute
¶
observation_space: Optional[Space[ObsType, BDeviceType, BDtypeType, BRNGType]] = None
class-attribute
instance-attribute
¶
action_space: Optional[Space[ActType, BDeviceType, BDtypeType, BRNGType]] = None
instance-attribute
¶
has_termination_signal = any(node.has_termination_signal for node in self._all_nodes)
instance-attribute
¶
has_truncation_signal = any(node.has_truncation_signal for node in self._all_nodes)
class-attribute
instance-attribute
¶
supported_render_modes = ('dict', 'auto')
property
¶
backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType]
Backend provided by the attached world.
pre_environment_step(dt, *, priority: int = 0)
Dispatch pre-step callbacks to the nodes (children and the optional root node) at the matching frequency.
set_next_action(action)
Route combined actions to child nodes (and the optional root node), respecting per-node control rates.
The incoming action is split by :meth:_split_child_actions (children
only). When a root_node owns an action space and no single node is
the direct-action passthrough provider, the keys that do not belong to
any named child are the root's action keys: they accumulate into the
root's cache entry under the reserved key as a dict. Every consumer
(children and root) is then dispatched at its own ratio tick, and only
once it has received at least one action (hold-last semantics, so a
child-only step re-delivers the root's previous action at its tick and
vice versa).
post_environment_step(dt, *, priority: int = 0)
Dispatch post-step callbacks to the nodes (children and the optional root node) at the matching frequency.
reset(*, priority: int = 0, seed=None, mask=None, pernode_kwargs: Dict[str, Any] = {})
Forward reset calls to the nodes (children and the optional root node)
that participate at priority.
reload(*, priority: int = 0, seed=None, mask=None, pernode_kwargs: Dict[str, Any] = {})
Forward reload calls to the nodes (children and the optional root node)
that participate at priority.
after_reset(*, priority: int = 0, mask=None)
Forward post-reset hooks and reset internal routing counters.
after_reload(*, priority: int = 0, mask=None)
Call after_reload on the nodes (children and the optional root node). Similar to after_reset but for the reload flow.
get_node(nested_keys: Union[str, Sequence[str]]) -> Optional[WorldNode]
Resolve a child node by dotted-path-like key traversal.
get_nodes_by_fn(fn: Callable[[WorldNode], bool]) -> list[WorldNode]
Return every node in the subtree that satisfies fn (root included).
get_nodes_by_type(node_type: Type[WorldNode]) -> list[WorldNode]
Return nodes in this subtree that are instances of node_type.
has_wrapper_attr(name: str) -> bool
Checks if the attribute name exists in the environment.
set_wrapper_attr(name: str, value: Any)
Sets the attribute name on the environment with value.
staticmethod
¶
aggregate_spaces(spaces: Dict[str, Optional[Space[Any, BDeviceType, BDtypeType, BRNGType]]], direct_return: bool = True) -> Tuple[Optional[str], Optional[DictSpace[BDeviceType, BDtypeType, BRNGType]]]
Aggregate child spaces into either a passthrough or a DictSpace.
staticmethod
¶
aggregate_data(data: Dict[str, Any], direct_return: bool = True) -> Optional[Union[Dict[str, Any], Any]]
Aggregate child outputs using the same direct-return convention as spaces.
get_priority_order(phase: str) -> List[int]
Return the cached descending priority order for a named lifecycle phase.
Supported phase names: 'reload', 'after_reload', 'reset',
'after_reset', 'pre_step', 'post_step'.
get_context() -> Optional[CombinedDataT]
Get context by flattening all node contexts (root node included) into one dictionary.
get_observation() -> CombinedDataT
Get observation by flattening all node observations (root node included) into one dictionary.
EnvAsWorldNode(env: Env, name: str, *, world: Optional[World] = None, control_timestep: Optional[float] = None, update_timestep: Optional[float] = None, hold_last_action: bool = False, expose_reward: bool = True, expose_termination: bool = True, expose_truncation: bool = True, forward_info: bool = True, reset_seed: Optional[int] = None, pre_step_priority: int = 0)
Bases: WorldNode[ContextType, ObsType, ActType, BArrayType, BDeviceType, BDtypeType, BRNGType]
Adapt an existing unbatched :class:Env so it can be composed as a WorldNode.
Design¶
Env.step(action) is atomic: it produces the next observation, reward,
termination/truncation signals, and info in a single call. A WorldNode
instead splits the step lifecycle into::
set_next_action(action)
-> pre_environment_step(dt, priority=...)
-> World.step()
-> post_environment_step(dt, priority=...)
-> get_observation() / get_reward() / ... / get_info()
This adapter bridges the mismatch by performing the atomic inner
Env.step inside pre_environment_step (using the action cached by the
preceding set_next_action call) and caching its five-tuple result for
the get_* accessors to expose. reset maps directly to
Env.reset, and the base WorldNode.reload default (which delegates to
reset) is exactly right for a pre-built environment, hence the
non-empty reset_priorities / reload_priorities sets.
The dt argument of pre_environment_step is intentionally ignored:
the wrapped environment is self-paced (e.g. real hardware running at its own
control rate), while the outer world is expected to be a bookkeeping-only
world such as :class:RealWorld. control_timestep / update_timestep
can still be set for step-ratio bookkeeping by the composer.
No schema transformation / re-keying is performed: the inner environment's action, observation, and context spaces (and the raw observation / context layouts) pass through unchanged.
Only unbatched environments (batch_size is None) are supported. The
optional world is only used for backend/device delegation; when provided
its backend type must match the inner environment's backend, since the node
serves the inner environment's arrays.
property
¶
effective_update_timestep: Optional[float]
Resolved update period, defaulting to control_timestep when unset.
class-attribute
instance-attribute
¶
pre_environment_step_priorities = {pre_step_priority}
class-attribute
instance-attribute
¶
post_environment_step_priorities = set()
instance-attribute
¶
update_timestep = update_timestep if update_timestep is not None else control_timestep
instance-attribute
¶
supported_render_modes = tuple(env.metadata.get('render_modes', ()))
property
¶
backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType]
Backend of the attached world, falling back to the inner env's backend.
property
¶
device: Optional[BDeviceType]
Device of the attached world, falling back to the inner env's device.
post_environment_step(dt: Union[float, BArrayType], *, priority: int = 0) -> None
This method is called after the environment step to update the sensor's internal state. Args: dt (float/BArrayType): The time elapsed between the last world step and the current step.
reload(*, priority: int = 0, seed: Optional[int] = None, mask: Optional[BArrayType] = None, **kwargs) -> None
Reload the node. By default, this just calls reset with the same parameters.
Simulation environments can override this to completely re-read assets and reload the node.
after_reset(*, priority: int = 0, mask: Optional[BArrayType] = None) -> None
This method is called after all WorldNodes has been called with reset (e.g. the environment reset is effectively done).
Use get_context, get_observation, and get_info to read the post-reset state.
after_reload(*, priority: int = 0, mask: Optional[BArrayType] = None) -> None
This method is called after the world has been rebuilt following a reload.
At this point: - All nodes have added their entities during the reload phase - The world scene has been built (simulation is ready) - Nodes can now cache references to entities, geoms, etc.
By default, this calls after_reset(). Override if you need specific
post-reload initialization that differs from post-reset.
Use get_context, get_observation, and get_info to read the state.
get_node(nested_keys: Union[str, Sequence[str]]) -> Optional[WorldNode]
Return a node by key path.
strkeys are treated as a single direct-child key.- Empty key sequence returns
self. - Vanilla nodes have no children and return
Nonefor non-empty lookup.
get_nodes_by_fn(fn: Callable[[WorldNode], bool]) -> list[WorldNode]
Return nodes in this subtree matching fn.
Vanilla nodes only evaluate self.
get_nodes_by_type(node_type: Type[WorldNode]) -> list[WorldNode]
Return nodes in this subtree that are instances of node_type.
has_wrapper_attr(name: str) -> bool
Checks if the attribute name exists in the environment.
set_wrapper_attr(name: str, value: Any)
Sets the attribute name on the environment with value.
set_next_action(action: ActType) -> None
Cache the action consumed by the next pre_environment_step call.
action=None is only accepted when hold_last_action=True (the
previously cached action is kept); in that case a node that has never
received an action still raises from pre_environment_step.
pre_environment_step(dt: Union[float, BArrayType], *, priority: int = 0) -> None
Run one atomic Env.step with the cached action.
The world-provided dt is ignored: the wrapped environment is
self-paced, so it advances by exactly one of its own steps per call.
reset(*, priority: int = 0, seed: Optional[int] = None, mask: Optional[BArrayType] = None, **kwargs) -> None
Reset the inner environment (also used as the default reload).
get_context() -> Optional[ContextType]
Return the context produced by the most recent inner reset.
get_observation() -> ObsType
Return the latest step observation, or the reset observation before the first step.
get_reward() -> Union[float, BArrayType]
Return the cached step reward (0.0 before the first step).
get_termination() -> Union[bool, BArrayType]
Return the cached termination signal (False before the first step).
get_truncation() -> Union[bool, BArrayType]
Return the cached truncation signal (False before the first step).
get_info() -> Optional[Dict[str, Any]]
Return a copy of the latest info dict when forward_info is enabled.