o Siu@sddlZddlZddlZddlZddlmZddlmZddlm Z m Z m Z m Z m Z mZmZmZddlmZmZmZmZddlmZmZmZmZedZGdd d ee ZGd d d Zed fd efddZ ded dfddZ!edefddZ"GdddZ#GdddZ$GdddeZ%GdddeZ&GdddeZ'Gdd d eZ(Gd!d"d"eZ)Gd#d$d$eZ*Gd%d&d&eZ+Gd'd(d(eZ,Gd)d*d*eZ-Gd+d,d,eZ.d-e d.e/d e fd/d0Z0d1efd2d3Z1dS)4N)contextmanager)partial)AnyCallableIterableListLiteralOptionalTypeVarUnion) LazyMixindecode_json_linedeserialize_item open_best)Pathlikefastcopyis_module_availablestreaming_shuffleTc @seZdZdZdeegeffddZdeegeffddZe dd d d d ed e e e e efde e edffddZe d d d dd e e e e efde e edfde e fddZ d!de ejde fddZd"de e defddZdd Zd S)#AlgorithmMixinz Helper base class with methods that are supposed to work identically on Lhotse manifest classes such as CutSet, RecordingSet, etc. predicatecs6t|}|jr|t|dS|fdd|DS)a' Return a new manifest containing only the items that satisfy ``predicate``. If the manifest is lazy, the filtering will also be applied lazily. :param predicate: a function that takes a cut as an argument and returns bool. :return: a filtered manifest. rc3s|] }|r|VqdSN).0cutrr?/home/ubuntu/.local/lib/python3.10/site-packages/lhotse/lazy.py 'sz(AlgorithmMixin.filter..)typeis_lazy LazyFilter from_items)selfrclsrrrfilterszAlgorithmMixin.filter transform_fncCs,t|}|t|j|d}|jr|S|S)a Apply `transform_fn` to each item in this manifest and return a new manifest. If the manifest is opened lazy, the transform is also applied lazily. :param transform_fn: A callable (function) that accepts a single item instance and returns a new (or the same) instance of the same type. E.g. with CutSet, callable accepts ``Cut`` and returns also ``Cut``. :return: a new ``CutSet`` with transformed cuts. )fn)r LazyMapperdatarto_eager)r"r%r#ansrrrmap)s  zAlgorithmMixin.mapFNr stop_earlyweightsseedr-r.r/trng randomizedcG|t||||dS)a Merges multiple manifest iterables into a new iterable by lazily multiplexing them during iteration time. If one of the iterables is exhausted before the others, we will keep iterating until all iterables are exhausted. This behavior can be changed with ``stop_early`` parameter. :param manifests: iterables to be multiplexed. They can be either lazy or eager, but the resulting manifest will always be lazy. :param stop_early: should we stop the iteration as soon as we exhaust one of the manifests. :param weights: an optional weight for each iterable, affects the probability of it being sampled. The weights are uniform by default. If lengths are known, it makes sense to pass them here for uniform distribution of items in the expectation. :param seed: the random seed, ensures deterministic order across multiple iterations. r,)LazyIteratorMultiplexer)r#r-r.r/ manifestsrrrmux9s zAlgorithmMixin.muxr.r/max_open_streamsr8cGr3)a Merges multiple manifest iterables into a new iterable by lazily multiplexing them during iteration time. Unlike ``mux()``, this method allows to limit the number of max open sub-iterators at any given time. To enable this, it performs 2-stage sampling. First, it samples with replacement the set of iterators ``I`` to construct a subset ``I_sub`` of size ``max_open_streams``. Then, for each iteration step, it samples an iterator ``i`` from ``I_sub``, fetches the next item from it, and yields it. Once ``i`` becomes exhausted, it is replaced with a new iterator ``j`` sampled from ``I_sub``. .. caution:: Do not use this method with inputs that are infinitely iterable as they will silently break the multiplexing property by only using a subset of the input iterables. .. caution:: This method is not recommended for multiplexing for a small amount of iterations, as it may be much less accurate than ``mux()`` depending on the number of open streams, iterable sizes, and the random seed. :param manifests: iterables to be multiplexed. They can be either lazy or eager, but the resulting manifest will always be lazy. :param weights: an optional weight for each iterable, affects the probability of it being sampled. The weights are uniform by default. If lengths are known, it makes sense to pass them here for uniform distribution of items in the expectation. :param seed: the random seed, ensures deterministic order across multiple iterations. :param max_open_streams: the number of iterables that can be open simultaneously at any given time. r7)"LazyInfiniteApproximateMultiplexer)r#r.r/r8r5rrr infinite_muxUs#zAlgorithmMixin.infinite_mux'rng buffer_sizecCsJt|}|dur t}|jr|t|j||dS|j}||||S)ad Shuffles the elements and returns a shuffled variant of self. If the manifest is opened lazily, performs shuffling on-the-fly with a fixed buffer size. :param rng: an optional instance of ``random.Random`` for precise control of randomness. :return: a shuffled copy of self, or a manifest that is shuffled lazily. N)r=r<)rrandomr LazyShufflerr(copyshuffle)r"r<r=r#newrrrrAs   zAlgorithmMixin.shuffletimes preserve_idcCst|}|t|||dS)aP Return a new, lazily evaluated manifest that iterates over the original elements ``times`` number of times. :param times: how many times to repeat (infinite by default). :param preserve_id: when ``True``, we won't update the element ID with repeat number. :return: a repeated manifest. )rCrD)r LazyRepeater)r"rCrDr#rrrrepeats zAlgorithmMixin.repeatcCst|}|t|j|jSr)rLazyIteratorChainr()r"otherr#rrr__add__szAlgorithmMixin.__add__)Nr;NF)__name__ __module__ __qualname____doc__rrboolr$r+ classmethodr rr intfloatrr6r:r>RandomrArFrIrrrrrsF  -  rc@s(eZdZdZhdZddZddZdS)Dillablea< Mix-in that will leverage ``dill`` instead of ``pickle`` when pickling an object. It is useful when the user can't avoid ``pickle`` (e.g. in multiprocessing), but needs to use unpicklable objects such as lambdas. If ``dill`` is not installed, it defers to what ``pickle`` does by default. >1yesTruetruecCs tr ddl}||jS|jSNr)is_dill_enableddilldumps__dict__)r"r[rrr __getstate__s zDillable.__getstate__cCs(trddl}|||_dS||_dSrY)rZr[loadsr])r"stater[rrr __setstate__s zDillable.__setstate__N)rKrLrMrN_ENABLED_VALUESr^rarrrrrTs   rT)rUrWrXrVreturncCstdo tjdd|vS)zKReturns bool indicating if dill-based pickling in Lhotse is enabled or not.r[LHOTSE_DILL_ENABLED0)rosenvironget)rbrrrrZsrZvaluecCs&tdsJd|r dndtjd<dS)z0Enable or disable dill-based pickling in Lhotse.r[z^Cannot enable dill because dill is not installed. Please run 'pip install dill' and try again.rUrerdN)rrfrg)rirrrset_dill_enableds rjccs"t}t|dVt|dS)a: Context manager that overrides the setting of Lhotse's dill-backed pickling and restores the previous value after exit. Example:: >>> import pickle ... with dill_enabled(True): ... pickle.dump(CutSet(...).filter(lambda c: c.duration < 5), open("cutset.pickle", "wb")) N)rZrj)ripreviousrrr dill_enableds   rlc@s>eZdZdZd dededdfddZd d Zdefd d Z dS)LazyTxtIteratorz LazyTxtIterator is a thin wrapper over builtin ``open`` function to iterate over lines in a (possibly compressed) text file. It can also provide the number of lines via __len__ via fast newlines counting. Tpathas_text_examplercNcCs||_||_d|_dSr)rnro_len)r"rnrorrr__init__s zLazyTxtIterator.__init__ccsddlm}d}t|jd}|D]}|}|jr||}|V|d7}qWdn1s1wY|jdur@||_dSdS)Nr) TextExampler)lhotse.cut.textrrrrnstriprorp)r"rrtotflinerrr__iter__s    zLazyTxtIterator.__iter__cC|jdur t|j|_|jSrrpcount_newlines_fastrnr"rrr__len__  zLazyTxtIterator.__len__)T) rKrLrMrNrrOrqrzrQrrrrrrms rmc@s8eZdZdZdeddfddZddZdefd d ZdS) LazyJsonlIteratorz LazyJsonlIterator provides the ability to read JSON lines as Python dicts. It can also provide the number of lines via __len__ via fast newlines counting. rnrcNcCs||_d|_dSr)rnrpr"rnrrrrqs zLazyJsonlIterator.__init__ccsjd}t|jd}|D] }t|}|V|d7}q Wdn1s$wY|jdur3||_dSdS)Nrrsrt)rrnr rp)r"rwrxryr(rrrrzs   zLazyJsonlIterator.__iter__cCr{rr|r~rrrr rzLazyJsonlIterator.__len__) rKrLrMrNrrqrzrQrrrrrr s  rc@sTeZdZdZdeddfddZedefddZd d Zde fd d Z dddZ dS)LazyManifestIteratora LazyManifestIterator provides the ability to read Lhotse objects from a JSONL file on-the-fly, without reading its full contents into memory. This class is designed to be a partial "drop-in" replacement for ordinary dicts to support lazy loading of RecordingSet, SupervisionSet and CutSet. Since it does not support random access reads, some methods of these classes might not work properly. rnrcNcCst||_dSr)rsourcerrrrrq1zLazyManifestIterator.__init__cCs|jjSr)rrnr~rrrrn4szLazyManifestIterator.pathccstt|jEdHdSr)r+rrr~rrrrz8szLazyManifestIterator.__iter__cC t|jSr)lenrr~rrrr; zLazyManifestIterator.__len__rGcC t||SrrGr"rHrrrrI>rzLazyManifestIterator.__add__rcrG) rKrLrMrNrrqpropertyrnrzrQrrIrrrrr&s rc @sbeZdZdZddddededeeee dfd dfd d Z d d Z d efddZ dddZ dS)rGa@ A thin wrapper over multiple iterators that enables to combine lazy manifests in Lhotse. It iterates all underlying iterables sequentially. It also supports shuffling the sub-iterators when it's iterated over. This can be used to implement sharding (where each iterator is a shard) with randomized shard order. Every iteration of this object will increment an internal counter so that the next time it's iterated, the order of shards is again randomized. .. note:: if any of the input iterables is a dict, we'll iterate only its values. FN) shuffle_itersr/ iteratorsrr/r0rccGsVg|_||_||_d|_|D]}t|tr"|jD]}|j|qq|j|qdSrY)rrr/ num_iters isinstancerGappend)r"rr/ritsub_itrrrrqPs  zLazyIteratorChain.__init__ccsddlm}|j}|jr,|jdurt}n t||j|j}|||jd7_|D]}t |t r9| }|EdHq.dS)Nr resolve_seedrt) lhotse.dataset.dataloadingrrrr/r>rSrrArdictvalues)r"rrr<rrrrrzbs     zLazyIteratorChain.__iter__cCtdd|jDS)Ncs|]}t|VqdSrrrrrrrrsz,LazyIteratorChain.__len__..sumrr~rrrrrzLazyIteratorChain.__len__cCrrrrrrrrIurzLazyIteratorChain.__add__r)rKrLrMrNrrOr r rQrrqrzrrIrrrrrGBs  rGc @steZdZdZdddddededeeee e fd ee e d fd df d d Z ddZ d e fddZdddZdS)r4aO A wrapper over multiple iterators that enables to combine lazy manifests in Lhotse. During iteration, unlike :class:`.LazyIteratorChain`, :class:`.LazyIteratorMultiplexer` at each step randomly selects the iterable used to yield an item. Since the iterables might be of different length, we provide a ``weights`` parameter to let the user decide which iterables should be sampled more frequently than others. When an iterable is exhausted, we will keep sampling from the other iterables, until we exhaust them all, unless ``stop_early`` is set to ``True``. FNrr,rr-r.r/r0rccGsjt||_||_||_t|jdksJd|dur$dgt|j|_n||_t|jt|jks3JdS)Nrtz5There have to be at least two iterables to multiplex.)listrr-r/rr.)r"r-r.r/rrrrrqs z LazyIteratorMultiplexer.__init__c #sddlm}t|j}ddjD}ddtt|Dfdd}|rgtddt tj D\}}|j ||d d d}||}z t |} | VWn t yad |<Yq)w|s,dSdS) NrrcSsg|]}t|qSr)iterrrrr sz4LazyIteratorMultiplexer.__iter__..cSsg|]}dqS)Fr)r_rrrrscsjrt St Sr)r-anyallr exhaustedr"rrshould_continues  z9LazyIteratorMultiplexer.__iter__..should_continuecSs g|] \}\}}|s||fqSrr)ri is_exhaustedwrrrrs  rtr.kT)rrr>rSr/rrangerzip enumerater.choicesnext StopIteration) r"rr<itersractive_indexesactive_weightsidxselecteditemrrrrzs,   z LazyIteratorMultiplexer.__iter__cCr)Ncsrrrrrrrrrz2LazyIteratorMultiplexer.__len__..rr~rrrrrzLazyIteratorMultiplexer.__len__rGcCrrrrrrrrIrzLazyIteratorMultiplexer.__add__r)rKrLrMrNrrOr rr rQrRrrqrzrrIrrrrr4ys& r4c @sfeZdZdZddddddededeeee e fd ee e d fd ee d df d dZ ddZ dS)r9a A variant of :class:`.LazyIteratorMultiplexer` that allows to control the number of iterables that are simultaneously open. It is useful for large-scale data sets where opening multiple file handles in many processes leads to exhaustion of the operating system resources. If the data sets are sharded, it is recommended to pass each shard as a separate iterator when creating objects of this class. It is OK to assign a dataset-level weight to each shard (e.g., if a dataset has a weight of 0.5, assign weight 0.5 to each of its shards). There are several differences between this class and :class:`.LazyIteratorMultiplexer`: * Objects of this class are infinite iterators. * We hold a list of ``max_open_streams`` open iterators at any given time. This list is filled by sampling input iterators with replacement. These differences are necessary to guarantee the weighted sampling property. If we did not sample with replacement or make it infinite, we would simply exhaust highly-weighted iterators towards the beginning of each "epoch" and keep sampling only lowly-weighted iterators towards the end of each "epoch". FNr)r-r.r/r8rr-r.r/r0r8rccGst||_||_||_||_|dus|t|jkrt|j|_t|jdks(J||_|dur8dgt|j|_t|jt|jksDJ|jdusV|jdksXJd|jdSdS)Nrrtzself.max_open_streams=)rrr-r/r8rr.)r"r-r.r/r8rrrrrqs    z+LazyInfiniteApproximateMultiplexer.__init__c#sddlm}t|jfdd}|dgjdgjttj}dtddffdd }tjD]}||q> j |t dkrQndd d d}|}z t |}|VWnt y{||t |}|VYnwqF) aZ Assumptions - we have N streams but can only open M at the time (M < N) - the streams are finite - each stream needs to be "short" to ensure the mux property - each stream may be interpreted as a shard belonging to some larger group of streams (e.g. multiple shards of a given dataset). rrc3sDtttj} j|jddd}j|j|fVq )NTrt)rr)rrrrrr.)indexesr)r<r"rrshuffled_streamss zELazyInfiniteApproximateMultiplexer.__iter__..shuffled_streamsNposrccs$t\}}t||<||<dSr)rr)rsampled_streamsampled_weight)active_streamsr stream_sourcerrsample_new_stream_ats   zILazyInfiniteApproximateMultiplexer.__iter__..sample_new_stream_atTrtr) rrr>rSr/r8rrrQrrrr)r"rrstream_indexesr stream_posrrr)rrr<r"rrrzs:        z+LazyInfiniteApproximateMultiplexer.__iter__)rKrLrMrNrrOr rr rQrRrrqrzrrrrr9s( r9c @sVeZdZdZ  ddededeejddfdd Z d d Z defd d Z dddZ dS)r?z A wrapper over an iterable that enables lazy shuffling. The shuffling algorithm is reservoir-sampling based. See :func:`lhotse.utils.streaming_shuffle` for details. r;Niteratorr=r<rccCs||_||_||_dSr)rr=r<)r"rr=r<rrrrqBs zLazyShuffler.__init__cCsttt|j|j|jdS)N)bufsizer<)rrrr=r<r~rrrrzLszLazyShuffler.__iter__cCrrrrr~rrrrUrzLazyShuffler.__len__rGcCrrrrrrrrIXrzLazyShuffler.__add__)r;Nr) rKrLrMrNrrQr r>rSrqrzrrIrrrrr?;s     r?c@sPeZdZdZdedeegefddfddZdd Z dd d Z de fd dZ dS)r z 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``. rrrcNcCsb||_||_t|jsJd|dt|jtjr+|jjdkr-tds/t ddSdSdSdS)Nz2LazyFilter: 'predicate' arg must be callable (got ).r[zA lambda was passed to LazyFilter: it may prevent you from forking this process. If you experience issues with num_workers > 0 in torch.utils.data.DataLoader, try passing a regular function instead.) rrcallablertypes LambdaTyperKrwarningswarn)r"rrrrrrqcs"   zLazyFilter.__init__cCst|j|jSr)r$rrr~rrrrztrzLazyFilter.__iter__rGcCrrrrrrrrIwrzLazyFilter.__add__cCtd)Na$LazyFilter does not support __len__ because it would require iterating over the whole iterator, which is not possible in a lazy fashion. If you really need to know the length, convert to eager mode first using `.to_eager()`. Note that this will require loading the whole iterator into memory. TypeErrorr~rrrrzzLazyFilter.__len__r) rKrLrMrNrrrrOrqrzrIrQrrrrrr \s   r c @sfeZdZdZ ddedeegefdeeegefddfddZ d d Z de fd d Z dddZ dS)r'a A wrapper over an iterable that enables lazy function evaluation on each item. It works like Python's `map` built-in by applying a callable ``fn`` to each element ``x`` and yielding the result of ``fn(x)`` further. New in Lhotse v1.22.0: ``apply_fn`` can be provided to decide whether ``fn`` should be applied to a given example or not (in which case it will return it as-is, i.e., it does not filter). Nrr&apply_fnrccCs||_||_||_t|jsJd|d|jdur(t|js(Jd|dt|jtjr5|jjdksEt|jtjrL|jjdkrNtsPt ddSdSdSdS)Nz+LazyMapper: 'fn' arg must be callable (got rz1LazyMapper: 'apply_fn' arg must be callable (got rzA lambda was passed to LazyMapper: it may prevent you from forking this process. If you experience issues with num_workers > 0 in torch.utils.data.DataLoader, try passing a regular function instead.) rr&rrrrrrKrZrr)r"rr&rrrrrqs.      zLazyMapper.__init__ccsT|jdurt|j|jEdHdS|jD]}||r"||}n|}|VqdSr)rr+r&r)r"rr*rrrrzs    zLazyMapper.__iter__cCrrrr~rrrrrzLazyMapper.__len__rGcCrrrrrrrrIrzLazyMapper.__add__rr)rKrLrMrNrrrr rOrqrzrQrrIrrrrr's    r'c@sBeZdZdZdeddfddZddZdd d Zdefd d Z dS) LazyFlattenera A wrapper over an iterable of collections that flattens it to an iterable of items. Example:: >>> list_of_cut_sets: List[CutSet] = [CutSet(...), CutSet(...)] >>> list_of_cuts: List[Cut] = list(LazyFlattener(list_of_cut_sets)) rrcNcCs ||_dSrr)r"rrrrrqrzLazyFlattener.__init__ccs|jD]}|EdHqdSrr)r"cutsrrrrzs  zLazyFlattener.__iter__rGcCrrrrrrrrIrzLazyFlattener.__add__cCr)Na'LazyFlattener does not support __len__ because it would require iterating over the whole iterator, which is not possible in a lazy fashion. If you really need to know the length, convert to eager mode first using `.to_eager()`. Note that this will require loading the whole iterator into memory.rr~rrrrrzLazyFlattener.__len__r) rKrLrMrNrrqrzrIrQrrrrrrs   rc @sReZdZdZ ddedeededdfdd Zd d Z defd d Z dddZ dS)rEz_ A wrapper over an iterable that enables to repeat it N times or infinitely (default). NFrrCrDrccCs2||_||_||_|jdus|jdksJdSdSrY)rrCrD)r"rrCrDrrrrqs zLazyRepeater.__init__ccsd}|jdus ||jkr>|jr|j}n t|jtt|d}d}|D]}d}|Vq"|s.dS|d7}|jdus ||jks dSdS)Nr)rFTrt)rCrDrr'rattach_repeat_idx_to_id)r"epochr at_least_oncerrrrrzs zLazyRepeater.__iter__cCs0|jdurtdt|jdt|j|jS)Nzobject of type 'z' is an infinite iterator)rCrrrKrrr~rrrrs zLazyRepeater.__len__rGcCrrrrrrrrIrzLazyRepeater.__add__rJr) rKrLrMrNrr rQrOrqrzrrIrrrrrEs rEc@sJeZdZdZdedededdfddZd d Zdd d ZdefddZ dS) LazySlicerzZ A wrapper over an iterable that enables selecting k-th element every n elements. rrnrcNcCs4||_||ksJd|d|d||_||_dS)NzKWhen selecting k-th element every n elements, k must be less than n (got k=z n=r)rrr)r"rrrrrrrqs   zLazySlicer.__init__ccs0t|jD]\}}||j|jkr|VqdSr)rrrr)r"rrrrrrzs zLazySlicer.__iter__rGcCrrrrrrrrIrzLazySlicer.__add__cCr)Na$LazySlicer does not support __len__ because it would require iterating over the whole iterator, which is not possible in a lazy fashion. If you really need to know the length, convert to eager mode first using `.to_eager()`. Note that this will require loading the whole iterator into memory.rr~rrrrrzLazySlicer.__len__r) rKrLrMrNrrQrqrzrIrrrrrrs  rrrcCs&t|ds|St||jd|dS)Nid_repeat)r)hasattrrr)rrrrrrs rrncCsddd}t|dks dnd}t||}tdd||jD}Wd|S1s+wY|S) z Counts newlines in a file using buffered chunk reads. The fastest possible option in Python according to: https://stackoverflow.com/a/68385697/5285891 (This is a slightly modified variant of that answer.) css(|d}|r|V|d}|sdSdS)Nir)readerbrrr _make_gen-s  z&count_newlines_fast.._make_gen-rbrscss|]}|dVqdS) N)count)rbufrrrr5sz&count_newlines_fast..N)strrrread)rnr read_moderxrrrrr}%s  r})2rfr>rr contextlibr functoolsrtypingrrrrrr r r lhotse.serializationr r rr lhotse.utilsrrrrrrrT frozensetrOrZrjrlrmrrrGr4r9r?r r'rrErrQrr}rrrrs>  (  7G{!'8)