Data Dataset Utils#

`dataset_utils` #

CLASS	DESCRIPTION
`ConcatDatasetWeighted`	Concatenate multiple DatasetWeighted instances into a single dataset.
`ChainDatasetWeighted`	Chain multiple datasets for sequential iteration without concatenation.

Classes#

`ConcatDatasetWeighted(datasets: list[DatasetWeightedType], labels: list[str] | None = None)` #

Bases: Dataset

Concatenate multiple DatasetWeighted instances into a single dataset.

Combines multiple datasets sequentially, maintaining weights and targets across all datasets. Useful for combining data from different sources or experimental conditions.

PARAMETER	DESCRIPTION
`datasets`	List of datasets to concatenate. TYPE: `list[DatasetWeightedType]`
`labels`	Feature labels for concatenated dataset. If None, uses labels from first dataset. TYPE: `list \| None, optional, by default None` DEFAULT: `None`

Properties

datasets : list[DatasetWeightedType] List of component datasets.

cumulative_sizes : list[int] Cumulative sizes for efficient indexing.

labels : list[str] Feature labels.

Examples:

Concatenate datasets:

>>> import torch
>>> from spectre.data import DatasetWeighted, ConcatDatasetWeighted
>>> ds1 = DatasetWeighted(torch.randn(100, 10), weights=torch.rand(100))
>>> ds2 = DatasetWeighted(torch.randn(50, 10), weights=torch.rand(50))
>>> concat_ds = ConcatDatasetWeighted([ds1, ds2])
>>> len(concat_ds)
150
>>> batch = concat_ds[0]
>>> batch.data.shape
torch.Size([10])

With targets:

>>> ds1 = DatasetWeighted(torch.randn(100, 10), target=torch.randint(0, 2, (100,)))
>>> ds2 = DatasetWeighted(torch.randn(50, 10), target=torch.randint(0, 2, (50,)))
>>> concat_ds = ConcatDatasetWeighted([ds1, ds2])
>>> batch = concat_ds[0]
>>> batch.has_target()
True

Notes

All datasets must have the same number of features
All datasets must have targets or none should have targets
Labels consistency is checked across datasets

Source code in spectre/data/dataset_utils.py

def __init__(
    self,
    datasets: list[DatasetWeightedType],
    labels: list[str] | None = None,
) -> None:
    if not datasets:
        raise ValueError("At least one dataset is required for concatenation.")

    if not all(isinstance(ds, DatasetWeightedType) for ds in datasets):
        raise TypeError(
            "Datasets must be of type `DatasetWeighted` or "
            "`DatasetWeightedMemoryMapped`."
        )

    # Validate feature dimensions
    n_features = datasets[0].data.shape[1]
    for i, ds in enumerate(datasets[1:], start=1):
        if ds.data.shape[1] != n_features:
            raise ValueError(
                f"Dataset {i} has {ds.data.shape[1]} features, expected {n_features}."
            )

    # Validate target consistency
    has_target = datasets[0].has_target()
    for i, ds in enumerate(datasets[1:], start=1):
        if ds.has_target() != has_target:
            raise ValueError(
                f"Dataset {i} target status inconsistent with first dataset."
            )

    self.datasets = datasets

    # Calculate cumulative sizes for efficient indexing
    self.cumulative_sizes = _cumsum(datasets)

    if labels is not None:
        if not (
            isinstance(labels, list)
            and all(isinstance(str_l, str) for str_l in labels)
        ):
            raise TypeError("Labels must be provided as a list of strings.")
        if len(labels) != n_features:
            raise ValueError(
                f"Provided labels length ({len(labels)}) does not match "
                f"number of features ({n_features})."
            )
        self.labels = labels
    else:
        self.labels = datasets[0].labels

`ChainDatasetWeighted(datasets: list[DatasetWeightedType])` #

Bases: Dataset

Chain multiple datasets for sequential iteration without concatenation.

Unlike ConcatDatasetWeighted, ChainDatasetWeighted does not validate feature consistency and allows iteration over heterogeneous datasets. Useful for streaming data from multiple sources.

PARAMETER	DESCRIPTION
`datasets`	List of datasets to chain. TYPE: `list[DatasetWeightedType]`