unienv_interface.world.funcnodes.combined_funcnode

CombinedDataT module-attribute

CombinedDataT = Union[Dict[str, Any], Any]

CombinedNodeStateT module-attribute

CombinedNodeStateT = Dict[str, Any]

ROOT_NODE_KEY module-attribute

ROOT_NODE_KEY = ''

CombinedFuncWorldNode

CombinedFuncWorldNode(name: str, nodes: Iterable[FuncWorldNode[WorldStateT, Any, Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]], direct_return: bool = True, render_mode: Optional[str] = 'auto', root_node: Optional[FuncWorldNode[WorldStateT, Any, Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]] = None)

Bases: FuncWorldNode[WorldStateT, CombinedNodeStateT, Optional[CombinedDataT], CombinedDataT, CombinedDataT, BArrayType, BDeviceType, BDtypeType, BRNGType], Generic[WorldStateT, BArrayType, BDeviceType, BDtypeType, BRNGType]

A functional counterpart to CombinedWorldNode that composes multiple FuncWorldNodes.

It aggregates spaces (context, observation, action) and runtime data (context, observation, info, reward, termination, truncation) across child nodes. If only one child exposes a given interface and direct_return=True, the value is passed through directly.

Root node

An optional root_node may be passed next to nodes. It is a regular FuncWorldNode (sharing the same FuncWorld) 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 -- and its node state is stored in the combined node_state under the reserved key :data:ROOT_NODE_KEY (""), a string key that cannot collide with a child node name (child names are validated non-empty) and therefore keeps persisted CombinedNodeStateT dictionaries loadable without migration.

• Spaces / data (context, observation, action -- independently). Spaces are aggregated once at construction (functional nodes declare static spaces). When the root node is the sole contributor of a channel, the existing direct_return rule applies verbatim: direct_return=True passes the root's own space/data through unwrapped (any Space type, e.g. a plain BoxSpace), while direct_return=False wraps 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 a DictSpace for 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 raises ValueError.
• 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 cache entries (exactly as before) and every remaining key accumulates into the root's entry (stored under the reserved key) 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_node never resolves the reserved key (it is not a user-facing path); get_nodes_by_fn does traverse the root node.

supported_render_modes class-attribute instance-attribute

supported_render_modes = ('dict', 'auto')

nodes instance-attribute

nodes = nodes

root_node instance-attribute

root_node = root_node

name instance-attribute

name = name

direct_return instance-attribute

direct_return = direct_return

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)

render_mode instance-attribute

render_mode = renderable_nodes[0].render_mode

world property

world

control_timestep property

control_timestep

update_timestep property

update_timestep

effective_update_timestep property

effective_update_timestep

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]

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

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.

unwrapped property

unwrapped: FuncWorldNode

prev_wrapper_layer property

prev_wrapper_layer: Optional[FuncWorldNode]

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.

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

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]

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_context

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

get_observation

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

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]

get_info

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

render

render(world_state, node_state)

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.