Data

pamiq_core.data.DataBuffer

DataBuffer(max_queue_size: int | None = None)

Bases: ABC, PersistentStateMixin

Interface for managing experience data collected during system execution.

DataBuffer provides an interface for collecting and managing experience data generated during system execution. It maintains a buffer of fixed maximum size.

Type Parameters

• T: The type of individual data elements.
• R: The return type of the get_data() method.

Initializes the DataBuffer.

PARAMETER	DESCRIPTION
max_queue_size	Maximum number of samples to store in the collector's queue. When the queue size exceeds this limit, old data will be deleted. If None, the queue will have unlimited size (may cause memory issues). TYPE: int | None DEFAULT: None

RAISES	DESCRIPTION
ValueError	If max_queue_size is negative.

Source code in src/pamiq_core/data/buffer.py

def __init__(self, max_queue_size: int | None = None) -> None:
    """Initializes the DataBuffer.

    Args:
        max_queue_size: Maximum number of samples to store in the collector's queue.
            When the queue size exceeds this limit, old data will be deleted.
            If None, the queue will have unlimited size (may cause memory issues).

    Raises:
        ValueError: If max_queue_size is negative.
    """
    super().__init__()
    if max_queue_size is not None:
        if max_queue_size < 0:
            raise ValueError("max_queue_size must be non-negative")
    else:
        warnings.warn(
            "max_queue_size is None. The collector's queue will have unlimited size, "
            "which may cause memory issues if data collection is faster than processing.",
            RuntimeWarning,
            stacklevel=2,
        )
    self._max_queue_size = max_queue_size

max_queue_size property

max_queue_size: int | None

Returns the maximum number of samples that can be stored in the collector's queue.

add abstractmethod

add(data: T) -> None

Adds a new data sample to the buffer.

PARAMETER	DESCRIPTION
data	Data element to add to the buffer. TYPE: T

Source code in src/pamiq_core/data/buffer.py

@abstractmethod
def add(self, data: T) -> None:
    """Adds a new data sample to the buffer.

    Args:
        data: Data element to add to the buffer.
    """
    pass

get_data abstractmethod

get_data() -> R

Retrieves all stored data from the buffer.

RETURNS	DESCRIPTION
R	Data structure containing all stored samples.

Source code in src/pamiq_core/data/buffer.py

@abstractmethod
def get_data(self) -> R:
    """Retrieves all stored data from the buffer.

    Returns:
        Data structure containing all stored samples.
    """
    pass

__len__ abstractmethod

__len__() -> int

Returns the current number of samples in the buffer.

RETURNS	DESCRIPTION
int	The number of samples currently stored in the buffer. TYPE: int

Source code in src/pamiq_core/data/buffer.py

@abstractmethod
def __len__(self) -> int:
    """Returns the current number of samples in the buffer.

    Returns:
        int: The number of samples currently stored in the buffer.
    """
    pass

pamiq_core.data.DataCollector

DataCollector(max_queue_size: int | None)

A thread-safe collector for buffered data.

This class provides concurrent data collection capabilities with thread safety, working in conjunction with DataUser to manage data collection and transfer.

Type Parameters

T: The type of individual data elements.

Initialize DataCollector with a specified queue size.

PARAMETER	DESCRIPTION
max_queue_size	Maximum size of the collection queue. If None, queue has no size limit. TYPE: int | None

Source code in src/pamiq_core/data/interface.py

def __init__(self, max_queue_size: int | None) -> None:
    """Initialize DataCollector with a specified queue size.

    Args:
        max_queue_size: Maximum size of the collection queue. If None, queue has no size limit.
    """
    self._max_queue_size = max_queue_size
    self._queue = TimestampingQueue[T](max_queue_size)
    self._lock = RLock()

collect

collect(data: T) -> None

Collect data in a thread-safe manner.

PARAMETER	DESCRIPTION
data	Data to be collected. TYPE: T

Source code in src/pamiq_core/data/interface.py

def collect(self, data: T) -> None:
    """Collect data in a thread-safe manner.

    Args:
        data: Data to be collected.
    """
    with self._lock:
        self._queue.append(data)

pamiq_core.data.DataUser

DataUser(buffer: DataBuffer[Any, R])

Bases: PersistentStateMixin

A class that manages data buffering and timestamps for collected data.

