pytupli.dataset.TupliDataset

class TupliDataset(storage: TupliStorage, info: dict[str, Any] | None = None)

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.

• info (dict[str, Any] | None) – Optional metadata or additional information about the dataset.

Methods

as_batch_generator	Returns a generator that yields batches of tuples from the dataset.
convert_to_d4rl_format	Converts the dataset tuples into the format used by D4RL.
convert_to_dataframe	Converts the dataset tuples into a pandas DataFrame.
convert_to_tensors	Converts the dataset tuples into tensors of the format specified by handing over the respective parser.
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.

Attributes

actions	Returns a list of actions from all tuples in the dataset.
infos	Returns a list of info dictionaries from all tuples in the dataset.
observations	Returns a list of observations from all tuples in the dataset.
rewards	Returns a list of rewards from all tuples in the dataset.
terminals	Returns a list of terminal flags from all tuples in the dataset.
timeouts	Returns a list of timeout flags from all tuples in the dataset.

_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.

property actions: list[list[float]]

Returns a list of actions from all tuples in the dataset.

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_d4rl_format() → dict[str, ndarray]

Converts the dataset tuples into the format used by D4RL.

Returns:

dict[str, np.ndarray] –

A dictionary containing:

• observations: Array of state observations

• actions: Array of actions

• next_observations: Array of next state observations

• rewards: Array of rewards

• terminals: Array of terminal flags

convert_to_dataframe()

Converts the dataset tuples into a pandas DataFrame.

Returns:

pd.DataFrame –

A DataFrame containing:

• observations: State observations

• actions: Actions taken

• next_observations: Next state observations

• rewards: Rewards received

• terminals: Terminal flags

Raises:

ImportError – If pandas is not installed.

convert_to_tensors(parser: type[~pytupli.dataset.BaseTupleParser] = <class 'pytupli.dataset.NumpyTupleParser'>) → tuple

Converts the dataset tuples into tensors of the format specified by handing over the respective parser.

Parameters:

parser (BaseTupleParser) – The parser to use for converting tuples.

Returns:

tuple –

A tuple containing tensors in the specified format:

• observations: Tensor/Array of state observations

• actions: Tensor/Array of actions

• rewards: Tensor/Array of rewards

• terminals: Tensor/Array of terminal flags

• timeouts: Tensor/Array of timeout flags

Raises:

• ValueError – If an unsupported format is specified.

• ImportError – If the required library for the format is not installed.

property infos: list[dict[str, Any]]

Returns a list of info dictionaries from all tuples in the dataset.

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.

property observations: list[list[float]]

Returns a list of observations from all tuples in the dataset.

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.

property rewards: list[float]

Returns a list of rewards from all tuples in the dataset.

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.

property terminals: list[bool]

Returns a list of terminal flags from all tuples in the dataset.

property timeouts: list[bool]

Returns a list of timeout flags from all tuples in the dataset.

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.