pytupli.dataset.TupliDataset

class TupliDataset(storage: TupliStorage)

Bases: object

A dataset class for downloading, managing and filtering offline RL tuple data.

This class provides functionality to load, filter, and process reinforcement learning data including benchmarks, episodes, and tuples. It supports various filtering operations and provides methods for batch processing and data conversion.

Parameters:

storage (TupliStorage) – The storage backend to fetch data from.

Methods

as_batch_generator	Returns a generator that yields batches of tuples from the dataset.
convert_to_numpy	Converts the dataset tuples into numpy arrays.
load	Loads all episode data including tuples and applies any filters.
preview	Returns a preview of the episodes without loading the full tuple data.
sample_episodes	Randomly samples episodes from the dataset.
set_seed	Sets the random seed for reproducibility.
with_benchmark_filter	Creates a new dataset with an additional benchmark filter.
with_episode_filter	Creates a new dataset with an additional episode filter.
with_tuple_filter	Creates a new dataset with an additional tuple filter function.

_fetch_episodes(with_tuples: bool = False) → None

Fetches episodes from storage based on current filters.

This internal method refreshes the episodes list based on the current benchmark and episode filters. It can optionally include the tuple data for each episode.

Parameters:

with_tuples (bool) – If True, includes tuple data in the fetched episodes.

as_batch_generator(batch_size: int, shuffle: bool = False) → Generator[List[RLTuple], None, None]

Returns a generator that yields batches of tuples from the dataset.

Parameters:

• batch_size (int) – The size of each batch.

• shuffle (bool) – Whether to shuffle the tuples before creating batches.

Yields:

List[RLTuple] – Batches of tuples of the specified size.

convert_to_numpy() → tuple[ndarray, ndarray, ndarray, ndarray, ndarray]

Converts the dataset tuples into numpy arrays.

Returns:

tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray] –

A tuple containing:

• observations: Array of state observations

• actions: Array of actions

• rewards: Array of rewards

• terminals: Array of terminal flags

• timeouts: Array of timeout flags

load() → None

Loads all episode data including tuples and applies any filters.

This method fetches all episode data and their associated tuples, then applies any tuple filters that have been set.

preview() → list[EpisodeHeader]

Returns a preview of the episodes without loading the full tuple data.

Returns:

list[EpisodeHeader] – A list of episode headers matching the current filters.

sample_episodes(n_samples: int) → list[EpisodeItem]

Randomly samples episodes from the dataset.

Parameters:

n_samples (int) – The number of episodes to sample.

Returns:

list[EpisodeItem] – A list of randomly sampled episodes.

set_seed(seed: int) → None

Sets the random seed for reproducibility.

Parameters:

seed (int) – The random seed to set.

with_benchmark_filter(filter: BaseFilter) → TupliDataset

Creates a new dataset with an additional benchmark filter.

Parameters:

filter (BaseFilter) – The filter to apply to benchmarks.

Returns:

TupliDataset – A new dataset instance with the applied filter.

with_episode_filter(filter: BaseFilter) → TupliDataset

Creates a new dataset with an additional episode filter.

Parameters:

filter (BaseFilter) – The filter to apply to episodes.

Returns:

TupliDataset – A new dataset instance with the applied filter.

with_tuple_filter(filter_fcn: Callable) → TupliDataset

Creates a new dataset with an additional tuple filter function.

Parameters:

filter_fcn (Callable) – A function that takes a tuple and returns a boolean.

Returns:

TupliDataset – A new dataset instance with the applied filter.