o Siۈ@sddlZddlZddlZddlmZmZddlmZddlmZddl m Z m Z ddl m Z ddlmZmZmZmZmZmZmZmZddlZddlmZdd lmZdd lmZmZdd lm Z dd l!m"Z"dd l#m$Z$ddl%m&Z&m'Z'm(Z(m)Z)Gdddee"Z*de+dee,ge,ffddZ-dede+de+ddfddZ.GdddedZ/e Gddde/Z0e Gdd d e/Z1e Gd!d"d"Z2e Gd#d$d$Z3Gd%d&d&Z4d'eege5fd(eege5fdeege5ffd)d*Z6dS)+N)ABCMetaabstractmethod) bisect_left)deepcopy)asdict dataclass)isclose)AnyCallableDictIterableLiteralOptionalTupleUnion) distributed)Sampler)CutCutSet) TextExample)Dillable)combine)Secondsexactly_one_not_nullifnone is_none_or_gtcseZdZdZ     d4dededeedeed eeed fd df fd d Z e ddZ deedeefddZ ded dfddZ deegefd dfddZdeegefd dfddZd eeeffddZdeeefd dfddZd d!Zd"d#Ze d eefd$d%Ze d eefd&d'Ze d eefd(d)Zd*d+Zd,d-Zd.eee ed/ffd dfd0d1Z!d efd2d3Z"Z#S)5 CutSamplera ``CutSampler`` is responsible for collecting batches of cuts, given specified criteria. It implements correct handling of distributed sampling in ``DataLoader``, so that the cuts are not duplicated across workers. Sampling in a ``CutSampler`` is intended to be very quick - it only uses the metadata in ``CutSet`` manifest to select the cuts, and is not intended to perform any I/O. CutSampler works similarly to PyTorch's DistributedSampler - when :attr:`shuffle=True`, you should call ``sampler.set_epoch(epoch)`` at each new epoch to have a different ordering of returned elements. However, its actual behaviour is different than that of DistributedSampler -- instead of partitioning the underlying cuts into equally sized chunks, it will return every N-th batch and skip the other batches (where ``N == world_size``). The formula used to determine which batches are returned is: ``(batch_idx + (world_size - rank)) % world_size == 0``. This ensures that we can return an equal number of batches in all distributed workers in spite of using a dynamic batch size, at the cost of skipping at most ``world_size - 1`` batches. Example usage:: >>> dataset = K2SpeechRecognitionDataset(cuts) >>> sampler = SimpleCutSampler(cuts, max_duration=200, shuffle=True) >>> loader = DataLoader(dataset, sampler=sampler, batch_size=None) >>> for epoch in range(start_epoch, n_epochs): ... sampler.set_epoch(epoch) ... train(loader) .. note:: For implementers of new samplers: Subclasses of CutSampler are expected to implement ``self._next_batch()`` to introduce specific sampling logic (e.g. based on filters such as max number of frames/tokens/etc.). CutSampler defines ``__iter__()``, which optionally shuffles the cut IDs, and resets ``self.cut_idx`` to zero (to be used and incremented inside of ``_next_batch()``. FNrshuffle drop_last world_sizerankseed) randomizedtrngreturncsPt||_||_||_d|_t|_d|_|j ||dt |_ g|_ dS)a :param shuffle: When ``True``, the cuts will be shuffled at the start of iteration. 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``, the last batch is dropped if it's incomplete. :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. rF)rr N) super__init__rrr!epochSamplingDiagnostics _diagnostics_just_restored_state_maybe_init_distributed_filter_nothing _filter_fn _transforms)selfrrrr r! __class__P/home/ubuntu/.local/lib/python3.10/site-packages/lhotse/dataset/sampling/base.pyr&;s  zCutSampler.__init__cCs|jS)z Info on how many cuts / batches were returned or rejected during iteration. This property can be overriden by child classes e.g. to merge diagnostics of composite samplers. )r)r/r2r2r3 diagnostics^szCutSampler.diagnosticscCs|dur |dks J|dur|dksJdtjvr5dtjvr5t|ttjd|_t|ttjd|_n%trNtrNt|t |_t|t |_n t|d|_t|d|_|j|jksbJdS)Nr WORLD_SIZERANK) osenvironrintrr dist is_availableis_initializedget_world_sizeget_rank)r/rr r2r2r3r+gs    z"CutSampler._maybe_init_distributedr'cCs(|j|kr |||_|j|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)r'allow_iter_to_reset_stater5 set_epochr/r'r2r2r3rB~s zCutSampler.set_epoch predicatecCs(t|jtr ||_|St|j||_|S)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) ) isinstancer-r,_and)r/rDr2r2r3filters zCutSampler.filterfncCs(t|s Jd|d|j||S)zAApply ``fn`` to each mini-batch of ``CutSet`` before yielding it.zAExpected a callable accepting and returning a CutSet, received: '')callabler.append)r/rHr2r2r3maps  zCutSampler.mapcCs&|j|j|j|j|j|j|jdS)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. )r'rrr r!rr5)r'rrr r!rr5 state_dictr4r2r2r3rMszCutSampler.state_dictrMcCs|d|_|d}|j|ksJd|jd|d|d=|j|dks)J|d}|j|kr@td |jd |d ||_|d |_|j |d t |dkscJdd | d|_ 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). rrzNCannot restore sampler with a different world_size (before load_state_dict(): z,attempted restoring to z]). Changing the world_size would result in different batches being returned from the sampler.r r!rzNOverriding the shuffle value in CutSampler based on state_dict(initialized to z; restored to ).r'r5rz:Error in CutSampler.load_state_dict(): Unexpected keys: -  - TN)poprrr!rwarningswarnr'r5load_state_dictlenjoinkeysr*)r/rMrrr2r2r3rSs6        zCutSampler.load_state_dictcCtd)Nz6Sub-classes of CutSampler have to implement __iter__()NotImplementedErrorr4r2r2r3__iter__zCutSampler.__iter__cCrW)Nz>Sub-classes of CutSampler have to implement self._next_batch()rXr4r2r2r3 _next_batchr[zCutSampler._next_batchcCrW)z 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). zCSub-classes of CutSampler have to implement self.remaining_durationrXr4r2r2r3remaining_durationzCutSampler.remaining_durationcCrW)z Remaining number of cuts in the sampler. Not available when the CutSet is read in lazy mode (returns None). z?Sub-classes of CutSampler have to implement self.remaining_cutsrXr4r2r2r3remaining_cutsr^zCutSampler.remaining_cutscCrW)z Total number of cuts in the sampler. Not available when the CutSet is read in lazy mode (returns None). z9Sub-classes of CutSampler have to implement self.num_cutsrXr4r2r2r3num_cutsr^zCutSampler.num_cutscCs d|_dS)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. FN)r*r4r2r2r3rAs z$CutSampler.allow_iter_to_reset_statec Cs$|g}t|jD]}z |}||Wq ty*|jdks'|jr(Yq wt|dkr4tt||jkrstdd|D}d}|jt|}dkrm||j |d t |}|d7}|jt|}dksQ| |j}||j }|||jD]}||}qt||j |jd|S)Nr6rcSsg|]}|dur|qSNr2).0br2r2r3 >sz'CutSampler.__next__..)first)r r)rArangerr\rK StopIterationrrTrsubset modify_idsmark_as_duplicatesplitr _log_diagnosticsr.attach_dataloading_info) r/batches_batchcombinedchunkdiffselectedtfnr2r2r3__next__s:        zCutSampler.__next__rp.cCsTt|tr |j|dSt|tr#t|dtr#|j|ddStd|)NrzObject with unexpected type: )rErr5keeptuple ValueError)r/rpr2r2r3rlNs zCutSampler._log_diagnosticscCs |jS)JReturns a string describing the statistics of the sampling process so far.)r5 get_reportr4r2r2r3r{V zCutSampler.get_report)FFNNr)$__name__ __module__ __qualname____doc__boolrr;rr r&propertyr5r+rBr rrGrrLr strr rMrSrZr\floatr]r_r`rArvrrlr{ __classcell__r2r2r0r3rsP&# .    "7r iterationr$csdtdtffdd }|S)Ncut_idr$cs|dS)N_dupr2)rrr2r3inner\sz mark_as_duplicate..inner)r)rrr2rr3rj[srjcutsr rcCs@tjj}|dur d}n|j}|||d}|D]}||_qdS)a= Attaches diagnostic info about dataloading to each cut under ``dataloading_info`` custom field. This information contains the rank, world_size, and worker_id. If the training is not distributed, rank and world_size are 0 and 1. If the num_workers argument in DataLoader was 0, worker_id is None. N)r r worker_id)torchutilsdataget_worker_infoiddataloading_info)rr r worker_inforinfocutr2r2r3rmbs  rmc @seZdZdZededdfddZedefddZedefd d Z edd d Z edede fd dZ ddededede fddZdddZdS)SamplingConstraintz Defines the interface for sampling constraints. A sampling constraint keeps track of the sampled examples and lets the sampler know when it should yield a mini-batch. exampler$NcCdS)z Update the sampling constraint with the information about the sampled example (e.g. current batch size, total duration). Nr2r/rr2r2r3addzzSamplingConstraint.addcCr)z4Inform if the sampling constraint has been exceeded.Nr2r4r2r2r3exceededzSamplingConstraint.exceededcCr)zVInform if we're going to exceed the sampling constraint after adding one more example.Nr2r4r2r2r3close_to_exceedingrz%SamplingConstraint.close_to_exceedingcCr)z?Resets the internal state (called after yielding a mini-batch).Nr2r4r2r2r3resetrzSamplingConstraint.resetcCr)z Returns the "size" of an example, used to create bucket distribution for bucketing samplers (e.g., for audio it may be duration; for text it may be number of tokens; etc.). Nr2rr2r2r3measure_lengthrz!SamplingConstraint.measure_lengthbuckets example_lencCs<t||sJd|d|d|dur||}t||S)a Given a list of buckets and an example, assign the example to the correct bucket. This is leveraged by bucketing samplers. Default implementation assumes that buckets are expressed in the same units as the output of :meth:`SamplingConstraint.measure_length` and returns the index of the first bucket that has a larger length than the example. zYselect_bucket requires either example= or example_len= as the input (we received example=z and example_len=rNN)rrr)r/rrrr2r2r3 select_buckets   z SamplingConstraint.select_bucketcCs t|S)z)Return a shallow copy of this constraint.)copyr4r2r2r3rr|zSamplingConstraint.copyr$N)NN)r$r)r}r~rrrr rrrrrrrr;rrr2r2r2r3rss.  r) metaclassc@s6eZdZUdZdZeeed<dZee ed<dZ e e efed<dZ e ed<dZ e e efed<dZeeed <d'd d Zd efd dZded dfddZded efddZd efddZd efddZd'ddZded efddZd eeeffddZdeeefd dfd d!Zd(d#d$Zd"dd efd%d&Z dS))TimeConstraintal Represents a time-based constraint for sampler classes. It is defined as maximum total batch duration (in seconds) and/or the total number of cuts. :class:`TimeConstraint` can be used for tracking whether the criterion has been exceeded via the `add(cut)`, `exceeded()` and `reset()` methods. It will automatically track the right criterion (i.e. select duration from the cut). It can also be a null constraint (never exceeded). When ``quadratic_duration`` is set, we will try to compensate for models that have a quadratic complexity w.r.t. the input sequence length. We use the following formula to determine the effective duration for each cut:: effective_duration = duration + (duration ** 2) / quadratic_duration We recomend setting quadratic_duration to something between 15 and 40 for transformer architectures. N max_durationmax_cutsrcurrentr` longest_seenquadratic_durationr$cC4t|jdsJt|jdsJt|jdsJdSNr)rrrrr4r2r2r3 __post_init__zTimeConstraint.__post_init__cCs|jdup |jduS)zAIs it an actual constraint, or a dummy one (i.e. never exceeded).N)rrr4r2r2r3 is_activeszTimeConstraint.is_activercCsD|jdur||j}|j|7_t|j||_|jd7_dS)z Increment the internal counter for the time constraint, selecting the right property from the input ``cut`` object. Nr6)r!_maybe_apply_quadratic_correctiondurationrmaxrr`)r/rrr2r2r3rs  zTimeConstraint.addrcC |jdur|S||d|jSN)r)r/rr2r2r3r z0TimeConstraint._maybe_apply_quadratic_correctioncC>|jdur |j|jkr dS|jdurdS|j|j}||jkSz"Is the constraint exceeded or not.NTFrr`rrr/effective_durationr2r2r3r    zTimeConstraint.exceededcCB|jdur |j|jkr dS|jdur|jd|j}||jkSdSa( 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. NTr6Frrr2r2r3r   z!TimeConstraint.close_to_exceedingcCd|_d|_d|_dSz{ Reset the internal counter (to be used after a batch was created, to start collecting a new one). rN)rr`rr4r2r2r3r zTimeConstraint.resetcC|jSra)rrr2r2r3rzTimeConstraint.measure_lengthcCt|Srarr4r2r2r3rM zTimeConstraint.state_dictrMcCs|d|_|d|_|d|_|d|_|dd|_|dd|_|dd|d d|d dt|dksIJd d | dS) Nrrrr`rrrstrict max_samples max_framesz>Error in TimeConstraint.load_state_dict(): Unexpected keys: - rO) rPrrrr`rrrTrUrVr/rMr2r2r3rSs        zTimeConstraint.load_state_dictotherc CsdD],}t||}t||}|duo|du}|s.t||s.Jd|d|d|d|d qt|j|j|j|j|j|jt|j|j|j dS)NrrrzXTo add two TimeConstraint objects, they need to represent the same constraint (got self.=z != other.rN)rrrr`rr) getattrrrrrrr`rrr)r/rkey self_attr other_attris_noner2r2r3__add__s0     zTimeConstraint.__add__cCs$|j|jko|j|jko|j|jkSrarr/rr2r2r3__eq__0s   zTimeConstraint.__eq__r)rrr$r)!r}r~rrrrr__annotations__rr;rrr`rrrrrrrrrrrrrr rr rMrSrrr2r2r2r3rs(       rc@seZdZUdZdZeed<dZeed<dZeed<dZ eed<dZ eed<dZ e eed <dd d Z d ed dfddZded efddZd efddZd efddZdddZd ed efddZdS)TokenConstrainta: Represents a token-based constraint for sampler classes that sample text data. It is defined as maximum total number of tokens in a mini-batch and/or max batch size. Similarly to :class:`TimeConstraint`, we support ``quadratic_length`` for quadratic token penalty when sampling longer texts. N max_tokens max_examplesrr num_examplesrquadratic_lengthr$cCrr)rrrrr4r2r2r3rIrzTokenConstraint.__post_init__rcCsH|jdur|||}|j|7_t|j||_|jd7_dS)z Increment the internal token counter for the constraint, selecting the right property from the input object. Nr6)rrrrrrr)r/rsizer2r2r3rNs zTokenConstraint.addrcCrr)r)r/rr2r2r3rYrz1TokenConstraint._maybe_apply_quadratic_correctioncCrrrrrrrr2r2r3rarzTokenConstraint.exceededcCrrr)r/effective_sizer2r2r3rjrz"TokenConstraint.close_to_exceedingcCrr)rrrr4r2r2r3ryrzTokenConstraint.resetcCrra) num_tokensrr2r2r3rrzTokenConstraint.measure_lengthr)r}r~rrrr;rrrrrrrrrrrrrrrrrr2r2r2r3r8s          rc@seZdZUdZeed<dZeed<dZeed<dZeed<dZ eed<e defdd Z e defd d Z de fd d Zdee effddZdee efddfddZdddZdS)EpochDiagnosticsrr' kept_cutsdiscarded_cuts kept_batchesdiscarded_batchesr$cC |j|jSra)rrr4r2r2r3 total_cuts zEpochDiagnostics.total_cutscCrra)rrr4r2r2r3 total_batchesrzEpochDiagnostics.total_batchescCs|jdks |jdkr dSd|jdd|jdd|jdd|j|jd d |jdd |jdd|jdd|j|jd d |jdd S)rzrzSampling statistics unavailable: EpochDiagnostics received no cuts or batches. If this is unexpected, and you're using a custom sampler, ensure that the sampler is registering the batches in SamplerDiagnostics/EpochDiagnostics.z| ep z>3dz | cuts kept d/ (.2%) | cuts discarded  | batches kept )| batches discarded  |)rrr'rrrrr4r2r2r3r{s"   zEpochDiagnostics.get_reportcCrrarr4r2r2r3rMrzEpochDiagnostics.state_dictrMcCsb|d|_|d|_|d|_|d|_|d|_t|dks/Jdd||S) Nr'rrrrrz@Error in EpochDiagnostics.load_state_dict(): Unexpected keys: - rO) rPr'rrrrrTrUrVrr2r2r3rSs      z EpochDiagnostics.load_state_dictrcCsD|j|jksJt|j|j|j|j|j|j|j|j|jdS)N)r'rrrr)r'rrrrrrr2r2r3rs    zEpochDiagnostics.__add__N)rrr$r)r}r~rr'r;rrrrrrrrrr{r r rMrSrr2r2r2r3rs     rc@sjeZdZUdZdZeed<dZeee fed<ddZ d2d d Z d eddfd d Z d2ddZ ede fddZdeeddfddZdeeddfddZdeddfddZedefddZedefddZedefddZedefd d!Zedefd"d#Zedefd$d%Zd3d'edefd(d)Zdeeeffd*d+Zd,eeefddfd-d.Z d4d0d1Z!dS)5r(zr Utility for collecting diagnostics about the sampling process: how many cuts/batches were discarded. r current_epochNstats_per_epochcCs$|jduri|_||jdSdSra)rrBrr4r2r2r3rs z!SamplingDiagnostics.__post_init__r$cCst|j|j|j<dSra)rrrr4r2r2r3reset_current_epochsz'SamplingDiagnostics.reset_current_epochr'cCs(||_||jvrt|d|j|<dSdS)N)r')rrrrCr2r2r3rBs zSamplingDiagnostics.set_epochcCs||jddSNr6)rBrr4r2r2r3 advance_epochz!SamplingDiagnostics.advance_epochcCs |j|jSra)rrr4r2r2r3current_epoch_statsrz'SamplingDiagnostics.current_epoch_statsrcCsHd}|D]}|jjd7_|d7}q|std|jjd7_dS)Nrr6z@Found and accepted batch with zero cuts. This could be an error.)rrrQrRrr/rcntrrr2r2r3rws zSamplingDiagnostics.keepcCsBd}|D]}|jjd7_|d7}q|r|jjd7_dSdS)Nrr6)rrrrr2r2r3discards zSamplingDiagnostics.discardrcCs|jjd7_dSr)rrr/rr2r2r3discard_singlerz"SamplingDiagnostics.discard_singlecCtdd|jDS)Ncs|]}|jVqdSra)rrbsr2r2r3 z0SamplingDiagnostics.kept_cuts..sumrvaluesr4r2r2r3rzSamplingDiagnostics.kept_cutscCr)Ncsrra)rrr2r2r3rrz5SamplingDiagnostics.discarded_cuts..rr4r2r2r3rr z"SamplingDiagnostics.discarded_cutscCr)Ncsrra)rrr2r2r3rrz3SamplingDiagnostics.kept_batches..rr4r2r2r3rr z SamplingDiagnostics.kept_batchescCr)Ncsrra)rrr2r2r3rrz8SamplingDiagnostics.discarded_batches..rr4r2r2r3rr z%SamplingDiagnostics.discarded_batchescCr)Ncsrra)rrr2r2r3r rz1SamplingDiagnostics.total_cuts..rr4r2r2r3r r zSamplingDiagnostics.total_cutscCr)Ncsrra)rrr2r2r3rrz4SamplingDiagnostics.total_batches..rr4r2r2r3r r z!SamplingDiagnostics.total_batchesF per_epochcCs|jdks |jdkr dSg}|r#t|jD] }||j|q|d|jdd|jdd|j|jdd|jdd |jdd|jdd|j|jdd |j dd d |S) rzrzSampling statistics unavailable: the SamplerDiagnostics received no cuts or batches. If this is unexpected, and you're using a custom sampler, ensure that the sampler is registering the batches in SamplerDiagnostics.z| total | cuts kept rrrrrrrr ) rrsortedrrKr{rrrrrU)r/r retr'r2r2r3r{s.   zSamplingDiagnostics.get_reportcCrrarr4r2r2r3rM+rzSamplingDiagnostics.state_dictrMcCs*|d|_dd|dD|_|S)NrcSsi|] \}}|t|qSr2)rrS)rbr'sdr2r2r3 0s z7SamplingDiagnostics.load_state_dict..r)rPritemsrrr2r2r3rS.s  z#SamplingDiagnostics.load_state_dictrcCsNt|j}|jD]\}}||vr|||||<q |||<q t|j|dS)N)rr)rrrr(r)r/rrr'statsr2r2r3r6s  zSamplingDiagnostics.__add__r)F)rr(r$r()"r}r~rrrr;rrr rrrrBrrrr rrwrrrrrrrrrrr{r rMrSrr2r2r2r3r(s:      r(c@seZdZdedefddZdS)r,rr$cCr)NTr2rr2r2r3__call__Dsz_filter_nothing.__call__N)r}r~rrrrr2r2r2r3r,Csr,fn1fn2csdtdtffdd }|S)Nrr$cs|o|Srar2)rrrr2r3 _and_wrapperKsz_and.._and_wrapper)rr)rrrr2rr3rFHsrF)7rr9rQabcrrbisectrr dataclassesrrmathrtypingr r r r r rrrrrr<torch.utils.datar lhotse.cutrrlhotse.cut.textr lhotse.lazyrlhotse.manipulationr lhotse.utilsrrrrrr;rrjrmrrrrr(r,rrFr2r2r2r3sP   (     G=M=~