o SiZA@sddlZddlZddlmZddlmZmZmZmZm Z m Z m Z m Z m Z mZddlmZmZddlmZddlmZddlmZmZmZmZmZddlmZmZGd d d eZGd d d Z Gd dde Z!de de de fddZ"dS)N)deque) AnyCallableDict GeneratorIterableListLiteralOptionalTupleUnion)CutSetSeconds)Cut) resolve_seed) CutSamplerEpochDiagnosticsSamplingConstraintSamplingDiagnosticsTimeConstraint)ifnonestreaming_shufflecs<eZdZdZddddddddddddd ded eed eed eed e d e de dedeedeedeede ee dfddffddZ de eefffdd Zde eefddffdd ZddZd)dd Zde eeeffd!d"Zedeefd#d$Zedeefd%d&Zedeefd'd(ZZS)*DynamicCutSamplera A dynamic (streaming) variant of sampler that doesn't stratify the sampled cuts in any way. It is a generalization of :class:`~lhotse.dataset.sampling.SimpleCutSampler` and :class:`~lhotse.dataset.sampling.CutPairsSampler` in that it allows to jointly iterate an arbitrary number of CutSets. When input CutSets are opened in lazy mode, this sampler doesn't require reading the whole cut set into memory. For scenarios such as ASR, VAD, Speaker ID, or TTS training, this class supports single CutSet iteration. Example:: >>> cuts = CutSet(...) >>> sampler = DynamicCutSampler(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 = DynamicCutSampler(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:`DynamicCutSampler.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. NFTi Nr) max_durationmax_cuts constraintshuffle drop_lastconsistent_idsshuffle_buffer_sizequadratic_duration world_sizerankseedstrictcutsrrrrrrrr r!r"r#)trng randomizedreturnc s~tj|| | | dtdd| Dstd| |_||_||_||_||_ ||_ ||_ ||_ | dur=tjdt ddSdS)aU :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 constraint: Provide a :class:`~lhotse.dataset.sampling.base.SamplingConstraint` object defining how the sampler decides when a mini-batch is complete. It also affects which attribute of the input examples decides the "size" of the example (by default it's ``.duration``). Before this parameter was introduced, Lhotse samplers used :class:`~lhotse.dataset.sampling.base.TimeConstraint` implicitly. Introduced in Lhotse v1.22.0. :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 shuffle_buffer_size: How many cuts (or cut pairs, triplets) are being held in memory a buffer used for streaming shuffling. Larger number means better randomness at the cost of higher 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 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. )rr!r"r#css |] }t|tr|jVqdSN) isinstancer is_lazy.0csr/S/home/ubuntu/.local/lib/python3.10/site-packages/lhotse/dataset/sampling/dynamic.py {sz-DynamicCutSampler.__init__..zYou are using DynamicCutSampler with an eagerly read CutSet. You won't see any memory/speed benefits with that setup. Use e.g. 'CutSet.from_jsonl_lazy' to read the CutSet lazily.Nz|In Lhotse v1.4 all samplers act as if 'strict=True'. Sampler's argument 'strict' will be removed in a future Lhotse release.)category)super__init__allwarningswarnr%rrrrrrr DeprecationWarning)selfrrrrrrrr r!r"r#r$r% __class__r/r0r4Gs*1 zDynamicCutSampler.__init__cs@|jdus Jdt}||j|j|j|j|jd|S)NzIstate_dict() is not supported with samplers that use a custom constraint.)rrrrr ) rr3 state_dictupdaterrrrr r9sdr:r/r0r<s   zDynamicCutSampler.state_dictr?cs`|d|_|d|_|d|_|d|_|d|_|ddt||dS)Nrrrrr r$) poprrrrr r3load_state_dict _fast_forwardr>r:r/r0rAs        z!DynamicCutSampler.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)r9rEnum_batches_to_iter_r/r/r0rBs     zDynamicCutSampler._fast_forwardc sjrSjtjddjD_jr'fddjD_tt jfddjd_t jj j j jjjd_tj_S)NcSsg|]}t|qSr/)rKr,r/r/r0 sz.DynamicCutSampler.__iter__..cs(g|]}t|tjjdqS))rngbufsize)rrandomRandomrCrr,r#r9r/r0rPscstfdd|DS)Nc3s|]}|VqdSr)) _filter_fnr-cr9r/r0r1z?DynamicCutSampler.__iter__....)r5)tplrYr/r0z,DynamicCutSampler.__iter__..)iterator predicaterD)rrrrr rD)rJrDreset_current_epochrr#r% cuts_iterrFilterzipDurationBatcherrrrrr rKrYr/rUr0__iter__s2      zDynamicCutSampler.__iter__cstt|j}|jr8t|tr8t|D]&}|djtfdd|ddDs7Jdddd|Ddq|S) Nrc3s|]}|jkVqdSr)idrW expected_idr/r0r1rZz0DynamicCutSampler._next_batch..zhThe input CutSet are not sorted by cut ID in the same way. We sampled the following mismatched cut IDs: z, css|]}|jVqdSr)rfrWr/r/r0r1sz[. If this is expected, pass the argument 'consistent_ids=False' to DynamicBucketingSampler.) rMrarr*tuplercrgr5join)r9batchr%r/rhr0 _next_batchs    zDynamicCutSampler._next_batchcCdSr)r/rYr/r/r0remaining_durationz$DynamicCutSampler.remaining_durationcCror)r/rYr/r/r0remaining_cutsrqz DynamicCutSampler.remaining_cutscCror)r/rYr/r/r0num_cutsrqzDynamicCutSampler.num_cuts)r(r)__name__ __module__ __qualname____doc__rr rintrboolr r r4rstrrr<rArBrer r rnpropertyfloatrprrrs __classcell__r/r/r:r0rsh+     J * rc@seZdZ      ddeeeeefdedee dee de deed ee d dfd d Z d eeeeefddffd dZd eeeeffddZdS)rdNFdatapiperrrrr rDr(cCsT||_t|_||_t|t|_t||||dur ||_dSt |||d|_dS)N)rrr ) r~rreuse_cuts_bufferrrrrDcheck_constraintrr)r9r~rrrrr rDr/r/r0r4s    zDurationBatcher.__init__ccs:t|j|_z |Vq tyYnwd|_dSr))rKr~ra_collect_batch StopIterationrYr/r/r0res    zDurationBatcher.__iter__c Csdtttttfdttttffdd}|jg} zt|j}Wn5t y[|r=|j r7|j r=||YSz |j |Wt tyZ|j |dYt ww|||jt|trm|dn||j r|jrt|dkrtd ||Sq) Nr%r(cSs^t|dtr*t|ddkrtdd|D}|Stt|}tdd|DSt|S)zJHelper to do the right thing whether we sampled single cuts or cut tuples.rrjcss|]}|dVqdS)rNr/r,r/r/r0r1)szDDurationBatcher._collect_batch..detuplify..cSsg|]}t|qSr/)r from_cutsr,r/r/r0rP-r]zEDurationBatcher._collect_batch..detuplify..)r*rklenr rlistrc)r%tuple_of_cut_listsr/r/r0 detuplify#s  z1DurationBatcher._collect_batch..detuplifyTrrjzWe have exceeded the max_duration constraint during sampling but have only 1 cut. This is likely because max_duration was set to a very low value ~10s, or you're using a CutSet with very long cuts (e.g. 100s of seconds long).)rr rr r rresetrMrarrclose_to_exceedingrDdiscardAttributeErrorappendaddr*rkexceededrr6r7)r9rr%next_cut_or_tplr/r/r0r"sN       zDurationBatcher._collect_batch)NNNFNN)rtrurvrr rr rr rxrryrr4rr rerr/r/r/r0rds4 $ rdc @sJeZdZdZ d dedeegefdee ddfddZ defd d Z dS) rbaH A wrapper over an iterable that enables lazy filtering. It works like Python's `filter` built-in by applying the filter predicate to each element and yielding it further if predicate returned ``True``. This variant additionally tracks the number of discarded items and updates the sampling statistics. Nr^r_rDr(cCs8||_||_t|t|_t|jsJd|ddS)Nz2LazyFilter: 'predicate' arg must be callable (got z).)r^r_rrrDcallable)r9r^r_rDr/r/r0r4js zFilter.__init__ccs0|jD]}||r|Vq|j|qdSr))r^r_rDr)r9itemr/r/r0rexs   zFilter.__iter__r)) rtrurvrwrrrryr rr4rer/r/r/r0rb`s   rbrrrcCsD|dur|dur |dusJddS|dus|dus JddSdS)Nz;Cannot specify both constraint= and max_duration=/max_cuts=zVAt least one of max_duration= or max_cuts= has to be defined (or provide constraint=).r/)rrrr/r/r0rs r)#rSr6 collectionsrtypingrrrrrrr r r r lhotser r lhotse.cutrlhotse.dataset.dataloadingrlhotse.dataset.sampling.baserrrrr lhotse.utilsrrrrdrbrr/r/r/r0s 0  c`