Module

select.py

import goldener.select

Selection of the most representative subset of samples.

The selection finds a coresubset: a subset of samples representing at best the full dataset from its vectors. It is used to choose the data to annotate and, by the goldener.split.GoldSplitter, to fill the sets of a split.

GoldSelectionTool is the abstract interface of the coresubset selection algorithms.
GoldGreedyKernelPointsSelectionTool, GoldGreedyKCenterSelectionTool, GoldGreedyClosestPointSelectionTool, GoldGreedyFarthestPointSelectionTool and GoldZCoreSelectionTool implement different selection algorithms.
GoldSelector runs the selection on a whole dataset of vectors, possibly in chunks, after a dimensionality reduction (goldener.reduce) or stratified by label, and stores the selected samples locally.

On this page 8 classes

DistanceType GoldSelectionTool GoldGreedyKernelPointsSelectionTool GoldGreedyKCenterSelectionTool GoldGreedyClosestPointSelectionTool GoldGreedyFarthestPointSelectionTool GoldZCoreSelectionTool GoldSelector

logger = <Logger goldener.select (INFO)>

class DistanceType(enum.Enum):

EUCLIDEAN = <DistanceType.EUCLIDEAN: 'euclidean'>

COSINE = <DistanceType.COSINE: 'cosine'>

distance_function: Callable[[torch.Tensor, torch.Tensor], torch.Tensor]

class GoldSelectionTool(abc.ABC):

Run selection of a subset of data points from vectorized samples using a coresubset selection algorithm.

Attributes:

distance_func: The distance function to use for computing distances between points, determined by the distance parameter in the constructor.

distance_func

@abstractmethod

def select( self, x: torch.Tensor, k: int, anchors: torch.Tensor | None = None) -> list[int]:

Select a subset of data points from the input data.

Arguments:

x: A 2D torch.Tensor where each row represents a data point.
k: The number of data points to select.
anchors: Optional 2D torch.Tensor of anchor points to consider during selection. The shape should be (num_anchors, feature_dim). If provided, the selection will be performed with respect to these anchor points and it will affect the selection process.

Returns:

class GoldGreedyKernelPointsSelectionTool(GoldSelectionTool):

Coresubset selection using Coreax's GreedyKernelPoints solver.

The solver is created at each request depending on the specified k points to select.

With this method, the anchors are not used for the selection process.

Attributes:

feature_kernel: The kernel to use for the GreedyKernelPoints solver.
random_key: Random key for reproducibility in the GreedyKernelPoints solver.
distance_func: Not used in this selection tool, included for compatibility with the base class.

GoldGreedyKernelPointsSelectionTool( feature_kernel: coreax.kernels.base.ScalarValuedKernel, random_key: int = 42, distance: DistanceType = <DistanceType.EUCLIDEAN: 'euclidean'>)

Initialize the GoldGreedyKernelPointsSelectionTool.

Arguments:

feature_kernel: The kernel to use for the GreedyKernelPoints solver.
random_key: Random key for reproducibility in the GreedyKernelPoints solver. Default is 42.
distance: Not used in this selection tool, included for compatibility with the base class.

random_key

feature_kernel

def select( self, x: torch.Tensor, k: int, anchors: torch.Tensor | None = None) -> list[int]:

Select a subset of data points from the input data using greedy kernel point coresubset sampling.

With this method, the anchors are not used for the selection process.

Arguments:

x: A 2D torch.Tensor where each row represents a data point.
k: The number of data points to select.
anchors: Not used in this selection tool, included for compatibility with the base class.

Returns:

Raises:

ValueError: If x is not a 2D tensor. If k is greater than the number of data points in x.

Inherited Members

GoldSelectionTool: distance_func

class GoldGreedyKCenterSelectionTool(GoldSelectionTool):

Create a coresubset from selecting with the K Center greedy algorithm.

This is a greedy algorithm that selects iteratively the farthest to the already selected points among the not selected points. The first center is selected as the point with the largest aggregated distance to all the other points.

Attributes:

device: The torch device to use for computations.
distance_func: The distance function to use for computing distances between points, determined by the distance parameter in the constructor.

GoldGreedyKCenterSelectionTool( device: torch.device | str, distance: DistanceType = <DistanceType.EUCLIDEAN: 'euclidean'>)

Initialize the GoldGreedyKCenterSelectionTool.

