o SiG @sddlZddlmZddlmZddlmZddlmZm Z m Z m Z m Z m Z mZmZddlZddlmZddlmZddlmZmZdd lmZGd d d eZd ed ede e edffddZd ed ede efddZdS)N)deepcopy)reduce)add)AnyCallableDictListOptionalTupleTypeUnion)CutSet)Cut) CutSamplerSamplingDiagnostics)SimpleCutSamplercseZdZdZedddddededed ed ed e d d ffddZ e d e e fddZe d e efddZe d e efddZded d ffdd Zdeegefd d fddZfddZd eee fffdd Zd eee fd d ffd!d" Zd5d#d$Zd eeeffd%d&Zd'd(Ze d efd)d*Ze d+d,Z d-e!eeed.ffd d fd/d0Z"e d e#fd1d2Z$d efd3d4Z%Z&S)6BucketingSamplera Sorts the cuts in a :class:`CutSet` by their duration and puts them into similar duration buckets. For each bucket, it instantiates a simpler sampler instance, e.g. :class:`SimpleCutSampler`. It behaves like an iterable that yields lists of strings (cut IDs). During iteration, it randomly selects one of the buckets to yield the batch from, until all the underlying samplers are depleted (which means it's the end of an epoch). Examples: Bucketing sampler with 20 buckets, sampling single cuts:: >>> sampler = BucketingSampler( ... cuts, ... # BucketingSampler specific args ... sampler_type=SimpleCutSampler, num_buckets=20, ... # Args passed into SimpleCutSampler ... max_duration=200 ... ) Bucketing sampler with 20 buckets, sampling pairs of source-target cuts:: >>> sampler = BucketingSampler( ... cuts, target_cuts, ... # BucketingSampler specific args ... sampler_type=CutPairsSampler, num_buckets=20, ... # Args passed into CutPairsSampler ... max_source_duration=200, max_target_duration=150 ... ) Fr) sampler_type num_buckets drop_lastseedcutsrrrrkwargsreturnNcstjdd|d|_|_|_|_tddjDr$tdtjd|i_ fdd j D_ t j j_d g|_d S) a BucketingSampler's constructor. :param cuts: one or more ``CutSet`` objects. The first one will be used to determine the buckets for all of them. Then, all of them will be used to instantiate the per-bucket samplers. :param sampler_type: a sampler type that will be created for each underlying bucket. :param num_buckets: how many buckets to create. :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 seed: random seed for bucket selection :param kwargs: Arguments used to create the underlying sampler for each bucket. r)r world_sizerankrcs|]}|jVqdSN)is_lazy).0csr#U/home/ubuntu/.local/lib/python3.10/site-packages/lhotse/dataset/sampling/bucketing.py Qz,BucketingSampler.__init__..zBucketingSampler does not support working with lazy CutSet (e.g., those opened with 'load_manifest_lazy', 'CutSet.from_jsonl_lazy', or 'CutSet.from_webdataset'). Please use lhotse.dataset.DynamicBucketingSampler instead.rcs"g|] }j|dijqS)r)rsampler_kwargs)r!bucket_cut_setsrselfr#r$ _sz-BucketingSampler.__init__..FN)super__init__rrr'cut_setsany ValueErrorcreate_buckets_equal_durationbucketsbucket_samplersrandomRandomrepoch bucket_rngdepleted)r*rrrrrr __class__r)r$r-/s0  zBucketingSampler.__init__cC,z tdd|jDWStyYdSw)a Remaining duration of data left in the sampler (may be inexact due to float arithmetic). Not available when the CutSet is read in lazy mode (returns None). .. note: For BucketingSampler, it's the sum of remaining duration in all buckets. cs|]\}}|jVqdSr)remaining_durationr!_sr#r#r$r%u  z6BucketingSampler.remaining_duration..Nsum_nondepleted_samplers_with_idxs TypeErrorr*r#r#r$r=l z#BucketingSampler.remaining_durationcCr;)z Remaining number of cuts in the sampler. Not available when the CutSet is read in lazy mode (returns None). .. note: For BucketingSampler, it's the sum of remaining cuts in all buckets. csr<r)remaining_cutsr>r#r#r$r%rAz2BucketingSampler.remaining_cuts..NrBrFr#r#r$rH{rGzBucketingSampler.remaining_cutscCr;)z Total number of cuts in the sampler. Not available when the CutSet is read in lazy mode (returns None). .. note: For BucketingSampler, it's the sum of num cuts in all buckets. csrr)num_cutsr!r@r#r#r$r%r&z,BucketingSampler.num_cuts..N)rCr3rErFr#r#r$rIs  zBucketingSampler.num_cutsr6cs&|jD]}||qt|dS)a Sets the epoch for this sampler. When :attr:`shuffle=True`, this ensures all replicas use a different random ordering for each epoch. Otherwise, the next iteration of this sampler will yield the same ordering. :param epoch: Epoch number. N)r3 set_epochr,)r*r6r@r9r#r$rKs  zBucketingSampler.set_epoch predicatecCs|jD]}||qdS)a  Add a constraint on individual cuts that has to be satisfied to consider them. Can be useful when handling large, lazy manifests where it is not feasible to pre-filter them before instantiating the sampler. Example: >>> cuts = CutSet(...) ... sampler = SimpleCutSampler(cuts, max_duration=100.0) ... # Retain only the cuts that have at least 1s and at most 20s duration. ... sampler.filter(lambda cut: 1.0 <= cut.duration <= 20.0) N)r3filter)r*rLsamplerr#r#r$rMs zBucketingSampler.filtercs"t|jD]}|qdS)a8 Enables re-setting to the start of an epoch when iter() is called. This is only needed in one specific scenario: when we restored previous sampler state via ``sampler.load_state_dict()`` but want to discard the progress in the current epoch and start from the beginning. N)r,allow_iter_to_reset_stater3)r*r@r9r#r$rOs   z*BucketingSampler.allow_iter_to_reset_statecsDt}||jt|jdd|jDt|j|j d|S)z Return the current state of the sampler in a state_dict. Together with ``load_state_dict()``, this can be used to restore the training loop's state to the one stored in the state_dict. cSsg|]}|qSr#) state_dictrJr#r#r$r+sz/BucketingSampler.state_dict..)rr8r3r'bucket_rng_state) r,rPupdaterrr8r3r'r7getstate)r*rPr9r#r$rPs  zBucketingSampler.state_dictrPcs|d}|j|ksJd|jd|d|dd|dd|d|_|d |_|j|d t|jt|d ksTJd t|jdt|d dt|j|d D] \}}| |q]t |dS) aX Restore the state of the sampler that is described in a state_dict. This will result in the sampler yielding batches from where the previous training left it off. .. caution:: The samplers are expected to be initialized with the same CutSets, but this is not explicitly checked anywhere. .. caution:: The input ``state_dict`` is being mutated: we remove each consumed key, and expect it to be empty at the end of loading. If you don't want this behavior, pass a copy inside of this function (e.g., using ``import deepcopy``). .. note:: For implementers of sub-classes of CutSampler: the flag ``self._just_restored_state`` has to be handled in ``__iter__`` to make it avoid resetting the just-restored state (only once). rzaError in BucketingSampler.load_state_dict(): Inconsistent number of buckets: current sampler has z, the state_dict has .proportional_samplingN bucket_methodr'r8rQr3zbError in BucketingSampler.load_state_dict(): Inconsistent number of samplers: current sampler has ) poprr'r8r7setstatelenr3zipload_state_dictr,)r*rPrrN sampler_sdr9r#r$r[s2         z BucketingSampler.load_state_dictcCsN|jr|S|j|j|j|j|jD]}t|qdg|j|_ |S)NF) _just_restored_state diagnosticsreset_current_epochr7rr6r3iterrr8)r*br#r#r$__iter__s   zBucketingSampler.__iter__cCs|jdjr |j|jS|j}t|dkr|dS|j|\}}|j|\}}z |j|j|j}Wn tyB||fYSw|j|krN||fS||fS)Nrr) r.r r7choicerDrYr=ZeroDivisionErrorr4)r*idx_sampler_pairsidx1sampler1idx2sampler2prob1r#r#r$_select_bucket_with_idx s"     z(BucketingSampler._select_bucket_with_idxcCsN||js$|\}}zt|WSty d|j|<Ynw|jrt)NT)rO is_depletedrknext StopIterationr8)r*idxrNr#r#r$ _next_batch-s   zBucketingSampler._next_batchcCs t|jSr)allr8rFr#r#r$rl7 zBucketingSampler.is_depletedcCsddtt|j|jDS)NcSs g|] \}\}}|s||fqSr#r#)r!robsr8r#r#r$r+=s  zDBucketingSampler._nondepleted_samplers_with_idxs..) enumeraterZr3r8rFr#r#r$rD;s  z0BucketingSampler._nondepleted_samplers_with_idxsbatch.cCsdSrr#)r*rur#r#r$_log_diagnosticsEsz!BucketingSampler._log_diagnosticscCsttdd|jDS)Ncsrr)r^r!bucketr#r#r$r%Jr&z/BucketingSampler.diagnostics..)rrr3rFr#r#r$r^HszBucketingSampler.diagnosticscCs |jS)zJReturns a string describing the statistics of the sampling process so far.)r^ get_reportrFr#r#r$ryLrrzBucketingSampler.get_report)rr)'__name__ __module__ __qualname____doc__rr r intboolrr-propertyr floatr=rHrIrKrrrMrOrstrrPr[rbr rrkrprlrDr rvrr^ry __classcell__r#r#r9r$rsV"=    )#  " rrrr.csX|djdd}t||dg}|ddD]|fdd|dDqtt|S) a Creates buckets of cuts with similar durations. Each bucket has the same cumulative duration, but a different number of cuts. :param cuts: One or more CutSets; the input CutSets are assumed to have the same cut IDs (i.e., the cuts correspond to each other and are meant to be sampled together as pairs, triples, etc.). :param num_buckets: The number of buckets. :return: A list of CutSet buckets (or tuples of CutSet buckets, depending on the input). rT) ascending)rrNc3s|] }j|jdVqdS))cut_idsN)subsetidsrwcut_setr#r$r%cs  z0create_buckets_equal_duration..)sort_by_duration%_create_buckets_equal_duration_singleappendlistrZ)rr first_cut_setbuckets_per_cutsetr#rr$r1Qs  r1cstdd|D}||}ttdt|dttt|dt|ddd}ddt|D}d d|D}tt||d d d }d \}} |d} ddt|D} ddt|Dd} i} t|dD]v\}\}}| durx| | krx| } |dr| | ||kr| dur| | krtdt | d|d} n t | d|d} | | |7<| | ||<qh| | ||kr| dur| | krt| dd} nt| dd} | | |7<| | ||<qht|D] \}}| | |qfddt|D}|S)z Helper method to partition a single CutSet into buckets that have the same cumulative duration. See also: :meth:`.create_buckets_from_duration_percentiles`. cSg|]}|jqSr#durationr!cr#r#r$r+vz9_create_buckets_equal_duration_single..rrcSsi|]\}}||qSr#r#)r!io_idxr#r#r$ sz9_create_buckets_equal_duration_single..cSrr#rrr#r#r$r+rcSs|dS)Nrr#)xr#r#r$sz7_create_buckets_equal_duration_single..)key)rrcSsi|]}|dqS)rr#r!rr#r#r$rrcSsi|]}|gqSr#r#rr#r#r$rrNcsg|] }t|qSr#)r from_cutsrbuckets_cut_dictr#r$r+s) nprCrrangerYrtsortedrZmaxminr)rrtotal_durationbucket_durationorder order2idx durationsordered_cut_durations last_order first_bucket last_bucket buckets_dict middle_bucketidx_to_bucket_idr order_idxrcut_idxcutr2r#rr$rmsD r)r4copyr functoolsroperatorrtypingrrrrr r r r numpyrlhotser lhotse.cutrlhotse.dataset.sampling.baserrlhotse.dataset.sampling.simplerrr~r1rr#r#r#r$s6   (   D