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

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_node such unclaimed keys are rejected with a ValueError.
• 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

context_space instance-attribute

context_space = self._flatten_channel_spaces('context_space')

observation_space instance-attribute

observation_space = self._flatten_channel_spaces('observation_space')

action_space instance-attribute

action_space = self._flatten_channel_spaces('action_space')

name instance-attribute

name = name

control_timestep property

control_timestep

update_timestep property

update_timestep

has_reward instance-attribute

has_reward = any(node.has_reward for node in self._all_nodes)

has_termination_signal instance-attribute

has_termination_signal = any(node.has_termination_signal for node in self._all_nodes)

has_truncation_signal instance-attribute

has_truncation_signal = any(node.has_truncation_signal for node in self._all_nodes)

supported_render_modes class-attribute instance-attribute

supported_render_modes = ('dict', 'auto')

render_mode instance-attribute

render_mode = renderable_nodes[0].render_mode

world property

world

initial_priorities property

initial_priorities: Set[int]

reset_priorities property

reset_priorities: Set[int]

reload_priorities property

reload_priorities: Set[int]

after_reset_priorities property

after_reset_priorities: Set[int]

after_reload_priorities property

after_reload_priorities: Set[int]

pre_environment_step_priorities property

pre_environment_step_priorities: Set[int]

post_environment_step_priorities property

post_environment_step_priorities: Set[int]

backend property

backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType]

Backend provided by the attached world.

device property

device: Optional[BDeviceType]

Device provided by the attached world.

can_render property

can_render: bool

Whether the node currently exposes a render mode.

effective_update_timestep property

effective_update_timestep

unwrapped property

unwrapped: FuncWorldNode

prev_wrapper_layer property

prev_wrapper_layer: Optional[FuncWorldNode]

nodes instance-attribute

nodes = nodes

root_node instance-attribute

root_node = root_node

direct_return instance-attribute

direct_return = direct_return

get_context

get_context(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Optional[CombinedDataT]

Get context by flattening all node contexts into one dictionary.

get_observation

get_observation(world_state: WorldStateT, node_state: CombinedNodeStateT) -> CombinedDataT

Get observation by flattening all node observations into one dictionary.

get_info

get_info(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Optional[Dict[str, Any]]

Get info by merging all node info dictionaries.

initial

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

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

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

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

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

pre_environment_step(world_state: WorldStateT, node_state: CombinedNodeStateT, dt: Union[float, BArrayType], *, priority: int = 0) -> Tuple[WorldStateT, CombinedNodeStateT]

get_reward

get_reward(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Union[float, BArrayType]

get_termination

get_termination(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Union[bool, BArrayType]

get_truncation

get_truncation(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Union[bool, BArrayType]

render

render(world_state, node_state)

set_next_action

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

post_environment_step(world_state: WorldStateT, node_state: CombinedNodeStateT, dt: Union[float, BArrayType], *, priority: int = 0) -> Tuple[WorldStateT, CombinedNodeStateT]

close

close(world_state: WorldStateT, node_state: CombinedNodeStateT) -> WorldStateT

Close every child node and the optional root node.

get_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

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

get_nodes_by_type(node_type: Type[FuncWorldNode]) -> list[FuncWorldNode]

Return nodes in this subtree that are instances of node_type.

has_wrapper_attr

has_wrapper_attr(name: str) -> bool

Checks if the attribute name exists in the environment.

get_wrapper_attr

get_wrapper_attr(name: str) -> Any

Gets the attribute name from the environment.

set_wrapper_attr

set_wrapper_attr(name: str, value: Any)

Sets the attribute name on the environment with value.

aggregate_spaces 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.

aggregate_data 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.