Arguments:

device: The torch device to use for computations.
distance: The distance type to use for selection.

device

def select( self, x: torch.Tensor, k: int, anchors: torch.Tensor | None = None) -> list[int]:

Select a subset of data points from the input data by choosing iteratively the point with the farthest nearest neighbors among the not selected points.

Arguments:

x: A 2D torch.Tensor where each row represents a data point.
k: The number of data points to select.
anchors: Optional 2D torch.Tensor of anchor points to consider during selection. The shape should be (num_anchors, feature_dim). If provided, the selection will be performed with respect to these anchor points and it will affect the selection process.

Returns:

Raises:

ValueError: If x is not a 2D tensor. If k is greater than the number of data points in x.

Inherited Members

GoldSelectionTool: distance_func

class GoldGreedyClosestPointSelectionTool(GoldSelectionTool):

Create a coresubset from selecting the closest points iteratively.

This is a greedy algorithm that selects iteratively the point with the closest nearest neighbors among the not selected points.

With this method, the anchors are not used for the selection process.

Attributes:

device: The torch device to use for computations.

GoldGreedyClosestPointSelectionTool( device: torch.device | str, distance: DistanceType = <DistanceType.EUCLIDEAN: 'euclidean'>)

Initialize the GoldGreedyClosestPointSelectionTool.

Arguments:

device: The torch device to use for computations.
distance: The distance type to use for selection.

device

def select( self, x: torch.Tensor, k: int, anchors: torch.Tensor | None = None) -> list[int]:

Select a subset of data points from the input data by choosing iteratively the point with the closest nearest neighbors among the not selected points.

Arguments:

x: A 2D torch.Tensor where each row represents a data point.
k: The number of data points to select.
anchors: Not used in this selection tool, included for compatibility with the base class.

Returns:

Raises:

ValueError: If x is not a 2D tensor. If k is greater than the number of data points in x.

Inherited Members

GoldSelectionTool: distance_func

class GoldGreedyFarthestPointSelectionTool(GoldSelectionTool):

Create a coresubset from selecting the farthest points iteratively.

This is a greedy algorithm that selects iteratively the point with the farthest nearest neighbors among the not selected points.

With this method, the anchors are not used for the selection process.

Attributes:

device: The torch device to use for computations.
distance_func: The distance function to use for computing distances between points, determined by the distance parameter in the constructor.

GoldGreedyFarthestPointSelectionTool( device: torch.device | str, distance: DistanceType = <DistanceType.EUCLIDEAN: 'euclidean'>)

Initialize the GoldGreedyFarthestPointSelectionTool.

Arguments:

device: The torch device to use for computations.
distance: The distance type to use for selection.

device

def select( self, x: torch.Tensor, k: int, anchors: torch.Tensor | None = None) -> list[int]:

Select a subset of data points from the input data by choosing iteratively the point with the farthest nearest neighbors among the not selected points.

Arguments:

x: A 2D torch.Tensor where each row represents a data point.
k: The number of data points to select.
anchors: Not used in this selection tool, included for compatibility with the base class.

Returns:

Raises:

ValueError: If x is not a 2D tensor. If k is greater than the number of data points in x.

Inherited Members

GoldSelectionTool: distance_func

class GoldZCoreSelectionTool(GoldSelectionTool):

Create a coresubset from selecting with the ZCore algorithm.

This algorithm compute a score of coverage penalized by a redundancy penalty for each point and select the points with the highest score.

This selection tool is inspired from Zero-Shot Coreset Selection via Iterative Subspace Sampling.

Attributes:

device: The torch device to use for computations.
distance_func: The distance function to use for computing distances between points
n_dim_for_score: The number of dimensions to use when computing the coverage score and redundancy penalty. Default is 2.
n_random_anchors: The number of random anchors to use for computing the coverage score. Default is 2 million.
n_redundancy: The number of closest points to consider for the redundancy penalty. Default is 1000.
redundancy_scale: The scale factor for the redundancy penalty. Default is 4.
eps: A small value to avoid division by zero in the redundancy penalty. Default is 1e-8.
random_state: Optional random state for reproducibility when generating random anchors. Default is None.

