unienv_interface.world¶
CombinedWorldNode(name: str, nodes: Iterable[WorldNode[Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]], direct_return: bool = True, render_mode: Optional[str] = 'auto')
Bases: WorldNode[Optional[CombinedDataT], CombinedDataT, CombinedDataT, BArrayType, BDeviceType, BDtypeType, BRNGType], Generic[BArrayType, BDeviceType, BDtypeType, BRNGType]
A WorldNode that combines multiple WorldNodes into one node, using a dictionary to store the data from each node.
The observation, reward, termination, truncation, and info are combined from all child nodes.
The keys in the dictionary are the names of the child nodes.
If there is only one child node that supports value and direct_return is set to True, the value is returned directly instead of a dictionary.
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
property
¶
backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType]
Backend provided by the attached world.
class-attribute
instance-attribute
¶
supported_render_modes = ('dict', 'auto')
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)
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_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.
pre_environment_step(dt, *, priority: int = 0)
Dispatch pre-step callbacks to child nodes at the matching frequency.
set_next_action(action)
Route combined actions to child nodes, respecting per-node control rates.
post_environment_step(dt, *, priority: int = 0)
Dispatch post-step callbacks to child nodes at the matching frequency.
reset(*, priority: int = 0, seed=None, mask=None, pernode_kwargs: Dict[str, Any] = {})
Forward reset calls to children that participate at priority.
reload(*, priority: int = 0, seed=None, mask=None, pernode_kwargs: Dict[str, Any] = {})
Forward reload calls to children 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 child nodes. Similar to after_reset but for reload flow.
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 |
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((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.
pre_environment_step(dt, *, priority: int = 0)
Dispatch pre-step callbacks to child nodes at the matching frequency.
post_environment_step(dt, *, priority: int = 0)
Dispatch post-step callbacks to child nodes at the matching frequency.
reset(*, priority: int = 0, seed=None, mask=None, pernode_kwargs: Dict[str, Any] = {})
Forward reset calls to children that participate at priority.
reload(*, priority: int = 0, seed=None, mask=None, pernode_kwargs: Dict[str, Any] = {})
Forward reload calls to children 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 child nodes. Similar to after_reset but for 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.
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_context() -> Optional[CombinedDataT]
Get context by flattening all node contexts into one dictionary.
get_observation() -> CombinedDataT
Get observation by flattening all node observations into one dictionary.
set_next_action(action: CombinedDataT) -> None
Set action by routing keys to appropriate nodes.
CombinedFuncWorldNode(name: str, nodes: Iterable[FuncWorldNode[WorldStateT, Any, Any, Any, Any, BArrayType, BDeviceType, BDtypeType, BRNGType]], direct_return: bool = True, render_mode: Optional[str] = 'auto')
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.
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
property
¶
backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType]
Backend provided by the attached world.
class-attribute
instance-attribute
¶
supported_render_modes = ('dict', 'auto')
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)
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.
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.
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]
set_next_action(world_state: WorldStateT, node_state: CombinedNodeStateT, action: CombinedDataT) -> Tuple[WorldStateT, CombinedNodeStateT]
post_environment_step(world_state: WorldStateT, node_state: CombinedNodeStateT, dt: Union[float, BArrayType], *, priority: int = 0) -> Tuple[WorldStateT, CombinedNodeStateT]
get_context(world_state: WorldStateT, node_state: CombinedNodeStateT) -> CombinedDataT
get_observation(world_state: WorldStateT, node_state: CombinedNodeStateT) -> CombinedDataT
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]
get_info(world_state: WorldStateT, node_state: CombinedNodeStateT) -> Optional[Dict[str, Any]]
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
¶
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.
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)
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.
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.