Module

utils.py

import goldener.utils
View source

Generic utilities shared by the Goldener features.

This module gathers helpers to validate inputs (shapes, sampling sizes, types), to compute sampling counts from ratios and to distribute them among chunks, and to filter or transform batches (e.g. binarizing multiple targets, or splitting multilabel samples into independent labels).

On this page 14 functions
check_x_and_y_shapes() get_ratio_list_sum() get_ratios_for_counts() split_sampling_among_chunks() filter_batch_from_indices() get_indices_with_labels() get_size_and_sampling_count_per_chunk() check_sampling_size() check_all_same_type() get_sampling_count_from_size() transform_batch_from_multiple_to_binarized_targets() transform_batch_from_multilabel_to_independent_labels() get_sampling_count_from_ratios() force_non_zero_count()
def check_x_and_y_shapes(x_shape: tuple[int, ...], y_shape: tuple[int, ...]) -> None:

Check compatibility of shapes of x and y tensors.

Arguments:
  • x_shape: Shape of the input tensor x.
  • y_shape: Shape of the target tensor y.
Raises:
  • ValueError: If the shapes are incompatible. See the conditions below for details.
Conditions:
  • If x is 1D, x and y must have the same shape.
  • If x is 2D or more, y must have a channel dimension of 1 (i.e., y_shape[1] == 1).
  • If y has a batch size greater than 1 (i.e., y_shape[0] > 1), x and y must have the same batch size (i.e., x_shape[0] == y_shape[0]).
  • If x has more than 2 dimensions, the additional (after channel) dimensions of x and y must match.
def get_ratio_list_sum(ratios: list[float]) -> float:

Get the sum of a list of ratios and validate it (total between 0 and 1).

Arguments:
  • ratios: A list of float ratios.
Returns:

The sum of the ratios.

Raises:
  • ValueError: If the sum of the ratios is not greater than 0 and at most 1.
def get_ratios_for_counts(counts: list[int]) -> list[float]:

Get ratios for a list of counts.

Arguments:
  • counts: A list of integer counts.
Returns:

A list of float ratios corresponding to the input counts.

Raises:
  • ValueError: If counts is an empty list or if the total count is zero.
def split_sampling_among_chunks(to_split: int, chunk_sizes: list[int]) -> list[int]:

Split a total sampling count among chunks based on their sizes.

Arguments:
  • to_split: The total number of samples to split among the chunks.
  • chunk_sizes: A list of integers representing the sizes of each chunk.
Returns:

A list of integers representing the number of samples to draw from each chunk, summing up to to_split.

Raises:
  • ValueError: If the input parameters are invalid. See the conditions below for details.
Conditions:
  • At least one chunk size must be provided.
  • If only one chunk size is provided, to_split must not be greater than that chunk size.
  • The total size of the chunks must be greater than 0.
def filter_batch_from_indices( batch: dict[str, typing.Any], to_remove: set[int], index_key: str = 'idx') -> dict[str, typing.Any]:

Filter a batch dictionary to only include items at specified indices.

Arguments:
  • batch: A dictionary representing a batch of data (each key corresponds to stacked information).
  • to_remove: A set of indices to filter out from the batch.
  • index_key: The key in the batch dictionary that contains the indices.
Returns:

A filtered batch dictionary without the specified indices.

def get_indices_with_labels( batch: dict[str, typing.Any], label_key: str, exclude_labels: set[str], index_key: str = 'idx') -> set[int]:

Get the set of index values for batch items whose sharing label values with exclude_labels.

Arguments:
  • batch: A dictionary representing a batch of data. It must contain the keys specified by label_key and index_key,
  • where the values are lists or tensors of the same length. Label values can be strings or iterables of strings.
  • label_key: The key in the batch dictionary that contains the labels.
  • exclude_labels: A set of label strings to exclude.
  • index_key: The key in the batch dictionary that contains the indices.
Returns:

A set of index values corresponding to items with excluded labels.

def get_size_and_sampling_count_per_chunk( total_size: int, sampling_size: int, max_chunk_size: int) -> tuple[list[int], list[int]]:

Get sizes and sampling sizes per chunk.

Arguments:
  • total_size: Total size of the data to sample from.
  • sampling_size: Total number of samples to draw.
  • max_chunk_size: Maximum size of each chunk.
Returns:

A tuple of two lists: - A list of integers representing the size of each chunk. - A list of integers representing the number of samples to draw from each chunk.

Raises:
  • ValueError: If sampling_size is greater than or equal to total_size.
