unienv_data.replay_buffer.trajectory_replay_buffer¶
Bases: TypedDict, Generic[BatchT, EpisodeBatchT]
Structured sample returned by TrajectoryReplayBuffer.
TrajectoryReplayBuffer(step_data_buffer: ReplayBuffer[BatchT, BArrayType, BDeviceType, BDtypeType, BRNGType], step_episode_id_buffer: ReplayBuffer[BArrayType, BArrayType, BDeviceType, BDtypeType, BRNGType], episode_data_buffer: Optional[ReplayBuffer[EpisodeBatchT, BArrayType, BDeviceType, BDtypeType, BRNGType]], current_episode_id: int = 0, episode_id_to_index_map: Optional[Dict[int, int]] = None, multiprocessing: bool = False)
Bases: BatchBase[TrajectoryData[BatchT, EpisodeBatchT], BArrayType, BDeviceType, BDtypeType, BRNGType], Generic[BatchT, EpisodeBatchT, BArrayType, BDeviceType, BDtypeType, BRNGType]
Replay buffer that tracks both step-level and episode-level data.
Each transition stores step data plus an episode id, and can optionally reference separate episode metadata for trajectory-aware sampling.
Assemble the step buffer, episode id buffer, and optional episode buffer.
instance-attribute
¶
episode_id_to_index_map = dict(episode_id_to_index_map) if episode_id_to_index_map is not None else None
staticmethod
¶
create(step_data_instance_space: Space[BatchT, BDeviceType, BDtypeType, BRNGType], step_data_storage_cls: Type[SpaceStorage[BatchT, BArrayType, BDeviceType, BDtypeType, BRNGType]], *, step_data_capacity: Optional[int] = None, step_episode_id_storage_cls: Optional[Type[SpaceStorage[BArrayType, BArrayType, BDeviceType, BDtypeType, BRNGType]]] = None, step_episode_id_capacity: Optional[int] = None, step_episode_id_storage_kwargs: Dict[str, Any] = {}, episode_data_instance_space: Optional[Space[EpisodeBatchT, BDeviceType, BDtypeType, BRNGType]] = None, episode_data_storage_cls: Optional[Type[SpaceStorage[EpisodeBatchT, BArrayType, BDeviceType, BDtypeType, BRNGType]]] = None, episode_data_capacity: Optional[int] = None, episode_data_storage_kwargs: Dict[str, Any] = {}, cache_path: Optional[Union[str, PathLike]] = None, multiprocessing: bool = False, **kwargs) -> TrajectoryReplayBuffer[BatchT, EpisodeBatchT, BArrayType, BDeviceType, BDtypeType, BRNGType]
Create a new trajectory replay buffer and all subordinate buffers.
staticmethod
¶
load_from(path: Union[str, PathLike], *, backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], device: Optional[BDeviceType] = None, read_only: bool = False, multiprocessing: bool = False, step_storage_kwargs: Dict[str, Any] = {}, step_episode_id_storage_kwargs: Dict[str, Any] = {}, episode_storage_kwargs: Dict[str, Any] = {}, **storage_kwargs) -> TrajectoryReplayBuffer[BatchT, EpisodeBatchT, BArrayType, BDeviceType, BDtypeType, BRNGType]
Load a trajectory replay buffer and its subordinate buffers from disk.
dumps(path: Union[str, PathLike])
Persist buffer metadata plus all subordinate replay buffers.
episode_id_to_episode_data_index(episode_id: Union[int, BArrayType]) -> Union[int, BArrayType]
append_flattened(value: BArrayType) -> None
Append one flattened sample to the batch.
remove_at(idx: Union[IndexableType, BArrayType]) -> None
Remove one or more samples from the batch.
extend_from(other: BatchBase[BatchT, BArrayType, BDeviceType, BDtypeType, BRNGType], chunk_size: int = 8, tqdm: bool = False) -> None
Copy data from another batch in bounded-size chunks.
get_slice(idx: Union[IndexableType, BArrayType]) -> BatchBase[BatchT, BArrayType, BDeviceType, BDtypeType, BRNGType]
Create a lazy view over a subset of indices.
get_column(nested_keys: Sequence[str]) -> BatchBase[Any, BArrayType, BDeviceType, BDtypeType, BRNGType]
Create a lazy view over a nested field inside each sample.