GoldZCoreSelectionTool( device: torch.device | str, distance: DistanceType = <DistanceType.EUCLIDEAN: 'euclidean'>, n_dim_for_score: int = 2, n_random_anchors: int = 2000000, n_redundancy: int = 1000, redundancy_scale: int = 4, eps: float = 1e-08, random_state: int | None = None)

Initialize the GoldZCoreSelectionTool.

Arguments:

device: The torch device to use for computations.
distance: The distance type to use for selection.
n_dim_for_score: The number of dimensions to use when computing the coverage score and redundancy penalty. Default is 2.
n_random_anchors: The number of random anchors to use for computing the coverage score. Default is 2 million.
n_redundancy: The number of closest points to consider for the redundancy penalty. Default is 1000.
redundancy_scale: The scale factor for the redundancy penalty. Default is 4.
eps: A small value to avoid division by zero in the redundancy penalty. Default is 1e-8.
random_state: Optional random state for reproducibility when generating random anchors. Default is None.

Raises:

ValueError: If n_dim_for_score is not a positive integer. If n_random_anchors is not a positive integer. If eps is not a positive float.

device

n_dim_for_score

n_random_anchors

n_redundancy

redundancy_scale

eps

np_generator

def select( self, x: torch.Tensor, k: int, anchors: torch.Tensor | None = None) -> list[int]:

Select a subset of data points from the ZCore algorithm.

Arguments:

x: A 2D torch.Tensor where each row represents a data point.
k: The number of data points to select.
anchors: This selection tool does not use anchors, included for compatibility with the base class.

Returns:

Raises:

ValueError: If x is not a 2D tensor. If k is greater than the number of data points in x. if n_dim_for_score is greater than the number of dimensions in x.

Inherited Members

GoldSelectionTool: distance_func

class GoldSelector(goldener.do.GoldDoerWithTable):

Select a subset of data points from vectorized samples.

The GoldSelector processes a dataset or PixelTable table to perform coresubset selection using a selection algorithm on already vectorized representations. The selection results are stored in a local PixelTable table (specified by table_path) so that the selection process is idempotent: calling the same operation multiple times will not duplicate or recompute selections that are already present in the table.

If the dataset is too big to fit into memory or the coresubset selection algorithm is too computationally expensive, the coresubset selection can be performed in chunks. All torch tensors will be converted to numpy arrays before saving.

The selection can operate in a sequential (single-process) mode or a distributed mode (not implemented). The table schema is created/validated automatically and will include minimal indexing columns (idx, idx_vector) required to link selected samples back to their originating data points.

Attributes:

table_path: Path to the PixelTable table where selection results will be stored locally.
selection_tool: The GoldSelectionTool implementing the selection algorithm.
reducer: Optional GoldReductionTool instance for dimensionality reduction before selection.
chunk: Optional chunk size for processing data in chunks to reduce memory consumption.
collate_fn: Optional function to collate dataset samples into batches composed of dictionaries with at least the key specified by vectorized_key returning a PyTorch Tensor. If None, collate_keeping_sequences_as_sequences is used, which preserves sequence-valued fields as one sequence per sample.
vectorized_key: Key in the batch dictionary that contains the vectorized data for selection. Default is "vectorized".
include_vectorized_in_table: Whether to include the vectorized data in the selection table. Defaults to False.
It is only applied if the cluster table is created from a Table (it is forced anyway for Dataset).
selection_key: Column name to store the selection value in the PixelTable table. Default is "selected".
label_key: Optional key for label-based stratified selection.
exclude_labels: Optional set of label strings to exclude from selection. Default is None.
to_keep_schema: Optional dictionary defining additional columns to keep from the original dataset/table into the selection table. The keys are the column names and the values are the PixelTable types.
min_pxt_insert_size: Minimum number of rows to accumulate before inserting into PixelTable. Default is 100.
batch_size: Batch size used when iterating over the data.
num_workers: Number of workers for the PyTorch DataLoader during iteration on data.
allow_existing: If False, an error will be raised when the table already exists. Default is True.
distribute: Whether to use distributed processing for selection and table population. Not implemented yet. Default is False.
drop_table: Whether to drop the selection table after creating the dataset with selection results. It is only applied when using select_in_dataset. Default is False.
max_batches: Optional maximum number of batches to process. Useful for testing on a small subset of the dataset.
random_state: Optional random state for reproducibility during chunk assignment. Default is None.