This class acts as a user of data buffers, handling the collection, storage, and retrieval of data along with their timestamps. It works in conjunction with a DataCollector to manage concurrent data collection.

Type Parameters

R: The return type of the buffer's get_data() method.

Initialize DataUser with a specified buffer.

PARAMETER	DESCRIPTION
buffer	Data buffer instance to store collected data. TYPE: DataBuffer[Any, R]

Source code in src/pamiq_core/data/interface.py

def __init__(self, buffer: DataBuffer[Any, R]) -> None:
    """Initialize DataUser with a specified buffer.

    Args:
        buffer: Data buffer instance to store collected data.
    """
    self._buffer = buffer
    self._timestamps: deque[float] = deque(maxlen=buffer.max_queue_size)
    # DataCollector instance is only accessed from DataUser and Container classes
    self._collector = DataCollector[Any](buffer.max_queue_size)

update

update() -> None

Update buffer with collected data from the collector.

Moves all collected data from the collector to the buffer and records their timestamps.

Source code in src/pamiq_core/data/interface.py

def update(self) -> None:
    """Update buffer with collected data from the collector.

    Moves all collected data from the collector to the buffer and
    records their timestamps.
    """
    queue = self._collector._move_data()  # pyright: ignore[reportPrivateUsage]
    for _ in range(len(queue)):
        data, t = queue.popleft()
        self._buffer.add(data)
        self._timestamps.append(t)

get_data

get_data() -> R

Update and retrieve data from the buffer.

RETURNS	DESCRIPTION
R	Current data stored in the buffer.

Source code in src/pamiq_core/data/interface.py

def get_data(self) -> R:
    """Update and retrieve data from the buffer.

    Returns:
        Current data stored in the buffer.
    """
    self.update()
    return self._buffer.get_data()

count_data_added_since

count_data_added_since(timestamp: float) -> int

Count the number of data points added after the specified timestamp.

NOTE: Use pamiq_core.time to retrieve timestamp.

PARAMETER	DESCRIPTION
timestamp	Reference timestamp to count from. TYPE: float

RETURNS	DESCRIPTION
int	Number of data points added after the specified timestamp.

Source code in src/pamiq_core/data/interface.py

def count_data_added_since(self, timestamp: float) -> int:
    """Count the number of data points added after the specified timestamp.

    NOTE: Use `pamiq_core.time` to retrieve `timestamp`.

    Args:
        timestamp: Reference timestamp to count from.

    Returns:
        Number of data points added after the specified timestamp.
    """
    for i, t in enumerate(reversed(self._timestamps)):
        if t <= timestamp:
            return i
    return len(self._timestamps)

save_state

save_state(path: Path) -> None

Save the state of this DataUser to the specified path.

This method first updates the buffer with any pending collected data, then delegates the state saving to the underlying buffer.

PARAMETER	DESCRIPTION
path	Directory path where the state should be saved TYPE: Path

Source code in src/pamiq_core/data/interface.py

@override
def save_state(self, path: Path) -> None:
    """Save the state of this DataUser to the specified path.

    This method first updates the buffer with any pending collected data,
    then delegates the state saving to the underlying buffer.

    Args:
        path: Directory path where the state should be saved
    """
    self.update()
    path.mkdir()
    self._buffer.save_state(path / "buffer")
    with open(path / "timestamps.pkl", "wb") as f:
        pickle.dump(self._timestamps, f)

load_state

load_state(path: Path) -> None

Load the state of this DataUser from the specified path.

This method delegates the state loading to the underlying buffer.

PARAMETER	DESCRIPTION
path	Directory path from where the state should be loaded TYPE: Path

Source code in src/pamiq_core/data/interface.py

@override
def load_state(self, path: Path) -> None:
    """Load the state of this DataUser from the specified path.

    This method delegates the state loading to the underlying buffer.

    Args:
        path: Directory path from where the state should be loaded
    """
    self._buffer.load_state(path / "buffer")
    with open(path / "timestamps.pkl", "rb") as f:
        self._timestamps = deque(pickle.load(f), maxlen=self._buffer.max_queue_size)

__len__

__len__() -> int

Returns the current number of samples in the buffer.

Source code in src/pamiq_core/data/interface.py

def __len__(self) -> int:
    """Returns the current number of samples in the buffer."""
    return len(self._buffer)