def check_sampling_size( sampling_size: int | float, total_size: int | None = None, force_max: bool = False) -> None:

Check the validity of the sampling size.

Arguments:
  • sampling_size: The sampling size to check (can be int or float).
  • total_size: The total size of the data (Optional).
  • force_max: Whether the sampling size is expected to be the maximum possible (1.0 for float, total for int)
Raises:
  • ValueError: If the sampling size is invalid based on its type and total size.
  • If sampling_size is a float, it must be in the range (0.0, 1.0].
  • If sampling_size is an int, it must be in the range (0, total_size].
def check_all_same_type(iterable: Iterable[Any]) -> None:

Check if all elements in an iterable are of the same type.

Arguments:
  • iterable: An iterable containing elements to check.
Raises:
  • TypeError: If any element in the iterable is not of the expected type.
def get_sampling_count_from_size(sampling_size: int | float, total_size: int | None = None) -> int:

Get the sampling count from the sampling size.

Arguments:
  • sampling_size: The sampling size (can be int or float). If int, it represents the exact number of samples
  • and it is required to be less than total_size. If float, it represents the fraction of total size, rounded
  • down with a minimum of one sample, and it is required to be in the range (0.0, 1.0].
  • total_size: The total size of the data (Optional, required if sampling_size is float).
Returns:

The calculated sampling count as an integer.

Raises:
  • ValueError: If the sampling size is invalid based on its type and total size.
def transform_batch_from_multiple_to_binarized_targets( batch: dict[str, typing.Any], target_key: str, label_key: str | None, target_to_label: dict[tuple[int, ...], str], merge_labels: bool = False, exclude_labels: set[str] | None = None, exclude_full_zero: bool = False) -> dict[str, typing.Any]:

Transform a batch with multiple targets into a batch with binarized targets for each label.

Arguments:
  • batch: A dictionary representing a batch of data, where the target key contains a target with multiple values corresponding to the labels in target_to_label.
  • target_key: The key in the batch dictionary that contains the target.
  • label_key: The key in the batch dictionary that containes the labels.
  • target_to_label: A mapping from unique target tuples to label values.
  • merge_labels: Whether to merge the labels into a single label (default: False).
  • exclude_labels: An optional set of labels to exclude from the transformation (default: None).
  • exclude_full_zero: Whether to exclude the all-zero target from the transformation (default: False).
Returns:

A new batch dictionary where the target key contains binarized targets for each label and the label key contains the corresponding labels, with other batch elements duplicated accordingly.

Raises:
  • ValueError: If a unique target tuple is not found in target_to_label.
  • ValueError: If no valid targets remain after applying the exclude_full_zero filter.
def transform_batch_from_multilabel_to_independent_labels( batch: dict[str, torch.Tensor | list], label_key: str, merge_labels: bool = False, exclude_labels: set[str] | None = None) -> dict[str, torch.Tensor | list]:

Transform a batch with multilabel into a batch with one entry per label.

Arguments:
  • batch: A dictionary representing a batch of data, where the label key contains several labels.
  • label_key: The key in the batch dictionary that contains the labels.
  • merge_labels: Whether to merge the labels into a single label (default: False). If False, the batch will be duplicated for each label. If True, the labels will be merged into a single label by concatenating them with a separator (e.g., "_").
  • exclude_labels: An optional set of labels to exclude from the transformation (default: None).
Returns:

A new batch dictionary where samples with multiple labels are split/merge in single entries.

def get_sampling_count_from_ratios( ratios: dict[~T, float], sampling_size: int, force_non_zero: bool = False) -> dict[~T, int]:

Get the sampling count for each key from the ratios.

Arguments:
  • ratios: A dictionary mapping keys to their corresponding ratios (values between 0 and 1). the initial counts are computed by multiplying the sampling size by the ratios and taking the floor of the result. If the sum of the initial counts is less than the sampling size, the remaining samples are distributed one by one starting from the most represented key until the total count matches the sampling size.
  • sampling_size: The total number of samples to draw.
  • force_non_zero: Whether to force a non-zero sampling count for each key (default: False). If activated, the least represented keys will have at least 1 sample, and the most represented keys will have their count reduced accordingly to maintain the total sampling size.
Returns:

A dictionary mapping keys to their corresponding sampling counts (integers). The dictionary is sorted by count in ascending order.

Raises:
  • ValueError: If the ratios do not sum to 1.0.
  • ValueError: If force_non_zero is True and the count for the highest-count key becomes zero while redistributing counts.
def force_non_zero_count(counts: dict[~T, int]) -> dict[~T, int]: