unienv_interface.world.nodes.env_node¶
WorldNode adapter that exposes an unbatched :class:Env as a node.
EnvAsWorldNode(env: Env, name: str, *, world: Optional[World] = None, control_timestep: Optional[float] = None, update_timestep: Optional[float] = None, hold_last_action: bool = False, expose_reward: bool = True, expose_termination: bool = True, expose_truncation: bool = True, forward_info: bool = True, reset_seed: Optional[int] = None, pre_step_priority: int = 0)
Bases: WorldNode[ContextType, ObsType, ActType, BArrayType, BDeviceType, BDtypeType, BRNGType]
Adapt an existing unbatched :class:Env so it can be composed as a WorldNode.
Design¶
Env.step(action) is atomic: it produces the next observation, reward,
termination/truncation signals, and info in a single call. A WorldNode
instead splits the step lifecycle into::
set_next_action(action)
-> pre_environment_step(dt, priority=...)
-> World.step()
-> post_environment_step(dt, priority=...)
-> get_observation() / get_reward() / ... / get_info()
This adapter bridges the mismatch by performing the atomic inner
Env.step inside pre_environment_step (using the action cached by the
preceding set_next_action call) and caching its five-tuple result for
the get_* accessors to expose. reset maps directly to
Env.reset, and the base WorldNode.reload default (which delegates to
reset) is exactly right for a pre-built environment, hence the
non-empty reset_priorities / reload_priorities sets.
The dt argument of pre_environment_step is intentionally ignored:
the wrapped environment is self-paced (e.g. real hardware running at its own
control rate), while the outer world is expected to be a bookkeeping-only
world such as :class:RealWorld. control_timestep / update_timestep
can still be set for step-ratio bookkeeping by the composer.
No schema transformation / re-keying is performed: the inner environment's action, observation, and context spaces (and the raw observation / context layouts) pass through unchanged.
Only unbatched environments (batch_size is None) are supported. The
optional world is only used for backend/device delegation; when provided
its backend type must match the inner environment's backend, since the node
serves the inner environment's arrays.
class-attribute
instance-attribute
¶
post_environment_step_priorities = set()
instance-attribute
¶
update_timestep = update_timestep if update_timestep is not None else control_timestep
instance-attribute
¶
supported_render_modes = tuple(env.metadata.get('render_modes', ()))
class-attribute
instance-attribute
¶
pre_environment_step_priorities = {pre_step_priority}
property
¶
backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType]
Backend of the attached world, falling back to the inner env's backend.
property
¶
device: Optional[BDeviceType]
Device of the attached world, falling back to the inner env's device.
property
¶
effective_update_timestep: Optional[float]
Resolved update period, defaulting to control_timestep when unset.
set_next_action(action: ActType) -> None
Cache the action consumed by the next pre_environment_step call.
action=None is only accepted when hold_last_action=True (the
previously cached action is kept); in that case a node that has never
received an action still raises from pre_environment_step.
pre_environment_step(dt: Union[float, BArrayType], *, priority: int = 0) -> None
Run one atomic Env.step with the cached action.
The world-provided dt is ignored: the wrapped environment is
self-paced, so it advances by exactly one of its own steps per call.
reset(*, priority: int = 0, seed: Optional[int] = None, mask: Optional[BArrayType] = None, **kwargs) -> None
Reset the inner environment (also used as the default reload).
get_context() -> Optional[ContextType]
Return the context produced by the most recent inner reset.
get_observation() -> ObsType
Return the latest step observation, or the reset observation before the first step.
get_reward() -> Union[float, BArrayType]
Return the cached step reward (0.0 before the first step).
get_termination() -> Union[bool, BArrayType]
Return the cached termination signal (False before the first step).
get_truncation() -> Union[bool, BArrayType]
Return the cached truncation signal (False before the first step).
get_info() -> Optional[Dict[str, Any]]
Return a copy of the latest info dict when forward_info is enabled.
post_environment_step(dt: Union[float, BArrayType], *, priority: int = 0) -> None
This method is called after the environment step to update the sensor's internal state. Args: dt (float/BArrayType): The time elapsed between the last world step and the current step.
reload(*, priority: int = 0, seed: Optional[int] = None, mask: Optional[BArrayType] = None, **kwargs) -> None
Reload the node. By default, this just calls reset with the same parameters.
Simulation environments can override this to completely re-read assets and reload the node.
after_reset(*, priority: int = 0, mask: Optional[BArrayType] = None) -> None
This method is called after all WorldNodes has been called with reset (e.g. the environment reset is effectively done).
Use get_context, get_observation, and get_info to read the post-reset state.
after_reload(*, priority: int = 0, mask: Optional[BArrayType] = None) -> None
This method is called after the world has been rebuilt following a reload.
At this point: - All nodes have added their entities during the reload phase - The world scene has been built (simulation is ready) - Nodes can now cache references to entities, geoms, etc.
By default, this calls after_reset(). Override if you need specific
post-reload initialization that differs from post-reset.
Use get_context, get_observation, and get_info to read the state.
get_node(nested_keys: Union[str, Sequence[str]]) -> Optional[WorldNode]
Return a node by key path.
strkeys are treated as a single direct-child key.- Empty key sequence returns
self. - Vanilla nodes have no children and return
Nonefor non-empty lookup.
get_nodes_by_fn(fn: Callable[[WorldNode], bool]) -> list[WorldNode]
Return nodes in this subtree matching fn.
Vanilla nodes only evaluate self.
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.