pamiq_core.data.impls.SequentialBuffer

SequentialBuffer(max_size: int)

Bases: DataBuffer[T, list[T]]

Implementation of DataBuffer that maintains data in sequential order.

This buffer stores collected data points in an ordered queue, preserving the insertion order with a maximum size limit.

Initialize a new SequentialBuffer.

PARAMETER	DESCRIPTION
max_size	Maximum number of data points to store. TYPE: int

Source code in src/pamiq_core/data/impls/sequential_buffer.py

@override
def __init__(self, max_size: int):
    """Initialize a new SequentialBuffer.

    Args:
        max_size: Maximum number of data points to store.
    """
    super().__init__(max_size)

    self._queue: deque[T] = deque(maxlen=max_size)

    self._max_size = max_size

max_size property

max_size: int

Returns the maximum number of data points that can be stored in the buffer.

add

add(data: T) -> None

Add a new data sample to the buffer.

PARAMETER	DESCRIPTION
data	Data element to add to the buffer. TYPE: T

Source code in src/pamiq_core/data/impls/sequential_buffer.py

@override
def add(self, data: T) -> None:
    """Add a new data sample to the buffer.

    Args:
        data: Data element to add to the buffer.
    """
    self._queue.append(data)

get_data

get_data() -> list[T]

Retrieve all stored data from the buffer.

RETURNS	DESCRIPTION
list[T]	List of all stored data elements preserving the original insertion order.

Source code in src/pamiq_core/data/impls/sequential_buffer.py

@override
def get_data(self) -> list[T]:
    """Retrieve all stored data from the buffer.

    Returns:
        List of all stored data elements preserving the original insertion order.
    """
    return list(self._queue)

__len__

__len__() -> int

Returns the current number of samples in the buffer.

RETURNS	DESCRIPTION
int	The number of samples currently stored in the buffer. TYPE: int

Source code in src/pamiq_core/data/impls/sequential_buffer.py

@override
def __len__(self) -> int:
    """Returns the current number of samples in the buffer.

    Returns:
        int: The number of samples currently stored in the buffer.
    """
    return len(self._queue)

save_state

save_state(path: Path) -> None

Save the buffer state to the specified path.

Saves the data queue to a pickle file with .pkl extension.

PARAMETER	DESCRIPTION
path	File path where to save the buffer state (without extension) TYPE: Path

Source code in src/pamiq_core/data/impls/sequential_buffer.py

@override
def save_state(self, path: Path) -> None:
    """Save the buffer state to the specified path.

    Saves the data queue to a pickle file with .pkl extension.

    Args:
        path: File path where to save the buffer state (without extension)
    """
    with open(path.with_suffix(".pkl"), "wb") as f:
        pickle.dump(self._queue, f)

load_state

load_state(path: Path) -> None

Load the buffer state from the specified path.

Loads data queue from pickle file with .pkl extension.

PARAMETER	DESCRIPTION
path	File path from where to load the buffer state (without extension) TYPE: Path

Source code in src/pamiq_core/data/impls/sequential_buffer.py

@override
def load_state(self, path: Path) -> None:
    """Load the buffer state from the specified path.

    Loads data queue from pickle file with .pkl extension.

    Args:
        path: File path from where to load the buffer state (without extension)
    """
    with open(path.with_suffix(".pkl"), "rb") as f:
        self._queue = deque(pickle.load(f), maxlen=self._max_size)

pamiq_core.data.impls.RandomReplacementBuffer

RandomReplacementBuffer(
    max_size: int,
    replace_probability: float | None = None,
    expected_survival_length: int | None = None,
)

Bases: DataBuffer[T, list[T]]

Buffer implementation that randomly replaces elements when full.

This buffer keeps track of collected data and, when full, randomly replaces existing elements based on a configurable probability.

Initialize a RandomReplacementBuffer.

PARAMETER	DESCRIPTION
max_size	Maximum number of data points to store. TYPE: int
replace_probability	Probability of replacing an existing element when buffer is full. Must be between 0.0 and 1.0 inclusive. If None and expected_survival_length is provided, this will be computed automatically. Default is 1.0 if both are None. TYPE: float | None DEFAULT: None
expected_survival_length	Expected number of steps that data should survive in the buffer. Used to automatically compute replace_probability if replace_probability is None. Cannot be specified together with replace_probability. TYPE: int | None DEFAULT: None

RAISES	DESCRIPTION
ValueError	If replace_probability is not between 0.0 and 1.0 inclusive, or if both replace_probability and expected_survival_length are specified.

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

def __init__(
    self,
    max_size: int,
    replace_probability: float | None = None,
    expected_survival_length: int | None = None,
) -> None:
    """Initialize a RandomReplacementBuffer.

    Args:
        max_size: Maximum number of data points to store.
        replace_probability: Probability of replacing an existing element when buffer is full.
            Must be between 0.0 and 1.0 inclusive. If None and expected_survival_length is provided,
            this will be computed automatically. Default is 1.0 if both are None.
        expected_survival_length: Expected number of steps that data should survive in the buffer.
            Used to automatically compute replace_probability if replace_probability is None.
            Cannot be specified together with replace_probability.

    Raises:
        ValueError: If replace_probability is not between 0.0 and 1.0 inclusive, or if both
            replace_probability and expected_survival_length are specified.
    """

    if replace_probability is None:
        if expected_survival_length is None:
            replace_probability = 1.0
        else:
            replace_probability = (
                self.compute_replace_probability_from_expected_survival_length(
                    max_size, expected_survival_length
                )
            )
    elif expected_survival_length is not None:
        raise ValueError(
            "Cannot specify both replace_probability and expected_survival_length. "
            "Please specify only one of them."
        )
    if not (1.0 >= replace_probability >= 0.0):
        raise ValueError(
            "replace_probability must be between 0.0 and 1.0 inclusive"
        )
    super().__init__(int(max_size / replace_probability))
    self._max_size = max_size
    self._data_list: list[T] = []

    self._replace_probability = replace_probability
    self._current_size = 0

max_size property

max_size: int

Returns the maximum number of data points that can be stored in the buffer.

is_full property

is_full: bool

Check if the buffer has reached its maximum capacity.

RETURNS	DESCRIPTION
bool	True if the buffer is full, False otherwise.

compute_replace_probability_from_expected_survival_length staticmethod

compute_replace_probability_from_expected_survival_length(
    max_size: int, survival_length: int
) -> float

Compute the replace probability from expected survival length.

This method calculates the replacement probability needed to achieve a desired expected survival length for data in the buffer.

The computation is based on the mathematical analysis described in below

https://zenn.dev/gesonanko/scraps/b581e75bfd9f3e

PARAMETER	DESCRIPTION
max_size	Maximum size of the buffer. TYPE: int
survival_length	Expected number of steps that data should survive. TYPE: int

RETURNS	DESCRIPTION
float	The computed replacement probability between 0.0 and 1.0.

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

@staticmethod
def compute_replace_probability_from_expected_survival_length(
    max_size: int, survival_length: int
) -> float:
    """Compute the replace probability from expected survival length.

    This method calculates the replacement probability needed to achieve
    a desired expected survival length for data in the buffer.

    The computation is based on the mathematical analysis described in below:
        https://zenn.dev/gesonanko/scraps/b581e75bfd9f3e

    Args:
        max_size: Maximum size of the buffer.
        survival_length: Expected number of steps that data should survive.

    Returns:
        The computed replacement probability between 0.0 and 1.0.
    """
    gamma = 0.5772156649015329  # Euler-Mascheroni constant
    p = max_size / survival_length * (math.log(max_size) + gamma)
    return min(max(p, 0.0), 1.0)  # Clamp value between 0 to 1.

add

add(data: T) -> None

Add a new data sample to the buffer.

If the buffer is full, the new data may replace an existing entry based on the configured replacement probability.

PARAMETER	DESCRIPTION
data	Data element to add to the buffer. TYPE: T

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

@override
def add(self, data: T) -> None:
    """Add a new data sample to the buffer.

    If the buffer is full, the new data may replace an existing entry
    based on the configured replacement probability.

    Args:
        data: Data element to add to the buffer.
    """
    if self.is_full:
        if random.random() > self._replace_probability:
            return
        replace_index = random.randint(0, self._max_size - 1)
        self._data_list[replace_index] = data
    else:
        self._data_list.append(data)
        self._current_size += 1

get_data

get_data() -> list[T]

