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

FlatCombinedWorldNode(name: str, nodes: Iterable[WorldNode[Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]], render_mode: Optional[str] = 'auto')

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.

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'

Raises:

Type	Description
ValueError	If node spaces have overlapping keys or non-DictSpace types

name instance-attribute

name = name

control_timestep property

control_timestep: Optional[float]

update_timestep property

update_timestep: Optional[float]

context_space class-attribute instance-attribute

context_space: Optional[Space[ContextType, BDeviceType, BDtypeType, BRNGType]] = None

observation_space class-attribute instance-attribute

observation_space: Optional[Space[ObsType, BDeviceType, BDtypeType, BRNGType]] = None

action_space class-attribute instance-attribute

action_space: Optional[Space[ActType, BDeviceType, BDtypeType, BRNGType]] = None

has_reward instance-attribute

has_reward = any((has_reward) for node in nodes)

has_termination_signal instance-attribute

has_termination_signal = any((has_termination_signal) for node in nodes)

has_truncation_signal instance-attribute

has_truncation_signal = any((has_truncation_signal) for node in nodes)

supported_render_modes class-attribute instance-attribute

supported_render_modes = ('dict', 'auto')

render_mode instance-attribute

render_mode = render_mode

world property

world: World[BArrayType, BDeviceType, BDtypeType, BRNGType]

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: Optional[float]

unwrapped property

unwrapped: WorldNode

prev_wrapper_layer property

prev_wrapper_layer: Optional[WorldNode]

nodes instance-attribute

nodes = nodes

direct_return instance-attribute

direct_return = direct_return

get_context

get_context() -> Optional[CombinedDataT]

Get context by flattening all node contexts into one dictionary.

get_observation

get_observation() -> CombinedDataT

Get observation by flattening all node observations into one dictionary.

get_info

get_info() -> Optional[Dict[str, Any]]

Get info by merging all node info dictionaries.

set_next_action

set_next_action(action: CombinedDataT) -> None

Set action by routing keys to appropriate nodes.

pre_environment_step

pre_environment_step(dt, *, priority: int = 0)

Dispatch pre-step callbacks to child nodes at the matching frequency.

get_reward

get_reward()

get_termination

get_termination()

get_truncation

get_truncation()

render

render()

post_environment_step

post_environment_step(dt, *, priority: int = 0)

Dispatch post-step callbacks to child nodes at the matching frequency.

reset

reset(*, priority: int = 0, seed=None, mask=None, pernode_kwargs: Dict[str, Any] = {})

Forward reset calls to children that participate at priority.

reload

reload(*, priority: int = 0, seed=None, mask=None, pernode_kwargs: Dict[str, Any] = {})

Forward reload calls to children that participate at priority.

after_reset

after_reset(*, priority: int = 0, mask=None)

Forward post-reset hooks and reset internal routing counters.

after_reload

after_reload(*, priority: int = 0, mask=None)

Call after_reload on child nodes. Similar to after_reset but for reload flow.

get_node

get_node(nested_keys: Union[str, Sequence[str]]) -> Optional[WorldNode]

Resolve a child node by dotted-path-like key traversal.

get_nodes_by_fn

get_nodes_by_fn(fn: Callable[[WorldNode], bool]) -> list[WorldNode]

Return every node in the subtree that satisfies fn.

get_nodes_by_type

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

Return nodes in this subtree that are instances of node_type.

close

close()

Close every child node.

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.