o %ݫi~@sdZddlmZddlmZddlmZddlZddl Z ddl m Z ddl m Z mZmZmZddlmZdd lmZeeZGd d d eZGd d d eZGdddeZGdddeZGddde ZGdddeZdS)zPyTorch compatible samplers. These determine the order of iteration through a dataset. Authors: * Aku Rouhe 2020 * Samuele Cornell 2020 * Ralf Leibold 2020 * Artem Ploujnikov 2021 * Andreas Nautsch 2021, 2023 * Adel Moumen 2023 )Counter) itemgetter)ListN)lognorm)DistributedSampler RandomSamplerSamplerWeightedRandomSampler)DynamicItemDataset) get_loggercs6eZdZdZd fdd ZddZfdd ZZS) ReproducibleRandomSamplera A modification of RandomSampler which always returns the same values. Also look at `torch.utils.data.RandomSampler`. This has mostly the same behaviour and arguments, except for adding 'seed' and 'epoch' and not supporting 'generator'. Note ---- Call `set_epoch` before every epoch. Otherwise, the sampler will produce the same sequence of indices every epoch. Arguments --------- data_source : Dataset The data source to sample indices for. seed : int The base seed to use for the random number generator. It is recommended to use a value which has a good mix of 0 and 1 bits. epoch : int The epoch to start at. **kwargs : dict Arguments to pass to parent class. Example ------- >>> import torch >>> from speechbrain.utils.checkpoints import Checkpointer >>> from speechbrain.dataio.dataloader import SaveableDataLoader >>> # An example "dataset" >>> dataset = torch.arange(10).unsqueeze(1) >>> # Create the random sampler: >>> sampler = ReproducibleRandomSampler(dataset) >>> dataloader = SaveableDataLoader(dataset, sampler = sampler, ... num_workers = 3) >>> # Setup the checkpointer. >>> # Note that the sampler doesn't need to be saved itself. >>> tmpdir = getfixture('tmpdir') >>> checkpointer = Checkpointer(tmpdir, {"dataloader": dataloader}) >>> # Iterate: >>> subset = [] >>> for i, data_point in enumerate(dataloader): ... # Say you save a checkpoint on the fourth batch: ... if i == 3: ... _ = checkpointer.save_checkpoint(end_of_epoch = False) ... # So let's save the numbers you would get if you continue ... if i >= 4: ... subset.append(data_point.item()) >>> # What if instead you had to restart the experiment? >>> new_sampler = ReproducibleRandomSampler(dataset) >>> new_dataloader = SaveableDataLoader(dataset, sampler = new_sampler, ... num_workers = 3) >>> new_checkpointer = Checkpointer(tmpdir, {"dataloader": new_dataloader}) >>> _ = new_checkpointer.recover_if_possible() >>> # You'll get the same random order again: >>> new_subset = [data_point.item() for data_point in new_dataloader] >>> assert subset == new_subset &l!rc sFd|vr d}t|tj|fi|t||_||_t|_dSN generatorzECannot give a separate generator when using ReproducibleRandomSampler ValueErrorsuper__init__intseedepochtorch Generatorr)self data_sourcerrkwargsMSG __class__N/home/ubuntu/.local/lib/python3.10/site-packages/speechbrain/dataio/sampler.pyr^s z"ReproducibleRandomSampler.__init__cC ||_dSz You can also just access self.epoch, but we maintain this interface to mirror torch.utils.data.distributed.DistributedSampler Nrrrrrr set_epochj z#ReproducibleRandomSampler.set_epochc|j|j|jtSNr manual_seedrrr__iter__rrrr r+q z"ReproducibleRandomSampler.__iter__)r r__name__ __module__ __qualname____doc__rr%r+ __classcell__rrrr r "s ; r cs:eZdZdZ  d fdd ZddZfdd ZZS) !ReproducibleWeightedRandomSamplera;A reproducible modification of WeightedRandomSampler. Also look at `torch.utils.data.WeightedRandomSampler`. This has the the same behaviour and arguments, except for adding 'seed' and 'epoch' and not supporting 'generator'. Note ---- Call `set_epoch` before every epoch. Otherwise, the sampler will produce the same sequence of indices every epoch. Arguments --------- weights : sequence of float Weights for each index. Doesn't need to sum to one. num_samples : int Number of samples to draw replacement : bool To draw with replacement or not (within an epoch of num_samples). seed : int The base seed to use for the random number generator. It is recommended to use a value which has a good mix of 0 and 1 bits. epoch : int The epoch to start at. **kwargs : dict Arguments to pass to parent class. Example ------- >>> a = ReproducibleWeightedRandomSampler([0.1, 0.9, 0.4, 0.7, 3.0, 0.6], 5, replacement=True) >>> b = ReproducibleWeightedRandomSampler([0.1, 0.9, 0.4, 0.7, 3.0, 0.6], 5, replacement=True) >>> list(a) [3, 1, 4, 4, 4] >>> list(b) [3, 1, 4, 4, 4] >>> a.set_epoch(1) >>> list(a) [4, 5, 4, 4, 3] >>> b.set_epoch(1) >>> list(b) [4, 5, 4, 4, 3] rc sJd|vr d}t|tj|||fi|t||_||_t|_dSrr)rweights num_samples replacementrrrrrrr rs  z*ReproducibleWeightedRandomSampler.__init__cCr!r"r#r$rrr r%r&z+ReproducibleWeightedRandomSampler.set_epochcr'r(r)r,rrr r+r-z*ReproducibleWeightedRandomSampler.__iter__)r5rr.rrrr r4vs2r4c@sHeZdZdZddeefddfddZdd Zd d Zd d Z ddZ dS)ConcatDatasetBatchSampleraThis sampler is built to work with a standard Pytorch ConcatDataset. It is used to retrieve elements from the different concatenated datasets placing them in the same batch with proportion specified by batch_sizes, e.g 8, 16 means each batch will be of 24 elements with the first 8 belonging to the first dataset in ConcatDataset object and the last 16 to the second. More than two datasets are supported, in that case you need to provide 3 batch sizes. Note ---- Batched are drawn from the datasets till the one with smallest length is exhausted. Thus number of examples in your training epoch is dictated by the dataset whose length is the smallest. Arguments --------- samplers : list or tuple a list or tuple of pytorch samplers batch_sizes: list Batch sizes. epoch : int The epoch to start at. Example ------- >>> import torch >>> from speechbrain.dataio.sampler import ConcatDatasetBatchSampler, ReproducibleRandomSampler >>> from speechbrain.dataio.sampler import ReproducibleRandomSampler >>> from speechbrain.dataio.dataloader import SaveableDataLoader >>> # example "datasets" >>> dataset1 = torch.arange(0, 10).unsqueeze(1) >>> dataset2 = torch.arange(20, 40).unsqueeze(1) >>> tot_dataset = torch.utils.data.ConcatDataset([dataset1, dataset2]) >>> sampler1 = ReproducibleRandomSampler(dataset1) >>> sampler2 = ReproducibleRandomSampler(dataset2) >>> tot_sampler = ConcatDatasetBatchSampler([sampler1, sampler2], [2, 4]) >>> dataloader = SaveableDataLoader(tot_dataset, batch_sampler = tot_sampler, ... num_workers = 3) >>> for data_point in dataloader: ... assert len(data_point) == 6 ... for i in range(2): ... assert data_point[i] in [x for x in range(0, 10)] ... for i in range(2, 4): ... assert data_point[i] in [x for x in range(10, 40)] r batch_sizesreturnNcCst|ttfstd|t|ttfstd|t|t|ks(td||_||_dgt dd|jD dd|_ ||_ | |j dS)NzKsamplers should be a list or tuple of Pytorch Samplers, but got samplers={}zIbatch_sizes should be a list or tuple of integers, but got batch_sizes={}z3batch_sizes and samplers should be have same lengthrcSg|]}t|qSr)len.0xrrr z6ConcatDatasetBatchSampler.__init__..) isinstancelisttuplerformatr=r:samplersnpcumsumtolistoffsetsrr%)rrHr:rrrr rs. z"ConcatDatasetBatchSampler.__init__ccs4g}|D]}|||t||kr|VqdSr()appendr=)r c_batch_size c_samplerc_offsetbatchidxrrr _iter_one_datasets z+ConcatDatasetBatchSampler._iter_one_datasetcCs.t|jddr|jD] }||q dSdS)zYou can also just access self.epoch, but we maintain this interface to mirror ``torch.utils.data.distributed.DistributedSampler``. rrN)hasattrrHr%)rrsrrr r%s   z#ConcatDatasetBatchSampler.set_epochccsdd|jD}g}tt|D]8}tt|jD])}g}t||j|kr>||j|t||t||j|ks'||q|Vg}qdS)NcSr<r)iterr?irrr rA"rBz6ConcatDatasetBatchSampler.__iter__..)rHranger=r:rMrLnextextend)r iterators tot_batchb_numsamp_idxc_batchrrr r+ s z"ConcatDatasetBatchSampler.__iter__cCs<td}t|jD]\}}t||j|}t||}q |S)Ninf)float enumeraterHr=r:min)rmin_lenrRsamplerc_lenrrr __len__0s  z!ConcatDatasetBatchSampler.__len__r) r/r0r1r2rFrErrSr%r+rhrrrr r9s0 r9c@seZdZdZddddddgdddd d f d ed ed ed ededeedeededededefddZddZ d ededeefddZ ddZ dd Z d!d"Z d#d$Zd%d&ZdS)'DynamicBatchSamplerarThis BatchSampler batches examples together by grouping them by their length. Every example in the batch have approximately the same length and thus padding is minimized. This enables faster training on datasets where length of examples can vary significantly (e.g Librispeech). Inspired by: https://www.tensorflow.org/api_docs/python/tf/data/experimental/bucket_by_sequence_length Dynamic batching is performed by specifying a max_batch_length which is the upper limit for the sum of the length of examples in a batch: e.g., if ex1 has length 4, ex2 length 5 and if max_batch_length is set to 6 ex1 and ex2 will be placed, alone, in two distinct batches. Length for each example can be obtained in two manners. If the input dataset is a DynamicItemDataset it can be obtained by specifying a length_func. Default assumes a "duration" entry is in the annotation. Length for each example can also be passed to this class upon instantiation by specifying a list containing the length for each example and passing it to lengths_list. Examples are grouped together by defining a set of possible discrete intervals (buckets). Examples whose length fall into these intervals can be batched together. The number of buckets can be specified by using the arg num_buckets. There is usually an optimal range for the value of this argument. If num_buckets == 1, all examples can be batched together. You have maximum randomization but your training speed will be slower due to the fact that a large amount of the values will be padding as long and short examples can be batched together. As the number of buckets grows only examples with similar length can be grouped together. This trades-off speed with randomization. TLDR: Low number -> better randomization, High number -> faster training. NOTE THAT: if set too high the training speed will decrease. If num_buckets -> number of examples in the dataset the batch size will be small impacting training speed and possibly performance. The buckets can also be specified by passing a list to the bucket_boundaries argument instead of specifying a left_bucket_length and a bucket_length_multiplier. Example ------- >>> import torch >>> import speechbrain as sb >>> from speechbrain.dataio.sampler import DynamicBatchSampler >>> from speechbrain.dataio.dataset import DynamicItemDataset >>> from speechbrain.dataio.dataloader import SaveableDataLoader >>> from speechbrain.dataio.batch import PaddedBatch >>> import numpy as np >>> item_lengths = sorted([np.random.randint(10, 100) for x in range(20)]) >>> dataset = {"ex_{}".format(x) : {"wav" :torch.randn(x)} for x in item_lengths} >>> dataset = DynamicItemDataset(dataset) >>> dataset.set_output_keys(["wav"]) >>> length_func = lambda x : len(x) # trivial in this example >>> bsampler = DynamicBatchSampler(dataset, 20, 4, length_func, shuffle=False, batch_ordering='descending') >>> dataloader = SaveableDataLoader(dataset, batch_sampler=bsampler, collate_fn=PaddedBatch) >>> for i, b in enumerate(dataloader): ... data, length = b["wav"] >>> assert data.shape[-1] == max(item_lengths) Arguments --------- dataset : torch.utils.data.Dataset Pytorch Dataset from which elements will be sampled. max_batch_length : int Upper limit for the sum of the length of examples in a batch. Should be chosen based on your GPU memory. num_buckets : int Number of discrete buckets used to group examples together. If num_buckets == 1, all examples can be batched together. As the number of buckets grows only examples with similar length can be grouped together. This trades-off speed with randomization. Low number -> better randomization, High number -> faster training. However if set too high the training speed will decrease. If num_buckets -> number of examples in the dataset the batch size will be small impacting training speed and possibly performance. NOTE: you have either to specify manually the bucket_boundaries or the number of buckets. length_func : callable Function used to get length of each example from the dataset. This argument can be used only when the dataset is a Speechbrain DynamicItemDataset object. Can be anything: e.g. lambda x: x["duration"]*16000 returns number of samples if duration key in the annotation is in seconds and the file has 16kHz sampling freq. shuffle : bool Whether or not shuffle examples between each epoch. batch_ordering : string If ``random``, batches are randomly permuted; otherwise ``ascending`` or ``descending`` sorted by length. max_batch_ex: int If set, it limits the maximum number of examples that can be in a batch superseding max_batch_length in instances where the amount of examples will exceed the value specified here. E.g. you have a lot of short examples and the batch size for those will be too high, you can use this argument to limit the batch size for these short examples. bucket_boundaries : list Overrides bucket_length_multiplier and left_bucket_length by specifying manually the buckets right boundaries. lengths_list: list Overrides length_func by passing a list containing the length of each example in the dataset. This argument must be set when the dataset is a plain Pytorch Dataset object and not a DynamicItemDataset object as length_func cannot be used on Pytorch Datasets. seed : int Random seed. epoch : int The epoch to start at. drop_last : bool If ``True``, the sampler will drop the last examples which have not been grouped. verbose: bool If ``True``, log also the stats for each batch at the first epoch. NcCs|dS)Ndurationrr@rrr szDynamicBatchSampler.Trandom*rFmax_batch_length num_bucketsshufflebatch_ordering max_batch_exbucket_boundaries lengths_listrr drop_lastverbosecs|_i_| _|durt|dkrtd| dur.tt| D] }| |jt|<q!n$t|ts7t dttjD]}|jj jj |jt|<q>t|dkrt dd|Dset dtt|t|ksst dtjjt|tt|dd tt|_n tj||d _|_|_|_| _| _|durtj}|_fd dttjDd g_| _dS) Nrz\Please specify either num_buckets or bucket boundaries.Check the docs, and/or the tutorial !zMDataset should be a Speechbrain DynamicItemDataset when using length functioncSsg|]}|dkqSrirr>rrr rArBz0DynamicBatchSampler.__init__..z@All elements in bucket boundaries should be non-negative (>= 0).z0Bucket_boundaries should not contain duplicates.z[The arg bucket_boundaries should be an ascending sorted list of non negative values values!)err_msg)rp num_quantilesc s.g|]}tjtdtjj|qS)rd _max_batch_exmaxr_max_batch_length_bucket_boundariesrWr,rr rAsr|) _dataset _ex_lengthsrxr= RuntimeErrorrYstrrDr NotImplementedErrordatadata_idsallrsetrItestingassert_array_equalarraysortedr_get_boundaries_through_warpingr _shuffle_ex_batch_ordering_seed _drop_lastrar} _bucket_lens_epoch_generate_batches)rdatasetrprq length_funcrrrsrtrurvrrrwrxindxrr,r rsn       zDynamicBatchSampler.__init__csfdd|DS)z,Gets durations of the elements in the batch.cg|] }jt|qSrrrr?rRr,rr rAz5DynamicBatchSampler.get_durations..rrrQrr,r get_durationssz!DynamicBatchSampler.get_durationsrzr;c std|d}td||||}t|d}|||dfddt|dD}tdt t djt t dj|t t S)Nz"Batch quantisation in latent spacer|rCcs g|] }|d|qSr{rr>rurr rAzGDynamicBatchSampler._get_boundaries_through_warping..z=Latent bucket boundary - buckets: {} - length multipliers: {}z{:.2f}) loggerinforIlinspacerppfrYdebugrGrEmapr)rrprznum_boundarieslatent_boundaries quantileslength_multipliersrrr rs&     z3DynamicBatchSampler._get_boundaries_through_warpingcsjdkr2t}|jjtjtj|d }g}|D] }| j|q"|_dSjdkrEt jfddd_dSjdkrYt jfddd d _dSt ) Nrnr ascendingctfdd|DS)Ncrrrrr,rr rA6rJDynamicBatchSampler._permute_batches....r~rlr,rr rm6z6DynamicBatchSampler._permute_batches..key descendingcr)Ncrrrrr,rr rA;rrrrlr,rr rm;rT)rreverse) rrrr*rrrandpermr=_batchesrKrMrr)rgrftmprRrr,r _permute_batches%s0        z$DynamicBatchSampler._permute_batchesc s<tdjr"t}|jjtjt j |d }nt t j }g_ ddjD}ddjD}|D]e}jt|}tj|}|||t||d|||d<t||d|||d<||d|7<||d d 7<t ||j|kst ||jkrj ||g||<q>js|D] }|rj |qjd krd gj } t t jD]P} z#|| dj} || d|| d|| d|| d } Wn tyd } d } Ynwtd | | | | | d j| || d | | d q͈jrgggd} j D]>}tfdd|D}| d|tfdd|Dtfdd|D}| d|| d||d q+d}d|}t t j D]$}t||| d|t j || d|| d|qwdSdSdS)Nz/DynamicBatchSampler: Generating dynamic batchesrcSsg|]}gqSrrrWrrr rAMsz9DynamicBatchSampler._generate_batches..cSs g|] }tjtj dddqS)r)rdr~totn_ex)rIrarWrrr rAOrrdr~rrr|rzDynamicBatchSampler: Bucket {} with boundary {:.1f}-{:.1f} and batch_size {}: Num Examples {:.1f}, Num Full Batches {:.3f}, Pad Factor {:.3f}.d) tot_framestot_pad_framespad_%crrrrr,rr rArrcrrrrr,rr rArcsg|] }jt|qSrrr max_framesrrr rAsrrzPBatch {} with {:.1f} frames with {} files - {:.1f} padding, {:.2f} (%) of total.zDynamicBatchSampler: ) rrrrrr*rrrr=rrKrYrrrrrI searchsortedrrMrdr~r}rrrZeroDivisionErrorrrGrxsum)rrrfbucket_batches stats_trackerrRitem_len bucket_idrQ boundaries bucket_indx num_batches pad_factor batch_statsrtot_padpadding_detailsrXrrr rAs                     #z%DynamicBatchSampler._generate_batchesccs<|jD]}|Vq|jr||jdkr|dSdS)Nrn)rrrrrrrrr r+s   zDynamicBatchSampler.__iter__cCs||_|dSr")rrr$rrr r%s zDynamicBatchSampler.set_epochcCs t|jSr()r=rr,rrr rhs zDynamicBatchSampler.__len__)r/r0r1r2rboolrrrrrrrr+r%rhrrrr rj:sbo      X #} rjcs8eZdZdZfddZfddZfddZZS)DistributedSamplerWrapperaThis wrapper allows using any sampler (for example batch) with Distributed Data Parallel (DDP) correctly. Passing blindly the sampler to each DDP process will cause to have access within each process to all the data in the dataset instead of only a subset of it which is unique to each process. This wrapper prevents this and allows to use only a subset of the original data for each process. NOTE ---- This is is automatically applied to any sampler in the Brain class when DDP training is used. cs tj|d|i|||_dS)Nr)rrrf)rrfargsrrrr rs z"DistributedSamplerWrapper.__init__cs(t|j}t}tt||Sr()rErfr+rrVr)rsampler_indicesindices_of_indicesrrr r+s z"DistributedSamplerWrapper.__iter__cs,t|t|jdr|j|dSdS)zBPass set_epoch() through to DistributedSampler and the wrapper oner%N)rr%rTrfr$rrr r%s  z#DistributedSamplerWrapper.set_epoch)r/r0r1r2rr+r%r3rrrr rs    rcs2eZdZdZ    d fdd Zdd ZZS) BalancingDataSampleraA data sampler that takes a single key from the dataset and ensures an approximately equal distribution by that key Arguments --------- dataset : DynamicItemDataset the dataset form which samples will be drawn key : str the key from which samples will be taken num_samples : int Number of samples to draw replacement : bool To draw with replacement or not (within an epoch of num_samples). seed : int The base seed to use for the random number generator. It is recommended to use a value which has a good mix of 0 and 1 bits. epoch : int The epoch to start at. **kwargs : dict Arguments to pass to parent class. Example ------- >>> from speechbrain.dataio.sampler import BalancingDataSampler >>> from speechbrain.dataio.dataset import DynamicItemDataset >>> sample_data = { ... 1: {"category": "A", ... "text": "This is a test"}, ... 2: {"category": "A", ... "text": "This is a second test"}, ... 3: {"category": "B", ... "text": "This is a third test"} ... } >>> dataset = DynamicItemDataset(data=sample_data) >>> sampler = BalancingDataSampler( ... dataset=dataset, ... key="category", ... num_samples=10 ... ) >>> sampler.weights tensor([0.5000, 0.5000, 1.0000], dtype=torch.float64) >>> it = iter(sampler) >>> [next(it) for _ in range(10)] [2, 2, 1, 2, 2, 0, 1, 1, 1, 2] NTr rc  s@||_||_|s t|}|}tj|||||fi|dSr()rrr=_compute_weightsrr) rrrr7r8rrrr6rrr r*s   zBalancingDataSampler.__init__csljjgfddjD}t|Wdn1s!wYdtfdd|D}|S)Ncsg|]}|jqSrr)r?itemr,rr rA?rz9BalancingDataSampler._compute_weights..r|csg|]}|qSrr)r?class_id) class_counterrr rABrB)routput_keys_asrrrtensor)r class_idsr6r)rrr r=s z%BalancingDataSampler._compute_weights)NTr r)r/r0r1r2rrr3rrrr rs2r)r2 collectionsroperatorrtypingrnumpyrIr scipy.statsrtorch.utils.datarrrr speechbrain.dataio.datasetr speechbrain.utils.loggerr r/rr r4r9rjrrrrrr s(     TNv&