Retrieve all stored data from the buffer.

RETURNS	DESCRIPTION
list[T]	List of all stored data elements.
list[T]	Returns a copy of the internal data to prevent modification.

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

@override
def get_data(self) -> list[T]:
    """Retrieve all stored data from the buffer.

    Returns:
        List of all stored data elements.
        Returns a copy of the internal data to prevent modification.
    """
    return self._data_list.copy()

__len__

__len__() -> int

Returns the current number of samples in the buffer.

RETURNS	DESCRIPTION
int	The number of samples currently stored in the buffer. TYPE: int

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

@override
def __len__(self) -> int:
    """Returns the current number of samples in the buffer.

    Returns:
        int: The number of samples currently stored in the buffer.
    """
    return self._current_size

save_state

save_state(path: Path) -> None

Save the buffer state to the specified path.

Saves the data list to a pickle file with .pkl extension.

PARAMETER	DESCRIPTION
path	File path where to save the buffer state (without extension). TYPE: Path

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

@override
def save_state(self, path: Path) -> None:
    """Save the buffer state to the specified path.

    Saves the data list to a pickle file with .pkl extension.

    Args:
        path: File path where to save the buffer state (without extension).
    """
    with open(path.with_suffix(".pkl"), "wb") as f:
        pickle.dump(self._data_list, f)

load_state

load_state(path: Path) -> None

Load the buffer state from the specified path.

Loads data list from pickle file with .pkl extension.

PARAMETER	DESCRIPTION
path	File path from where to load the buffer state (without extension). TYPE: Path

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

@override
def load_state(self, path: Path) -> None:
    """Load the buffer state from the specified path.

    Loads data list from pickle file with .pkl extension.

    Args:
        path: File path from where to load the buffer state (without extension).
    """
    with open(path.with_suffix(".pkl"), "rb") as f:
        self._data_list = list(pickle.load(f))[: self._max_size]
    self._current_size = len(self._data_list)

pamiq_core.data.impls.DictSequentialBuffer

DictSequentialBuffer(keys: Iterable[str], max_size: int)

Bases: DataBuffer[Mapping[str, T], dict[str, list[T]]]

Buffer implementation that stores dictionary data in sequential order.

See: SequentialBuffer

Initialize a DictSequentialBuffer.

PARAMETER	DESCRIPTION
keys	The keys that must be present in all data dictionaries. TYPE: Iterable[str]
max_size	Maximum number of data points to store. TYPE: int

Source code in src/pamiq_core/data/impls/sequential_buffer.py

def __init__(
    self,
    keys: Iterable[str],
    max_size: int,
) -> None:
    """Initialize a DictSequentialBuffer.

    Args:
        keys: The keys that must be present in all data dictionaries.
        max_size: Maximum number of data points to store.
    """
    self._buffer = SequentialBuffer[dict[str, T]](max_size)
    super().__init__(self._buffer.max_queue_size)

    self._keys = set(keys)

    self.save_state = self._buffer.save_state
    self.load_state = self._buffer.load_state

max_size property

max_size: int

Returns the maximum number of data points that can be stored in the buffer.

add

add(data: Mapping[str, T]) -> None

Add a new data sample to the buffer.

The data must contain exactly the keys specified during initialization. If the buffer is full, the oldest entry will be removed.

PARAMETER	DESCRIPTION
data	Dictionary containing data for each key. TYPE: Mapping[str, T]

RAISES	DESCRIPTION
ValueError	If the data keys don't match the expected keys.

Source code in src/pamiq_core/data/impls/sequential_buffer.py

@override
def add(self, data: Mapping[str, T]) -> None:
    """Add a new data sample to the buffer.

    The data must contain exactly the keys specified during initialization.
    If the buffer is full, the oldest entry will be removed.

    Args:
        data: Dictionary containing data for each key.

    Raises:
        ValueError: If the data keys don't match the expected keys.
    """
    if set(data.keys()) != self._keys:
        raise ValueError(
            f"Data keys {set(data.keys())} do not match expected keys {self._keys}"
        )
    return self._buffer.add(dict(data))

get_data

get_data() -> dict[str, list[T]]

Retrieve all stored data from the buffer.

RETURNS	DESCRIPTION
dict[str, list[T]]	Dictionary mapping each key to a list of its stored values.
dict[str, list[T]]	The lists maintain the sequential order in which data was added.