GoldSelector( table_path: str, selection_tool: GoldSelectionTool = <GoldGreedyKernelPointsSelectionTool object>, reducer: goldener.reduce.GoldReductionTool | None = None, chunk: int | None = None, collate_fn: Optional[Callable] = None, vectorized_key: str = 'vectorized', include_vectorized_in_table: bool = False, selection_key: str = 'selected', label_key: str | None = None, exclude_labels: set[str] | None = None, to_keep_schema: dict[str, type] | None = None, min_pxt_insert_size: int = 100, batch_size: int = 1, num_workers: int = 0, allow_existing: bool = True, distribute: bool = False, drop_table: bool = False, max_batches: int | None = None, random_state: int | None = None)

Initialize the GoldSelector.

Arguments:

table_path: Path to store the PixelTable table.
selection_tool: The GoldSelectionTool implementing the selection algorithm.
reducer: Optional GoldReductionTool instance for dimensionality reduction before selection.
chunk: Optional chunk size for processing data in chunks.
collate_fn: Optional collate function for the DataLoader. If None, collate_keeping_sequences_as_sequences is used.
vectorized_key: Key pointing to the vector for selection. Defaults to "vectorized".
include_vectorized_in_table: Whether to include the vectorized data in the selection table. Defaults to False. It is only applied if the selection table is created from a Table (it is forced anyway for Dataset).
selection_key: Key for storing selection values. Defaults to "selected".
label_key: Optional key for label stratification.
exclude_labels: Optional set of label strings to exclude from selection. Default is None.
to_keep_schema: Optional schema for additional columns to preserve.
min_pxt_insert_size: Minimum number of rows to accumulate before inserting into PixelTable. Default is 100.
batch_size: Batch size used when iterating over the data.
num_workers: Number of workers for the PyTorch DataLoader during iteration on data.
allow_existing: Whether to allow using an existing table. Defaults to True.
distribute: Whether to use distributed selection. Not implemented yet. Defaults to False.
drop_table: Whether to drop the table after dataset creation. Defaults to False.
max_batches: Optional maximum number of batches to process.
random_state: Optional random state for reproducibility during chunk assignment. Default is None.

Raises:

ValueError: If exclude_labels is provided but label_key is None.
NotImplementedError: If distribute is True.

table_path

selection_tool

reducer

chunk

vectorized_key

include_vectorized_in_table

selection_key

label_key

exclude_labels

random_state

distribute: bool

Whether distributed processing is enabled.

def setup_selection_table( self, select_from: torch.utils.data.dataset.Dataset | pixeltable.catalog.table.Table) -> pixeltable.catalog.table.Table:

Set up the selection table based on the source dataset or table.

Arguments:

select_from: Dataset or Table to select from. If a Dataset is provided, each item should be a dictionary with at least the vectorized_key and idx keys after applying the collate_fn. If a Table is provided, it should contain at least the vectorized_key, idx and idx_vector columns.

Returns: A PixelTable Table with the appropriate schema for selection, populated with rows corresponding to the source dataset or table.

Raises:

ValueError: If allow_existing is False and the table at table_path already exists, or if an existing selection table is missing the expected primary key idx_vector.

def select_in_dataset( self, select_from: torch.utils.data.dataset.Dataset | pixeltable.catalog.table.Table, select_size: int | float, value: str, restrict_to: set[int] | None = None, restriction_idx_key: str = 'idx_vector') -> goldener.pxt_utils.GoldPxtTorchDataset:

Select a subset of samples using coresubset selection and return results as a GoldPxtTorchDataset.

The selection process applies a coresubset selection algorithm on already vectorized representations of the data points and stores results in a PixelTable table specified by table_path. When the chunk attribute is set, the selection is performed in chunks to reduce memory consumption. If a reducer is provided, the vectors are reduced in dimension before applying the coresubset selection.

This is a convenience wrapper that runs select_in_table to populate (or resume populating) the PixelTable table, then wraps the table into a GoldPxtTorchDataset for downstream consumption. If drop_table is True, the table will be removed after the dataset is created.

Arguments:

select_from: Dataset or Table to select from. If a Dataset is provided, each item should be a dictionary with at least the vectorized_key and idx keys after applying the collate_fn. If a Table is provided, it should contain at least the vectorized_key, idx and idx_vector columns.
select_size: Number or ratio (between 0 and 1) of data points to select
value: Value to set in the selection_key column for selected samples.
restrict_to: Optional set of indices to restrict the search to. If provided, only the selected indices in restriction_idx_key will be used for the selection.
restriction_idx_key: Column name used to get sample indices for restriction.

