replaybuffer package

Submodules

replaybuffer.replaybuffer module

class replaybuffer.replaybuffer.ReplayBuffer(max_size: int = inf)

Bases: object

get_values(buffer: str, idx: Union[int, Iterable[int], Generator[int, Any, Any]]) → Union[object, Iterable[object]]

Get buffer value(s) at some index or indices.

Args:

buffer (str): The name of the buffer to get value(s) from. idx (Union[int, Iterable[int], Generator[int, Any, Any]]): The index or indices to get values from.

Returns:

Union[object, Iterable[object]]: The value(s) at the given index or indices.

initialize_buffer(buffer_name: str) → None

Initialize a new buffer in memory.

Args:

buffer_name (str): The name of the buffer.

max_size: int = inf

previous(n: int) → dict

Get the most recent entries to the buffer.

Args:

n (int): Number of previous entries to look back.

Returns:

dict: Buffer name to iterable of previous values key, value pairs

for each buffer.

reset() → None

Clears all buffer values.

sample(n: int) → dict

Random uniform sample over all buffers.

Args:

n (int): The size of the sample.

Returns:

dict: Buffer name to iterable of sample values key, value pairs

for each buffer.

set_values(buffer: str, idx: Union[int, Iterable[int], Generator[int, Any, Any]], val: Union[object, Iterable[object], Generator[object, Any, Any]]) → None

Set explicit buffer value(s) at some index or indices.

Args:

buffer (str): The name of the buffer to modify. idx (Union[int, Iterable[int], Generator[int, Any, Any]]): The index or indices to modify. val (Union[object, Iterable[object], Generator[object]]): The value(s) to set.

store(**kwargs: object) → None

Store items in each buffer.

Example:

buffer = ReplayBuffer()
buffer.initialize_buffer("chain1")
buffer.initialize_buffer("chain2")

for i in range(10):
    buffer.store(
        chain1 = i,
        chain2 = 2*i
    )

take(*args, **kwargs) → dict

Wrapper around Numpy’s take function. Used to get a range of values, and only

necessary when wrapping splicing from the end of the buffer back to the beginning, or vice versa. Otherwise, __getitem__ is preferable.

Returns:

dict: Buffer name to iterable of taken values key, value pairs

for each buffer.

Example:

buffer = ReplayBuffer()
buffer.initialize_buffer("chain1")
buffer.initialize_buffer("chain2")

for i in range(10):
    buffer.store(
        chain1 = i,
        chain2 = 2*i
    )

buffer.take(range(0, 5))

replaybuffer.utils module

replaybuffer.utils.remove_nones(*arrays: Iterable) → Tuple[Iterable]

Take inputted arrays that may contain None values, and return copies without Nones.

Returns:

tuple[Iterable]: New arrays with only non-None values

Module contents