unienv_interface.world¶
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.
CombinedFuncWorldNode(name: str, nodes: Iterable[FuncWorldNode[WorldStateT, Any, Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]], direct_return: bool = True, render_mode: Optional[str] = 'auto', root_node: Optional[FuncWorldNode[WorldStateT, Any, Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]] = None)
Bases: FuncWorldNode[WorldStateT, CombinedNodeStateT, Optional[CombinedDataT], CombinedDataT, CombinedDataT, BArrayType, BDeviceType, BDtypeType, BRNGType], Generic[WorldStateT, BArrayType, BDeviceType, BDtypeType, BRNGType]
A functional counterpart to CombinedWorldNode that composes multiple FuncWorldNodes.
It aggregates spaces (context, observation, action) and runtime data (context, observation, info, reward, termination, truncation)
across child nodes. If only one child exposes a given interface and direct_return=True, the value is passed through directly.
Root node¶
An optional root_node may be passed next to nodes. It is a regular
FuncWorldNode (sharing the same FuncWorld) 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 -- and its node state is
stored in the combined node_state under the reserved key
:data:ROOT_NODE_KEY (""), a string key that cannot collide with a child
node name (child names are validated non-empty) and therefore keeps persisted
CombinedNodeStateT dictionaries loadable without migration.
- Spaces / data (context, observation, action -- independently). Spaces are
aggregated once at construction (functional nodes declare static spaces).
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 cache entries (exactly as before) and every remaining key accumulates into the root's entry (stored under the reserved key) 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[FuncWorldNode]) -> list[FuncWorldNode]
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_node(nested_keys: Union[str, Sequence[str]]) -> Optional[FuncWorldNode]
Resolve a child node by dotted-path-like key traversal.
get_nodes_by_fn(fn: Callable[[FuncWorldNode], bool]) -> list[FuncWorldNode]
Return every node in the subtree that satisfies fn (root included).
initial(world_state: WorldStateT, *, priority: int = 0, seed: Optional[int] = None, pernode_kwargs: Dict[str, Dict[str, Any]] = {}) -> Tuple[WorldStateT, CombinedNodeStateT]
Create node states (children and the optional root node, whose state is stored under the reserved key) for the current initialization priority.
reload(world_state: WorldStateT, *, priority: int = 0, seed: Optional[int] = None, pernode_kwargs: Dict[str, Dict[str, Any]] = {}) -> Tuple[WorldStateT, CombinedNodeStateT]
Recreate node states (children and the optional root node) for the current reload priority.
reset(world_state: WorldStateT, node_state: CombinedNodeStateT, *, priority: int = 0, seed: Optional[int] = None, mask: Optional[BArrayType] = None, pernode_kwargs: Dict[str, Dict[str, Any]] = {}, **kwargs) -> Tuple[WorldStateT, CombinedNodeStateT]
Forward reset calls to the nodes (children and the optional root node)
that participate at priority.
after_reset(world_state: WorldStateT, node_state: CombinedNodeStateT, *, priority: int = 0, mask: Optional[BArrayType] = None) -> Tuple[WorldStateT, CombinedNodeStateT]
Forward post-reset hooks and reset internal routing counters.
after_reload(world_state: WorldStateT, node_state: CombinedNodeStateT, *, priority: int = 0, mask: Optional[BArrayType] = None) -> Tuple[WorldStateT, CombinedNodeStateT]
Call after_reload on child nodes. Similar to after_reset but for reload flow.
pre_environment_step(world_state: WorldStateT, node_state: CombinedNodeStateT, dt: Union[float, BArrayType], *, priority: int = 0) -> Tuple[WorldStateT, CombinedNodeStateT]
set_next_action(world_state: WorldStateT, node_state: CombinedNodeStateT, action: CombinedDataT) -> Tuple[WorldStateT, CombinedNodeStateT]
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 (stored 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(world_state: WorldStateT, node_state: CombinedNodeStateT, dt: Union[float, BArrayType], *, priority: int = 0) -> Tuple[WorldStateT, CombinedNodeStateT]
close(world_state: WorldStateT, node_state: CombinedNodeStateT) -> WorldStateT
Close every child node and the optional root node.
get_context(world_state: WorldStateT, node_state: CombinedNodeStateT) -> CombinedDataT
get_observation(world_state: WorldStateT, node_state: CombinedNodeStateT) -> CombinedDataT
get_reward(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Union[float, BArrayType]
get_termination(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Union[bool, BArrayType]
get_truncation(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Union[bool, BArrayType]
get_info(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Optional[Dict[str, Any]]
FlatCombinedFuncWorldNode(name: str, nodes: Iterable[FuncWorldNode[WorldStateT, Any, Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]], render_mode: Optional[str] = 'auto', root_node: Optional[FuncWorldNode[WorldStateT, Any, Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]] = None)
Bases: CombinedFuncWorldNode[WorldStateT, BArrayType, BDeviceType, BDtypeType, BRNGType]
A FuncWorldNode that combines multiple FuncWorldNodes and flattens their data.
Unlike CombinedFuncWorldNode which stores data as {node_name: {key: value}}, FlatCombinedFuncWorldNode 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 (its state lives under that same
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 FlatCombinedFuncWorldNode.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Name of this combined node |
required |
nodes
|
Iterable[FuncWorldNode[WorldStateT, Any, 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[FuncWorldNode[WorldStateT, Any, 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 |
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.
instance-attribute
¶
observation_space = self._flatten_channel_spaces('observation_space')
initial(world_state: WorldStateT, *, priority: int = 0, seed: Optional[int] = None, pernode_kwargs: Dict[str, Dict[str, Any]] = {}) -> Tuple[WorldStateT, CombinedNodeStateT]
Create node states (children and the optional root node, whose state is stored under the reserved key) for the current initialization priority.
reload(world_state: WorldStateT, *, priority: int = 0, seed: Optional[int] = None, pernode_kwargs: Dict[str, Dict[str, Any]] = {}) -> Tuple[WorldStateT, CombinedNodeStateT]
Recreate node states (children and the optional root node) for the current reload priority.
reset(world_state: WorldStateT, node_state: CombinedNodeStateT, *, priority: int = 0, seed: Optional[int] = None, mask: Optional[BArrayType] = None, pernode_kwargs: Dict[str, Dict[str, Any]] = {}, **kwargs) -> Tuple[WorldStateT, CombinedNodeStateT]
Forward reset calls to the nodes (children and the optional root node)
that participate at priority.
after_reset(world_state: WorldStateT, node_state: CombinedNodeStateT, *, priority: int = 0, mask: Optional[BArrayType] = None) -> Tuple[WorldStateT, CombinedNodeStateT]
Forward post-reset hooks and reset internal routing counters.
after_reload(world_state: WorldStateT, node_state: CombinedNodeStateT, *, priority: int = 0, mask: Optional[BArrayType] = None) -> Tuple[WorldStateT, CombinedNodeStateT]
Call after_reload on child nodes. Similar to after_reset but for reload flow.
pre_environment_step(world_state: WorldStateT, node_state: CombinedNodeStateT, dt: Union[float, BArrayType], *, priority: int = 0) -> Tuple[WorldStateT, CombinedNodeStateT]
get_reward(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Union[float, BArrayType]
get_termination(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Union[bool, BArrayType]
get_truncation(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Union[bool, BArrayType]
set_next_action(world_state: WorldStateT, node_state: CombinedNodeStateT, action: CombinedDataT) -> Tuple[WorldStateT, CombinedNodeStateT]
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 (stored 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(world_state: WorldStateT, node_state: CombinedNodeStateT, dt: Union[float, BArrayType], *, priority: int = 0) -> Tuple[WorldStateT, CombinedNodeStateT]
close(world_state: WorldStateT, node_state: CombinedNodeStateT) -> WorldStateT
Close every child node and the optional root node.
get_node(nested_keys: Union[str, Sequence[str]]) -> Optional[FuncWorldNode]
Resolve a child node by dotted-path-like key traversal.
get_nodes_by_fn(fn: Callable[[FuncWorldNode], bool]) -> list[FuncWorldNode]
Return every node in the subtree that satisfies fn (root included).
get_nodes_by_type(node_type: Type[FuncWorldNode]) -> list[FuncWorldNode]
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_context(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Optional[CombinedDataT]
Get context by flattening all node contexts into one dictionary.
get_observation(world_state: WorldStateT, node_state: CombinedNodeStateT) -> CombinedDataT
Get observation by flattening all node observations into one dictionary.
get_info(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Optional[Dict[str, Any]]
Get info by merging all node info dictionaries.