unienv_interface.world.nodes.flat_combined_node¶
FlatCombinedWorldNode - A stateful node that flattens combined node data structures.
Unlike CombinedWorldNode which nests data under node names as keys, FlatCombinedWorldNode merges dictionary values directly, requiring all nodes to have DictSpace observations/actions/contexts with unique keys.
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.
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.
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'.