o Si! @sddlZddlZddlZddlZddlmZmZddlmZddl m Z ddl m Z m Z mZmZmZmZmZmZmZmZmZmZddlZddlZddlmZmZddlmZddl m!Z!dd l"m#Z#m$Z$m%Z%m&Z&m'Z'dd l(m)Z)m*Z*m+Z+dd l,m-Z-Gd d d e#Z.eGddde%Z/ d$deede0dee%dee1fddZ2GdddZ3GdddZ4de dej5de6deeeeedffddffddZ7Gd d!d!e8Z9d"d#Z:dS)%N)asdict dataclass)isliceQueue) AnyCallableDequeDict GeneratorIterableListLiteralOptionalSequenceTupleUnion)CutSetSeconds)Cut) resolve_seed) CutSamplerEpochDiagnosticsSamplingConstraintSamplingDiagnosticsTimeConstraint)DurationBatcherFiltercheck_constraint)ifnonec%sdeZdZdZddddddddddddddddddd d ed eed eed eedeede de de de edededeedeedeede ee dfde de ddf$fddZ deeefffdd Zd!eeefddffd"d# Zd$d%Zd0d&d'Zde eeeffd(d)Zedeefd*d+Zedeefd,d-Zedeefd.d/ZZS)1DynamicBucketingSampleraB A dynamic (streaming) variant of :class:`~lhotse.dataset.sampling.bucketing.BucketingSampler`, that doesn't require reading the whole cut set into memory. The basic idea is to sample N (e.g. ~10k) cuts and estimate the boundary durations for buckets. Then, we maintain a buffer of M cuts (stored separately in K buckets) and every time we sample a batch, we consume the input cut iterable for the same amount of cuts. The memory consumption is limited by M at all times. For scenarios such as ASR, VAD, Speaker ID, or TTS training, this class supports single CutSet iteration. Example:: >>> cuts = CutSet(...) >>> sampler = DynamicBucketingSampler(cuts, max_duration=100) >>> for batch in sampler: ... assert isinstance(batch, CutSet) For other scenarios that require pairs (or triplets, etc.) of utterances, this class supports zipping multiple CutSets together. Such scenarios could be voice conversion, speech translation, contrastive self-supervised training, etc. Example:: >>> source_cuts = CutSet(...) >>> target_cuts = CutSet(...) >>> sampler = DynamicBucketingSampler(source_cuts, target_cuts, max_duration=100) >>> for batch in sampler: ... assert isinstance(batch, tuple) ... assert len(batch) == 2 ... assert isinstance(batch[0], CutSet) ... assert isinstance(batch[1], CutSet) .. note:: for cut pairs, triplets, etc. the user is responsible for ensuring that the CutSets are all sorted so that when iterated over sequentially, the items are matched. We take care of preserving the right ordering internally, e.g., when shuffling. By default, we check that the cut IDs are matching, but that can be disabled. .. caution:: when using :meth:`DynamicBucketingSampler.filter` to filter some cuts with more than one CutSet to sample from, we sample one cut from every CutSet, and expect that all of the cuts satisfy the predicate -- otherwise, they are all discarded from being sampled. N FT'i Nr) max_durationmax_cuts constraint num_bucketsshuffle drop_lastconsistent_ids duration_binsnum_cuts_for_bins_estimate buffer_sizequadratic_duration world_sizerankseed sync_buckets concurrentstrictshuffle_buffer_sizecutsr#r$r%r&r'r(r)r*r+r,r-r.r/r0) randomizedtrngr1r2returncs*tj|| | |dtdd|Dstd||_||_||_||_||_ ||_ | |_ | |_ | |_ ||_||_d|_t||||durMtjdtd|dur[t|j |7_ |durwt|t|kskJd||_t|d |_dS|durt|j|j|j d }tt|jd | ||d |_dS) a- :param cuts: one or more CutSets (when more than one, will yield tuples of CutSets as mini-batches) :param max_duration: The maximum total recording duration from ``cuts``. Note: with multiple CutSets, ``max_duration`` constraint applies only to the first CutSet. :param max_cuts: The maximum total number of ``cuts`` per batch. When only ``max_duration`` is specified, this sampler yields static batch sizes. :param num_buckets: how many buckets to create. Ignored if duration_bins are provided. :param shuffle: When ``True``, the cuts will be shuffled dynamically with a reservoir-sampling-based algorithm. Convenient when mini-batch loop is inside an outer epoch-level loop, e.g.: `for epoch in range(10): for batch in dataset: ...` as every epoch will see a different cuts order. :param drop_last: When ``True``, we will drop all incomplete batches. A batch is considered incomplete if it depleted a bucket before hitting the constraint such as max_duration, max_cuts, etc. :param consistent_ids: Only affects processing of multiple CutSets. When ``True``, at each sampling step we check cuts from all CutSets have the same ID (i.e., the first cut from every CutSet should have the same ID, same for the second, third, etc.). :param duration_bins: A list of floats (seconds); when provided, we'll skip the initial estimation of bucket duration bins (useful to speed-up the launching of experiments). :param num_cuts_for_bins_estimate: We will draw this many cuts to estimate the duration bins for creating similar-duration buckets. Larger number means a better estimate to the data distribution, possibly at a longer init cost. :param buffer_size: How many cuts (or cut pairs, triplets) we hold at any time across all of the buckets. Increasing ``max_duration`` (batch_size) or ``num_buckets`` might require increasing this number. Larger number here will also improve shuffling capabilities. It will result in larger memory usage. :param quadratic_duration: When set, it adds an extra penalty that's quadratic in size w.r.t. a cuts duration. This helps get a more even GPU utilization across different input lengths when models have quadratic input complexity. Set between 15 and 40 for transformers. :param sync_buckets: When set, we'll try to make each DDP rank sample from as close duration buckets as possible to minimize the tail worker effect. :param concurrent: Enabling concurrency eliminates most of the waiting to pre-populate the bucketing buffers before the sampler starts yielding examples. For tarred/Lhotse Shar data this can speed up the start of the training. Note that enabling concurrency will cause the sampling results to be non-deterministic. This feature is experimental. :param world_size: Total number of distributed nodes. We will try to infer it by default. :param rank: Index of distributed node. We will try to infer it by default. :param seed: Random seed used to consistently shuffle the dataset across different processes. )r(r.r/r0css |] }t|tr|jVqdSN) isinstanceris_lazy.0csr?]/home/ubuntu/.local/lib/python3.10/site-packages/lhotse/dataset/sampling/dynamic_bucketing.py sz3DynamicBucketingSampler.__init__..zYou are using DynamicBucketingSampler with an eagerly read CutSet. You won't see any memory/speed benefits with that setup. Either use 'CutSet.from_jsonl_lazy' to read the CutSet lazily, or use a BucketingSampler instead.Nz|In Lhotse v1.4 all samplers act as if 'strict=True'. Sampler's argument 'strict' will be removed in a future Lhotse release.categoryz)Duration bins must be sorted ascendingly.r#r$r-r)r&r%)super__init__allwarningswarnr5r#r$r%r'r)r+r,r-r1r2rngrDeprecationWarning!_emit_shuffle_buffer_size_warninglistsortedr*lenr&restimate_duration_bucketsr)selfr#r$r%r&r'r(r)r*r+r,r-r.r/r0r1r2r3r4r5 __class__r?r@rGQs^?  z DynamicBucketingSampler.__init__c sD|jdus Jdt}||j|j|j|j|j|j d|S)NzIstate_dict() is not supported with samplers that use a custom constraint.)r#r$r)r,r+r-) r%rF state_dictupdater#r$r)r,r+r-)rRsdrSr?r@rUs   z"DynamicBucketingSampler.state_dictrWcs|d|_|d|_|d|_|d|_|d|_d|vr1t|d}|j|7_|dd|_|ddt || dS) Nr#r$r)r+r,r4r-r3) popr#r$r)r+r,rMr-rFload_state_dict _fast_forward)rRrWr4rSr?r@rYs         z'DynamicBucketingSampler.load_state_dictcCs\|jj}|jjj}||t|d|jj|<d|_t|t |D]}t |q"d|_dS)N)epochFT) diagnostics current_epochcurrent_epoch_stats total_batches set_epochrstats_per_epoch_just_restored_stateiterrangenext)rRr]num_batches_to_iter_r?r?r@rZs     z%DynamicBucketingSampler._fast_forwardcsjrStj}t|j_jr-d}tj j }|dur'||j 7}t|}nd}j ddjD}tt|fddj d}t|jjjjjjjjjj|jj d}t|_S)NicSsg|]}t|qSr?)rcr<r?r?r@ z4DynamicBucketingSampler.__iter__..cstfdd|DS)Nc3s|]}|VqdSr9) _filter_fnr=crRr?r@rAzEDynamicBucketingSampler.__iter__....)rH)tplrmr?r@z2DynamicBucketingSampler.__iter__..)iterator predicater\) r*r.r#r$r%r(r,r-r'rK bucket_rngr2r\)rbrr0randomRandomr[rKr1torchutilsdataget_worker_infoidr\reset_current_epochr5rzipDynamicBucketerr*r.r#r$r%r(r,r-r'r2rc cuts_iter)rRr0bucket_rng_seed worker_infortrr?rmr@__iter__sH       z DynamicBucketingSampler.__iter__cstt|j}|jr8t|tr8t|D]&}|djtfdd|ddDs7Jdddd|Ddq|S) Nrc3s|]}|jkVqdSr9r{rk expected_idr?r@rA/rnz6DynamicBucketingSampler._next_batch..rDzhThe input CutSet are not sorted by cut ID in the same way. We sampled the following mismatched cut IDs: z, css|]}|jVqdSr9rrkr?r?r@rA1sz[. If this is expected, pass the argument 'consistent_ids=False' to DynamicBucketingSampler.) rerr)r:tupler}r{rHjoin)rRbatchr5r?rr@ _next_batch*s    z#DynamicBucketingSampler._next_batchcCdSr9r?rmr?r?r@remaining_duration6z*DynamicBucketingSampler.remaining_durationcCrr9r?rmr?r?r@remaining_cuts:rz&DynamicBucketingSampler.remaining_cutscCrr9r?rmr?r?r@num_cuts>rz DynamicBucketingSampler.num_cuts)r8r )__name__ __module__ __qualname____doc__r rrintrboolr rrrGr strrrUrYrZrrrrpropertyfloatrrr __classcell__r?r?rSr@r (s+     t 4 r c@seZdZUdZeeed<eeed<dZe edfed<dZ eed<dd Z d e fd d Z d ed dfddZd e fddZd e fddZd#ddZd ed efddZd eeeffddZdeeefd dfddZ  d$ddZdd d e fd!d"ZdS)%FixedBucketBatchSizeConstrainta Sampling constraint that accepts a pre-defined batch size for each bucket. It uses the example's sequence length to determine which bucket we're sampling for, and otherwise the batch size is locally static for each bucket. This constraint doesn't support samples longer than the upper bound of the last bucket; if such sample is provided, we will raise an exception. max_seq_len_buckets batch_sizesNcurrent_bucketrrcCst|jt|jks JdSr9)rOrrNrmr?r?r@ __post_init__Ssz,FixedBucketBatchSizeConstraint.__post_init__r8cCsdSNTr?rmr?r?r@ is_activeVsz(FixedBucketBatchSizeConstraint.is_activeexamplecCs||}|j|j|d}|t|jks"Jd|d|jdd|jdur+||_n|j|ks?Jd|d|d |jd|jd 7_dS) z Increment the internal counter for the time constraint, selecting the right property from the input ``cut`` object. )buckets example_lenz&Received example with sequence length z) that exceeds the highest allowed length .NzUser error: FixedBucketBatchSizeConstraint is supposed to be used only on one bucket. The example we received has sequence length z9 which is outside of the allowed bounds for bucket index z in buckets rD)measure_length select_bucketrrPrr)rRrseqlen bucket_idxr?r?r@addYs*   z"FixedBucketBatchSizeConstraint.addcCs|j|j|jkS)z"Is the constraint exceeded or not.rrrrmr?r?r@exceededpsz'FixedBucketBatchSizeConstraint.exceededcCs|j|j|jkS)a( Check if the batch is close to satisfying the constraints. We define "closeness" as: if we added one more cut that has duration/num_frames/num_samples equal to the longest seen cut in the current batch, then the batch would have exceeded the constraints. rrmr?r?r@close_to_exceedingtsz1FixedBucketBatchSizeConstraint.close_to_exceedingcCsd|_d|_dS)z{ Reset the internal counter (to be used after a batch was created, to start collecting a new one). Nr)rrrmr?r?r@reset}s z$FixedBucketBatchSizeConstraint.resetcCs|jSr9)duration)rRrr?r?r@rsz-FixedBucketBatchSizeConstraint.measure_lengthcCst|Sr9)rrmr?r?r@rUsz)FixedBucketBatchSizeConstraint.state_dictrUcCsV|d|_|d|_|d|_|d|_t|dks)Jdd|dS)NrrrrrzNError in FixedBucketBatchSizeConstraint.load_state_dict(): Unexpected keys: - z - )rXrrrrrPrkeys)rRrUr?r?r@rYs     z.FixedBucketBatchSizeConstraint.load_state_dictotherc CszdD]+}t||}t||}|duo|du}|s-||ks-Jd|d|d|d|d qt|j|j|j|j|jdS)N)rrrzXTo add two TimeConstraint objects, they need to represent the same constraint (got self.=z != other.).)rrrr)getattrrrrrr)rRrkey self_attr other_attris_noner?r?r@__add__s,   z&FixedBucketBatchSizeConstraint.__add__rcCs.t|to|j|jko|j|jko|j|jkSr9)r:rrrr)rRrr?r?r@__eq__s    z%FixedBucketBatchSizeConstraint.__eq__)r8N)rrr8r)rrrrr r__annotations__rrrrrrrrrrrrrr rrrUrYrrr?r?r?r@rCs*      rr5r&r%r8cs|dksJdur ttfdd|D}|||jdks2Jd|d|jdd||}g}d }|D]}||krK||d }||7}q>|S) a  Given an iterable of cuts and a desired number of buckets, select duration values that should start each bucket. The returned list, ``bins``, has ``num_buckets - 1`` elements. The first bucket should contain cuts with duration ``0 <= d < bins[0]``; the last bucket should contain cuts with duration ``bins[-1] <= d < float("inf")``, ``i``-th bucket should contain cuts with duration ``bins[i - 1] <= d < bins[i]``. :param cuts: an iterable of :class:`lhotse.cut.Cut`. :param num_buckets: desired number of buckets. :param constraint: object with ``.measure_length()`` method that's used to determine the size of each sample. If ``None``, we'll use ``TimeConstraint``. :return: a list of boundary duration values (floats). rDNcsg|]}|qSr?)rrkr%r?r@rhrqz-estimate_duration_buckets..rzThe number of buckets (z7) must be smaller than or equal to the number of cuts (rg)rnparraysortshapesumappend)r5r&r%sizessize_per_bucketbinstotsizer?rr@rQs&    rQc@sjeZdZdZdejdededdfddZdefd d Zde e e ffd d Z d e e e fddfddZ dS)BucketSelectionStatea Helper class used in the context of bucket selection synchronization across DDP ranks. It's only necessary when using a map-style dataset (i.e., the sampler lives in the training loop process) and world_size is greater than 1. In these cases we have to use the same bucket idx ``world_size`` times to ensure each rank uses the same bucket. This is due to how CutSampler distributes mini-batches across ranks, ensuring the number of steps is always equal for each rank. rtr&r.r8NcCs"||_||_||_d|_d|_dSNr) _bucket_rng _num_buckets _world_size _usage_count _bucket_idx)rRrtr&r.r?r?r@rGs  zBucketSelectionState.__init__cCs@|jdus |j|jkr|j|j|_d|_|jd7_|jS)NrrD)rrrr randrangerrmr?r?r@select_bucket_idxs z&BucketSelectionState.select_bucket_idxcCs|j|j|jdS)N)rrr)rgetstaterrrmr?r?r@saveszBucketSelectionState.saveckptcCs(|j|d|d|_|d|_dS)Nrrr)rsetstaterr)rRrr?r?r@restores zBucketSelectionState.restore)rrrrrurvrrGrr rrrrr?r?r?r@rs  rc@seZdZ           d&deeeeefdeede de ede e d e e d e d e d e ed e de jde jde de eddfddZdeeddffddZdedefddZdede fddZddZdd Zd!e ddfd"d#Zd$d%ZdS)'r~NFr"r5r*r.r#r$r%r(r,r-r'rKrtr2r\r8cCs*||_||_||_||_||_||_||_||_| |_t |t |_ | dur*t } | |_| |_| |_| |_|t|ksDJd|dt||||jdurZt|j|j|jd|_|durt|}||}|t|d}||krtd|d|ddd tt|dD|_d|_d |_dS) NzJArgument list for 'duration_bins' is expected to be in sorted order (got: rrErDzYour 'buffer_size' setting of z1 might be too low to satisfy a 'max_duration' of z (given our best guess).cSsg|]}tqSr?r)r=rgr?r?r@rh@sz,DynamicBucketer.__init__..F)r5r*r.r#r$r%r(r,r-rrr\rurvrKrtr'r2rOrrrmeanrPrIrJrdr_producer_thread_source_exhausted)rRr5r*r.r#r$r%r(r,r-r'rKrtr2r\ mean_durationexpected_buffer_durationexpected_bucket_durationr?r?r@rGsV     zDynamicBucketer.__init__c cst|j|_|jrd|_||n||jt |j t |j |j d}zz ||}|}g}|jr@t||j|d}n|j t|j}Wdn1sSwYt||j|jd}tt|}t|trut |d}nt |}|V|r|jdd|j|j}|D]} || =qWdn1swYn t|D]} |q|jr|n||q+tyYnwW|jr|j !rd|_|j "d|_ d|_dS|jr|j !rd|_|j "d|_ d|_w)NF)rtr&r.T)rKout_indexes_used)r%r\r)reverse)#rcr5rr2r_start_data_producer_thread_maybe_wait_for_producer_collect_cuts_in_bucketsr,rrtrPrr._select_bucketr'pick_at_randomrKmutexrNqueuerr%copyr\rer:rrrdget StopIterationris_aliver) rRstatesampling_bucketmaybe_shuffled indexes_usedbatcherr batch_size_qidxrgr?r?r@rEs~             )   zDynamicBucketer.__iter__rc sjdur-fddjD}|s'ddjD}js"t|dkr%t|}j|Sdttgt fdt ffdd } }z|j }Wn't ypjrVtz ||d d }Wn t ymtwYnwj|S) Ncsg|] }|r|qSr?) _is_readyr=brmr?r@rhsz2DynamicBucketer._select_bucket..cSsg|]}|r|qSr?r?rr?r?r@rhrirrsr8csdtffdd }d}}}|r|jsS|dkr-|tjkr-t|d7}|ddkr:dnd|t|}t|}|r|jrS)Nr8csdko tjkSSr)rPrr?)rrRr?r@ valid_idxszGDynamicBucketer._select_bucket..scan_buckets..valid_idxrrDr)rrrrPBucketsDontHaveEnoughDataminmax)rsr num_attemptsseen_minseen_maxrRr)rr@ scan_bucketss    z4DynamicBucketer._select_bucket..scan_bucketscSs |dkSrqsize)rr?r?r@rps z0DynamicBucketer._select_bucket..)rtrr(rPrrKchoicerrrrrrrr)rRr ready_bucketsnon_empty_bucketsrrselected_bucket_idxr?rr@rs0  "     zDynamicBucketer._select_bucketbucketcCsr|j}|j t|j}Wdn1swY|D]}|t|tr,|dn||r6dSqdS)NrTF) r%rrrNrrr:rr)rRrrcontentsrlr?r?r@rs  zDynamicBucketer._is_readycs(fdd}tj|d_jdS)zEStart concurrent filling of the bucket buffer in a background thread.csz>js.producer..g?rrrT)rrrr,timesleeprerr%rr*r:rputr)r5rrmr?r@producers   z=DynamicBucketer._start_data_producer_thread..producer)targetN) threadingThreadrstart)rRr r?rmr@rs z+DynamicBucketer._start_data_producer_threadcCsbtdd|jD|jdkr+|js/tdtdd|jD|jdkr-|jrdSdSdSdS)zLTriggers wait for producer if the bucket buffers are less than 10% utilized.csrr9rrr?r?r@rArz;DynamicBucketer._maybe_wait_for_producer..r!g?N)rrr,rrr rmr?r?r@rs z(DynamicBucketer._maybe_wait_for_producern_cutscCshz)t|D]!}t|j}|jj|jt|tr|dn|d}|j| |qWdSt y3YdSw)zIFetches ``n_cuts`` from the input data iterable. Doesn't use concurrency.rrN) rdrerr%rr*r:rrr r)rRrrgr5rr?r?r@rs   z(DynamicBucketer._collect_cuts_in_bucketscCs:|jr|jdur|jrd|_|jdSdSdSdSr)r2rrrrrmr?r?r@__del__s zDynamicBucketer.__del__) NNNFr"NFNNFN)rrrr rrrr rrrrrrurvrrGr rrrrrrrrrrr?r?r?r@r~sd      BJN  r~rrKr.ccsn|j t|j}Wdn1swYttt|}|||D] }||||Vq(dS)z Generator which will yield items in a sequence in a random order. It will append the indexes of items yielded during iteration via ``out_used_indexes``. N)rrNrrdrPr'r)rrKrindexesrr?r?r@rs    rc@s eZdZdS)rN)rrrr?r?r?r@r-srcCstjdtddS)NaSince Lhotse v1.20 'shuffle_buffer_size' is deprecated, because DynamicBucketingSampler does not require a separate shuffling buffer anymore. To improve both shuffling and sampling randomness, increase 'buffer_size' instead. To maintain backward compatibility, we will increase 'buffer_size' by 'shuffling_buffer_size' for you. This argument will be deprecated in a future Lhotse version.rB)rIrJrLr?r?r?r@rM1s rMr9);rur rrI dataclassesrr itertoolsrrrtypingrrr r r r r rrrrrnumpyrrwlhotserr lhotse.cutrlhotse.dataset.dataloadingrlhotse.dataset.sampling.baserrrrrlhotse.dataset.sampling.dynamicrrr lhotse.utilsrr rrrrQrr~rvrNr ExceptionrrMr?r?r?r@s\  8   o ,&