Returns:

def select_in_table( self, select_from: torch.utils.data.dataset.Dataset | pixeltable.catalog.table.Table, select_size: int | float, value: str | None, restrict_to: set[int] | None = None, restriction_idx_key: str = 'idx_vector') -> pixeltable.catalog.table.Table:

Select a subset of samples using coresubset selection and store results in a PixelTable table.

This method is idempotent (i.e., failure proof), meaning that if it is called multiple times on the same dataset or table, it will restart the selection process based on the vectors already present in the PixelTable table.

Arguments:

select_from: Dataset or Table to select from. If a Dataset is provided, each item should be a dictionary with at least the vectorized_key and idx keys after applying the collate_fn. If a Table is provided, it should contain at least the vectorized_key, idx and idx_vector columns.
select_size: Number or ratio (between 0 and 1) of data points to select.
value: Value to set in the selection_key column for selected samples.
restrict_to: Optional set of indices to restrict the search to. If provided, only the selected indices in restriction_idx_key will be used for the selection.
restriction_idx_key: Column name used to get sample indices for restriction.

Returns:

Raises:

ValueError: If restriction_idx_key is not "idx" or "idx_vector", or if select_size is a float not in the range ]0.0, 1.0].

@staticmethod

def get_selection_pxt_query( table: pixeltable.catalog.table.Table, value: str | None, selection_key: str, label_key: str | None = None, label_value: str | None = None, idx_key: str = 'idx', restrict_to: set[int] | None = None, restriction_idx_key: str = 'idx_vector') -> pixeltable._query.Query:

Get the Pixeltable query to access samples selected with a given value.

Arguments:

table: PixelTable table to query.
value: Value in the selection_key column to filter selected samples.
selection_key: Column name used to store selection values.
label_key: Optional column name used to filter samples by label.
label_value: Optional label value to filter samples by label.
idx_key: Column name used to get sample indices.
restrict_to: Optional set of indices to restrict the search to. If provided, only the selected indices in restriction_idx_key will be used for the selection.
restriction_idx_key: Column name used to get sample indices for restriction.

Returns:

Raises:

ValueError: If only one of label_key or label_value is provided (both must be set together).

@staticmethod

def get_selection_indices( table: pixeltable.catalog.table.Table, value: str | None, selection_key: str = 'selected', label_key: str | None = None, label_value: str | None = None, idx_key: str = 'idx', restrict_to: set[int] | None = None, restriction_idx_key: str = 'idx_vector') -> set[int]:

Get the indices of samples selected with a given value.

Arguments:

table: PixelTable table to query.
value: Value in the selection_key column to filter selected samples.
selection_key: Column name used to store selection values.
label_key: Optional column name used to filter samples by label.
label_value: Optional label value to filter samples by label.
idx_key: Column name used to get sample indices.
restrict_to: Optional set of indices to restrict the search to. If provided, only the selected indices in restriction_idx_key will be used for the selection.
restriction_idx_key: Column name used to get sample indices for restriction.

Returns:

Raises:

ValueError: If only one of label_key or label_value is provided (both must be set together).

@staticmethod

def get_selection_count( table: pixeltable.catalog.table.Table, value: str | None, selection_key: str, label_key: str | None = None, label_value: str | None = None, idx_key: str = 'idx', restrict_to: set[int] | None = None, restriction_idx_key: str = 'idx_vector') -> int:

Get the number of samples selected with a given value.

Arguments:

table: PixelTable table to query.
value: Value in the selection_key column to filter selected samples.
selection_key: Column name used to store selection values.
label_key: Optional column name used to filter samples by label.
label_value: Optional label value to filter samples by label.
idx_key: Column name used to get sample indices.
restrict_to: Optional set of indices to restrict the search to. If provided, only the selected indices in restriction_idx_key will be used for the selection.
restriction_idx_key: Column name used to get sample indices for restriction.

Returns:

Inherited Members

goldener.do.GoldDoerWithTable: collate_fn; to_keep_schema; min_pxt_insert_size; batch_size; num_workers
goldener.do.GoldDoer: allow_existing; drop_table; max_batches