unienv_interface.env_base.vec_env

SyncVecEnv

SyncVecEnv(env_fn: Iterable[Callable[[], Env[BArrayType, ContextType, ObsType, ActType, RenderFrame, BDeviceType, BDtypeType, BRNGType]]], seed: Optional[int] = None)

Bases: Env[BArrayType, ContextType, ObsType, ActType, RenderFrame, BDeviceType, BDtypeType, BRNGType]

Run multiple non-batched environments synchronously in one process.

Instantiate every environment immediately and batch their interfaces.

envs instance-attribute

envs = [(fn()) for fn in env_fn]

action_space instance-attribute

action_space = batch_differing_spaces([(action_space) for env in (envs)], device=device)

observation_space instance-attribute

observation_space = batch_differing_spaces([(observation_space) for env in (envs)], device=device)

context_space instance-attribute

context_space = None if context_space is None else batch_differing_spaces([(context_space) for env in (envs)], device=device)

rng instance-attribute

rng = random_number_generator(seed, device=device)

metadata property

metadata: Dict[str, Any]

render_mode property

render_mode: Optional[str]

render_fps property

render_fps: Optional[int]

backend property

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

device property

device: Optional[BDeviceType]

batch_size property

batch_size: Optional[int]

unwrapped property

unwrapped: Env

prev_wrapper_layer property

prev_wrapper_layer: Optional[Env]

reset

reset(*args, mask: Optional[BArrayType] = None, seed: Optional[int] = None, **kwargs) -> Tuple[ContextType, ObsType, Dict[str, Any]]

Reset all selected workers and concatenate their results.

step

step(action: ActType) -> Tuple[ObsType, BArrayType, BArrayType, BArrayType, Dict[str, Any]]

Step every worker environment once and concatenate the outputs.

render

render() -> Sequence[RenderFrame] | None

Collect render outputs from every worker that returns a frame.

close

close()

sample_space

sample_space(space: Space) -> Any

Sample from space using and updating self.rng.

sample_action

sample_action() -> ActType

Sample one action from action_space.

sample_observation

sample_observation() -> ObsType

Sample one observation from observation_space.

sample_context

sample_context() -> Optional[ContextType]

Sample one context value if context_space is defined.

update_observation_post_reset

update_observation_post_reset(old_obs: ObsType, newobs_masked: ObsType, mask: BArrayType) -> ObsType

Merge masked reset observations back into a full batched observation.

update_context_post_reset

update_context_post_reset(old_context: ContextType, new_context: ContextType, mask: BArrayType) -> ContextType

Merge masked reset contexts back into a full batched context.

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.

AsyncVecEnv

AsyncVecEnv(env_fn: Iterable[Callable[[], Env[BArrayType, ContextType, ObsType, ActType, RenderFrame, BDeviceType, BDtypeType, BRNGType]]], seed: Optional[int] = None, ctx: Optional[BaseContext] = None, daemon: bool = True)

Bases: Env[BArrayType, ContextType, ObsType, ActType, RenderFrame, BDeviceType, BDtypeType, BRNGType]

Run multiple non-batched environments in separate worker processes.

Spawn worker processes and batch their exposed spaces.

command_pipes instance-attribute

command_pipes: List[Connection] = []

processes instance-attribute

processes: List[Process] = []

error_queue instance-attribute

error_queue: Queue = Queue()

backend instance-attribute

backend = backend

device instance-attribute

device = device

metadata instance-attribute

metadata = metadata

render_mode instance-attribute

render_mode = render_mode

render_fps instance-attribute

render_fps = render_fps

action_space instance-attribute

action_space = batch_space(action_space, len(processes))

observation_space instance-attribute

observation_space = batch_space(observation_space, len(processes))

context_space instance-attribute

context_space = None if context_space is None else batch_space(context_space, len(processes))

rng instance-attribute

rng = random_number_generator(seed, device=device)

batch_size property

batch_size: Optional[int]

unwrapped property

unwrapped: Env

prev_wrapper_layer property

prev_wrapper_layer: Optional[Env]

send_command

send_command(index: int, cmd: Literal['reset', 'step', 'render', 'close'], *args, **kwargs)

Send a command to one worker process.

get_command_result

get_command_result(index: int)

Receive one worker result and surface worker-side exceptions.

reset

reset(*args, mask: Optional[BArrayType] = None, seed: Optional[int] = None, **kwargs) -> Tuple[ContextType, ObsType, Dict[str, Any]]

Convenience wrapper around reset_async plus reset_wait.

reset_async

reset_async(*args, mask: Optional[BArrayType] = None, seed: Optional[int] = None, **kwargs) -> None

Dispatch reset commands to all selected workers.

reset_wait

reset_wait() -> Tuple[ContextType, ObsType, Dict[str, Any]]

Collect the pending reset results and batch them.

step

step(action: ActType) -> Tuple[ObsType, BArrayType, BArrayType, BArrayType, Dict[str, Any]]

Convenience wrapper around step_async plus step_wait.

step_async

step_async(action: ActType) -> None

Dispatch one action batch to all worker processes.

step_wait

step_wait() -> Tuple[ObsType, BArrayType, BArrayType, BArrayType, Dict[str, Any]]

Collect the pending step results and batch them.

close

close()

Shut down all worker processes and communication pipes.

render abstractmethod

render() -> RenderFrame | Sequence[RenderFrame] | None

Render the current environment state using the configured render mode.

sample_space

sample_space(space: Space) -> Any

Sample from space using and updating self.rng.

sample_action

sample_action() -> ActType

Sample one action from action_space.

sample_observation

sample_observation() -> ObsType

Sample one observation from observation_space.

sample_context

sample_context() -> Optional[ContextType]

Sample one context value if context_space is defined.

update_observation_post_reset

update_observation_post_reset(old_obs: ObsType, newobs_masked: ObsType, mask: BArrayType) -> ObsType

Merge masked reset observations back into a full batched observation.

update_context_post_reset

update_context_post_reset(old_context: ContextType, new_context: ContextType, mask: BArrayType) -> ContextType

Merge masked reset contexts back into a full batched context.

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.

data_stack

data_stack(data: Any, backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], device: Optional[BDeviceType] = None)

Recursively stack Python/array data into a batched backend representation.