Source code in src/pamiq_core/data/impls/sequential_buffer.py

@override
def get_data(self) -> dict[str, list[T]]:
    """Retrieve all stored data from the buffer.

    Returns:
        Dictionary mapping each key to a list of its stored values.
        The lists maintain the sequential order in which data was added.
    """
    out = {k: list[T]() for k in self._keys}
    for data in self._buffer.get_data():
        for k, v in data.items():
            out[k].append(v)
    return out

__len__

__len__() -> int

Returns the current number of samples in the buffer.

RETURNS	DESCRIPTION
int	The number of samples currently stored in the buffer. TYPE: int

Source code in src/pamiq_core/data/impls/sequential_buffer.py

@override
def __len__(self) -> int:
    """Returns the current number of samples in the buffer.

    Returns:
        int: The number of samples currently stored in the buffer.
    """
    return len(self._buffer)

pamiq_core.data.impls.DictRandomReplacementBuffer

DictRandomReplacementBuffer(
    keys: Iterable[str],
    max_size: int,
    replace_probability: float | None = None,
    expected_survival_length: int | None = None,
)

Bases: DataBuffer[Mapping[str, T], dict[str, list[T]]]

Buffer implementation that stores dictionary data with random replacement.

See RandomReplacementBuffer.

Initialize a DictRandomReplacementBuffer.

PARAMETER	DESCRIPTION
keys	The keys that must be present in all data dictionaries. TYPE: Iterable[str]

Other arguments are same as RandomReplacementBuffer.

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

def __init__(
    self,
    keys: Iterable[str],
    max_size: int,
    replace_probability: float | None = None,
    expected_survival_length: int | None = None,
) -> None:
    """Initialize a DictRandomReplacementBuffer.

    Args:
        keys: The keys that must be present in all data dictionaries.

    Other arguments are same as [`RandomReplacementBuffer`][pamiq_core.data.impls.RandomReplacementBuffer].
    """
    self._buffer = RandomReplacementBuffer[dict[str, T]](
        max_size, replace_probability, expected_survival_length
    )
    super().__init__(self._buffer.max_queue_size)

    self._keys = set(keys)

    self.save_state = self._buffer.save_state
    self.load_state = self._buffer.load_state

max_size property

max_size: int

Returns the maximum number of data points that can be stored in the buffer.

add

add(data: Mapping[str, T]) -> None

Add a new data sample to the buffer.

The data must contain exactly the keys specified during initialization.

PARAMETER	DESCRIPTION
data	Dictionary containing data for each key. TYPE: Mapping[str, T]

RAISES	DESCRIPTION
ValueError	If the data keys don't match the expected keys.

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

@override
def add(self, data: Mapping[str, T]) -> None:
    """Add a new data sample to the buffer.

    The data must contain exactly the keys specified during initialization.

    Args:
        data: Dictionary containing data for each key.

    Raises:
        ValueError: If the data keys don't match the expected keys.
    """
    if set(data.keys()) != self._keys:
        raise ValueError(
            f"Data keys {set(data.keys())} do not match expected keys {self._keys}"
        )
    return self._buffer.add(dict(data))

get_data

get_data() -> dict[str, list[T]]

Retrieve all stored data from the buffer.

RETURNS	DESCRIPTION
dict[str, list[T]]	Dictionary mapping each key to a list of its stored values.
dict[str, list[T]]	The lists maintain the order in which data was added/replaced.

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

@override
def get_data(self) -> dict[str, list[T]]:
    """Retrieve all stored data from the buffer.

    Returns:
        Dictionary mapping each key to a list of its stored values.
        The lists maintain the order in which data was added/replaced.
    """
    out = {k: list[T]() for k in self._keys}
    for data in self._buffer.get_data():
        for k, v in data.items():
            out[k].append(v)
    return out

__len__

__len__() -> int

Returns the current number of samples in the buffer.

RETURNS	DESCRIPTION
int	The number of samples currently stored in the buffer. TYPE: int

Source code in src/pamiq_core/data/impls/random_replacement_buffer.py

@override
def __len__(self) -> int:
    """Returns the current number of samples in the buffer.

    Returns:
        int: The number of samples currently stored in the buffer.
    """
    return len(self._buffer)