unienv_interface.world.funcnodes.flat_combined_funcnode¶
FlatCombinedFuncWorldNode - A functional node that flattens combined node data structures.
Unlike CombinedFuncWorldNode which nests data under node names as keys, FlatCombinedFuncWorldNode merges dictionary values directly, requiring all nodes to have DictSpace observations/actions/contexts with unique keys.
FlatCombinedFuncWorldNode(name: str, nodes: Iterable[FuncWorldNode[WorldStateT, Any, Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]], render_mode: Optional[str] = 'auto')
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.
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'
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If node spaces have overlapping keys or non-DictSpace types |
instance-attribute
¶
context_space = _flatten_spaces([(context_space) for node in (nodes) if context_space is not None], ignore_duplicate_keys=False)
instance-attribute
¶
observation_space = _flatten_spaces([(observation_space) for node in (nodes) if observation_space is not None], ignore_duplicate_keys=False)
instance-attribute
¶
action_space = _flatten_spaces([(action_space) for node in (nodes) if action_space is not None], ignore_duplicate_keys=True)
instance-attribute
¶
has_termination_signal = any((has_termination_signal) for node in nodes)
instance-attribute
¶
has_truncation_signal = any((has_truncation_signal) for node in 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(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.
set_next_action(world_state: WorldStateT, node_state: CombinedNodeStateT, action: CombinedDataT) -> tuple[WorldStateT, CombinedNodeStateT]
Set action by routing keys to appropriate nodes.
initial(world_state: WorldStateT, *, priority: int = 0, seed: Optional[int] = None, pernode_kwargs: Dict[str, Dict[str, Any]] = {}) -> Tuple[WorldStateT, CombinedNodeStateT]
Create child node states 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 child node states 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 children 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]
post_environment_step(world_state: WorldStateT, node_state: CombinedNodeStateT, dt: Union[float, BArrayType], *, priority: int = 0) -> Tuple[WorldStateT, CombinedNodeStateT]
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.
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.