unienv_interface.world.nodes.combined_node¶
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
¶
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)
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.
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.
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.