o æS™iŒZã@sNdZddlZddlZddlZddlmZddlmZmZm Z m Z m Z m Z m Z mZddlZddlmZmZddlmZddlmZmZmZ      d)d ed ed e ed ededededededefdd„ZGdd„dƒZGdd„dƒZ     d*deee efdedededed ef d!d"„Zdefd#d$„Z d+d%d&„Z!Gd'd(„d(ƒZ"dS),u$ High-level architecture of the Lhotse+WebDataset solution. Read the documentation of the items below to understand each component better. ┌────────────────────────────────────────────────────────────────────────┠│┌──────────────────────────────────────────────────────────────────────â”│ ││ Training loop ││ │└──────────────────────────────────────────────────────────────────────┘│ │ │ │ │ â–¼ │ │ ┌─────────────────────────────┠│ │ │ torch.utils.data.DataLoader │ │ │ └─────────────────────────────┘ Main process│ └────────────────────────────────────┬───────────────────────────────────┘ ┌─────────────────────────────┼───────────────────────────────┠▼ ┌─────────────────────▼───────────────────────┠▼ ┌─────────┠│ ┌─────────┠Sub-process #i │ ┌─────────┠│Worker #1│ │ │Worker #i│ │ │Worker #N│ └─────────┘ │ └─────────┘ │ └─────────┘ │ │ │ │ â–¼ │ │ ┌────────────────────────┠│ │ │ IterableDatasetWrapper │ │ │ └────────────────────────┘ │ │ │ │ │ ┌─────────┴──────┠│ │ â–¼ â–¼ │ │ ┌─────────────────┠┌───────────┠│ │ │Map-style Dataset│ │ Sampler │ │ │ │ (task-specific) │ │ │ │ │ └─────────────────┘ └───────────┘ │ │ │ │ │ â–¼ │ │ ┌───────────┠│ │ │ CutSet │ │ │ └───────────┘ │ │ │ │ │ â–¼ │ │ ┌────────────────────────┠│ │ │Lazy WebDataset Iterator│ │ │ │(discards shard_idx % N)│ │ │ └────────────────────────┘ │ │ │ │ │ ┌───────────┼───────────┠│ │ â–¼ â–¼ â–¼ │ │ ┌────────┠┌────────┠┌────────â”│ │ │Shard #1│ │Shard #j│ │Shard #M││ │ └────────┘ └────────┘ └────────┘│ └─────────────────────────────────────────────┘ éN)Úpartial)ÚCallableÚDictÚ GeneratorÚIterableÚListÚOptionalÚSequenceÚUnion)ÚCutSetÚMonoCut©ÚLazyIteratorChain)ÚPathlikeÚis_module_availableÚsuppress_and_warnTÚflacÚcutsÚ output_pathÚ shard_sizeÚverboseÚ audio_formatÚ load_audioÚ load_featuresÚ load_customÚfault_tolerantÚreturnc  CsÖt|||||||d} d} d} | 9tjjd| d } |D]} | d7} |  | ¡}| t|ƒ7} |  ¡qWdƒn1s>wYWdƒn1sMwY| j}t d| ›d| ›d |›d | | ›d ¡|S) að Saves the CutSet metadata along with audio/features data into a WebDataset archive. The audio and feature data is read, decoded, and encoded into ``audio_format`` for audio, lilcom for features and arrays with floating point type, and pickle for all other dtypes. The intended use of this function is to speed up the I/O in training data pipelines by converting random access reads to sequential access reads. Supported values for ``audio_format`` are the same as for the ``format`` argument in ``torchaudio.save`` function with ``sox_io`` backend. If ``shard_size`` is specified, we will leverage WebDataset's ``ShardWriter`` to create multiple tarballs with ``shard_size`` items per shard. In that mode, we expect that ``output_path`` contains a pattern like "/path/to/shard-%06d.tar", which will be internally expanded with the shard index. Returns number of written shards if sharding is enabled, otherwise 0. .. note: By default, we'll skip cuts which failed to load for any reason and proceed with exporting. To raise an exception and stop, set ``fault_tolerant=False``. **Examples** Export cuts with audio, features, and all custom data to a single tarball, converting audio to FLACs:: >>> cuts = CutSet.from_jsonl_lazy("data/cuts-train.jsonl") >>> n_shards = export_to_webdataset( ... cuts=cuts, ... output_path="data/cuts-train.tar", ... audio_format="flac", ... ) Export cuts with audio, features, and all custom data to a directory with shards counting 10000 cuts each, converting audio to SPHERE (sph):: >>> cuts = CutSet.from_jsonl_lazy("data/cuts-train.jsonl") >>> n_shards = export_to_webdataset( ... cuts=cuts, ... output_path="data/cuts-train-wds/shard-%06d.tar", ... shard_size=10000, ... audio_format="sph", ... ) The same, but export cuts with only the features being read into memory (recording and custom data still refers to external storage):: >>> cuts = CutSet.from_jsonl_lazy("data/cuts-train.jsonl") >>> n_shards = export_to_webdataset( ... cuts=cuts, ... output_path="data/cuts-train-wds/shard-%06d.tar", ... shard_size=10000, ... load_audio=False, ... load_custom=False, ... ) Export cuts to sharded tarballs stored in the cloud (in this example AWS S3, using AWS CLI):: >>> cuts = CutSet.from_jsonl_lazy("data/cuts-train.jsonl") >>> n_shards = export_to_webdataset( ... cuts=cuts, ... output_path="pipe:aws s3 cp - s3://my-bucket/data/shard-%06d.tar", ... shard_size=10000, ... ) )Ú path_or_urlrrrrrrrzCreating WebDataset tarball(s))ÚdescÚdisableéNz Exported z cuts out of z total into z shards (there were z cuts with errors).) ÚWebdatasetWriterÚtqdmÚautoÚwriteÚintÚupdateÚnum_shards_writtenÚloggingÚinfo)rrrrrrrrrÚwriterÚtotalÚokÚpbarÚcutÚsuccessr'©r0úM/home/ubuntu/.local/lib/python3.10/site-packages/lhotse/dataset/webdataset.pyÚexport_to_webdataset@s>Mù  ÿ   üý€ ÿÿr2c@sŽeZdZdZ      ddedeededed ed ed ed dfd d„Z ddd„Z ddd„Z ddd„Z de d efdd„Zd eefdd„ZdS)r!a= Saves the CutSet metadata along with audio/features data into a WebDataset archive. The audio and feature data is read, decoded, and encoded into ``audio_format`` for audio, lilcom for features and arrays with floating point type, and pickle for all other dtypes. The intended use of this function is to speed up the I/O in training data pipelines by converting random access reads to sequential access reads. Supported values for ``audio_format`` are the same as for the ``format`` argument in ``torchaudio.save`` function with ``sox_io`` backend. If ``shard_size`` is specified, we will leverage WebDataset's ``ShardWriter`` to create multiple tarballs with ``shard_size`` items per shard. In that mode, we expect that ``output_path`` contains a pattern like "/path/to/shard-%06d.tar", which will be internally expanded with the shard index. Returns number of written shards if sharding is enabled, otherwise 0. .. note: By default, we'll skip cuts which failed to load for any reason and proceed with exporting. To raise an exception and stop, set ``fault_tolerant=False``. **Example** Export cuts with audio, features, and all custom data to a tarball shards with 500 cuts each:: >>> cuts = CutSet.from_jsonl_lazy("data/cuts-train.jsonl") >>> with WebdatasetWriter("data/tars/shard-%06d.tar", shard_size=500) as writer: ... for cut in cuts: ... writer.write(cut) >>> output_paths = writer.output_manifest_paths() See also: :func`.export_to_webdataset` NrTrrrrrrrrc Cs˜tdƒstdƒ‚ddlm}||_||_||_||_||_||_ ||_ |jdur:|jdks/J‚t t |j|jd|_ nt ||jƒ|_ d|_d|_d|_dS)NÚ webdatasetú&Please 'pip install webdataset' first.r©Ú TarWriter)Úmaxcount)rÚ ImportErrorr3r6rrrrrrrrÚ ShardWriterÚwriter_init_fnr*r'Úfinished) Úselfrrrrrrrr6r0r0r1Ú__init__Ïs&     ÿ zWebdatasetWriter.__init__cCs| ¡|_d|_|S)NF)r:r*r;©r<r0r0r1Ú __enter__ós zWebdatasetWriter.__enter__cOs | ¡dS©N©Úclose)r<ÚargsÚkwargsr0r0r1Ú__exit__øs zWebdatasetWriter.__exit__cCs*t|jtƒr |jj|_|j ¡d|_dS)NT)Ú isinstancer*r9Úshardr'rBr;r>r0r0r1rBûs    zWebdatasetWriter.closeÚmanifestcCsrtt|jd(|j|j|j|j|jd}t  |  ¡¡}|j   |j |dœ¡ WdƒdS1s2wYdS)zÒ Converts a Cut to a dict, pickles it, and then stores into a tarfile. :param manifest: the manifest to be written. :return: bool indicating whether the writing was successful. )Úenabled)rrrr)Ú__key__ÚdataNTF)rÚ ExceptionrÚmove_to_memoryrrrrÚpickleÚdumpsÚto_dictr*r$Úid)r<rHr.rKr0r0r1r$sü ÷ zWebdatasetWriter.writecsJˆjdur tdƒ‚ˆjstdƒ‚ˆjdurˆjgS‡fdd„tˆjƒDƒS)a Return the a list of paths/urls where the data was written. The list can be used directly to initialize :class:`.LazyWebdatasetIterator` or :meth:`lhotse.cut.CutSet.from_webdataset`. Useful when writing into shards with a specified pattern. Nz(The writer has not written anything yet.zWThe writer was not closed -- call writer.close() first, or use it as a context manager.csg|]}ˆj|‘qSr0)r)Ú.0Úir>r0r1Ú $sz:WebdatasetWriter.output_manifest_paths..)r;Ú ValueErrorr'rÚranger>r0r>r1Úoutput_manifest_pathss ÿ z&WebdatasetWriter.output_manifest_paths)NrTTTT)rr!©rN)Ú__name__Ú __module__Ú __qualname__Ú__doc__rrr%ÚstrÚboolr=r?rErBr r$rrWr0r0r0r1r!¬s<%øþýüûúùø ÷ $  r!c@s¨eZdZdZdeeeefddfdd„Zdeddfdd „Z dd d „Z de fd d „Z de ddfdd„Zddd„Zdd„Zdd„Zdefdd„Zdd„Zdefdd„ZdS)ÚLazyWebdatasetIteratoraF LazyWebdatasetIterator provides the ability to read Lhotse objects from a WebDataset tarball 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. The behaviour of the underlying ``WebDataset`` instance can be customized by providing its kwargs directly to the constructor of this class. ÚsourcerNcKs tdƒstdƒ‚||_||_dS©Nr3r4)rr8r`Ú wds_kwargs)r<r`rbr0r0r1r=5s zLazyWebdatasetIterator.__init__ÚepochcCs||jd<dS)Nrc)rb)r<rcr0r0r1Ú set_epoch>óz LazyWebdatasetIterator.set_epochcCs6tdƒstdƒ‚t|jfi|j¤Ž|_t|jƒ|_dSra)rr8Úmini_webdatasetr`rbÚ_dsÚiterÚ_ds_iterr>r0r0r1Ú_resetAszLazyWebdatasetIterator._resetcCs|j|jdœ}|S)zò Store the state for pickling -- we'll only store the path + kwargs, and re-initialize this iterator when unpickled. This is necessary to transfer this object across processes for PyTorch's DataLoader workers. ©r`rbrk©r<Ústater0r0r1Ú __getstate__Hsz#LazyWebdatasetIterator.__getstate__rmcCs|j |¡dS)z!Restore the state when unpickled.N)Ú__dict__r&rlr0r0r1Ú __setstate__Qsz#LazyWebdatasetIterator.__setstate__cCs | ¡|Sr@)rjr>r0r0r1Ú__iter__UszLazyWebdatasetIterator.__iter__cCs:ddlm}t|jƒ}t |d¡}||ƒ}|d|_|S)Nr)Údeserialize_itemrKÚ__url__)Úlhotse.serializationrrÚnextrirNÚloadsÚ shard_origin)r<rrÚ data_dictrKÚitemr0r0r1Ú__next__Ys   zLazyWebdatasetIterator.__next__ccs|EdHdSr@r0r>r0r0r1Úvaluesbs€zLazyWebdatasetIterator.valuescCódd„|DƒS)Ncss|]}|jVqdSr@©rQ©rRryr0r0r1Ú fs€z.LazyWebdatasetIterator.keys..r0r>r0r0r1ÚkeyserezLazyWebdatasetIterator.keyscCr|)Ncss|]}|j|fVqdSr@r}r~r0r0r1ris€z/LazyWebdatasetIterator.items..r0r>r0r0r1ÚitemshrezLazyWebdatasetIterator.itemscCs t||ƒSr@r )r<Úotherr0r0r1Ú__add__ks zLazyWebdatasetIterator.__add__rX)rr_)rYrZr[r\r rr r=r%rdrjÚdictrnrrprqrzr{r]r€rrrƒr0r0r0r1r_'s" ÿ þ   r_FÚurlsrcÚshuffle_shardsÚsplit_by_workerÚ split_by_nodeÚignore_error_shardscCs¦tdƒstdƒ‚ddlm}m}m}ddlm} ddlm} ddlm} m } |||dƒ} |r4|   | ¡|r;|   | ¡|rE|   t |d ¡|   | |rL| n|d ¡| S) a# Return a pipeline for WebDataset-style data files. This is a convenience function for constructing a partial pipeline that reads from a set of sharded tar files, extracts the individual files, and groups them together into samples (dictionaries). You can use all the methods from `Composable` (`then`, `compose`) and from `Shorthands` (`batched`, `unbatched`, `decode`, `shuffle`, etc.) on the result. .. note: This is a reduced version of ``webdataset.WebDataset`` function, that only uses the functionalities relevant to Lhotse, and makes it possible to disable the node/worker splitting. :param urls: the source URLs: a string or a list. :param epoch: epoch number (used only when ``shuffle_shards`` is enabled). :param shuffle_shards: shuffle the shards if True. Only takes effect when ``urls`` is a list of shard paths/urls. :param split_by_worker: DEPRECATED: always acts as if True. If True, shards are split per DataLoader worker subprocesses, otherwise each dataloader worker will yield the same data. Only takes effect when ``urls`` is a list of shard paths/urls. :param split_by_node: if True, shards are split per node in DDP training, otherwise on each node we'll yield the same data. Only takes effect when ``urls`` is a list of shard paths/urls. :param ignore_error_shards: when ``True``, we tell WebDataset to ignore shards that failed during loading and emit a warning. When ``False``, we won't catch the exceptions. r3r4r)Ú DataPipelineÚSimpleShardListÚreraise_exception)rˆ)r‡)Útarfile_to_samplesÚwarn_and_continue)r…)rc)Úhandler) rr8r3rŠr‹rŒrˆr‡rrŽÚappendÚcreate_shard_shuffler)r…rcr†r‡rˆr‰rŠr‹rŒÚsplit_by_node_Úsplit_by_worker_rrŽÚwdsr0r0r1rfos&%     ÿÿrfcCs&ddlm}Gdd„d|ƒ}||dS)Nr)Ú PipelineStagec@seZdZddd„Zdd„ZdS) z-create_shard_shuffler..detshuffle_allréÿÿÿÿcSs||_||_dSr@)Úseed_Úepoch_)r<r—r˜r0r0r1r=¯s z6create_shard_shuffler..detshuffle_all.__init__cSsB|jd7_t ¡}| t|j|jfƒ¡t|ƒ}| |¡|S)Nr )r˜ÚrandomÚRandomÚseedÚhashr—ÚlistÚshuffle)r<ÚsrcÚrngrr0r0r1Úrun³s  z1create_shard_shuffler..detshuffle_all.runN)rr–)rYrZr[r=r¡r0r0r0r1Údetshuffle_all®s  r¢)r˜)r3r•)rcr•r¢r0r0r1r‘«s  r‘ccs|EdHdS)zn Helper fn that works normally with single-node training, but duplicates data in multi-node training. Nr0)rŸÚgroupr0r0r1Ú/_single_node_or_multi_node_with_duplicated_data¾s€r¤c @sleZdZdZ    ddededed eed ef d d „Z d d„Z dd„Z dd„Z dd„Z dd„Zdd„ZdS)r9aª Like ``webdataset.TarWriter`` but splits into multiple shards. Note: this implementation is copied from webdataset and adapted to allow shard writing using the "pipe:" notation. E.g., this is possible:: >>> writer = ShardWriter("pipe:gzip -c > data/shard-%06d.tar.gz") Source: https://github.com/webdataset/webdataset/blob/ccfe88086cdb21a0dc23a6454ce3e3723b6b8033/webdataset/writer.py#L359 é †çÀ ZæANrÚpatternr7ÚmaxsizeÚpostÚ start_shardcKsvtdƒstdƒ‚d|_||_||_||_||_d|_||_||_ |j dks)Jdƒ‚d|_ d|_ d|_ d|_ | ¡dS)aCreate a ShardWriter. :param pattern: output file pattern :param maxcount: maximum number of records per shard (Default value = 100000) :param maxsize: maximum size of each shard (Default value = 3e9) :param kw: other options passed to TarWriter r3r4r Nú-z3Dash '-' is not an allowed pattern for ShardWriter.r)rr8rÚkwr7r¨r©Ú tarstreamrGr§r+ÚcountÚsizeÚfnameÚ next_stream)r<r§r7r¨r©rªr¬r0r0r1r=Òs$ ÿ zShardWriter.__init__cCs|ddlm}| ¡|j|j|_|jr$td|j|jd|j d|j ƒ|jd7_||jfi|j ¤Ž|_ d|_d|_ dS)z.Close the current stream and move to the next.rr5z # writingz%.1f GBgeÍÍAr N) Úwebdataset.writerr6Úfinishr§rGr°rÚprintr®r¯r+r¬r­)r<r6r0r0r1r±÷s  û zShardWriter.next_streamcCsd|jdus|j|jks|j|jkr| ¡|j |¡}|jd7_|jd7_|j|7_dS)zBWrite a sample. :param obj: sample to be written Nr )r­r®r7r¯r¨r±r$r+)r<Úobjr¯r0r0r1r$ s    zShardWriter.writecCsF|jdur!|j ¡|jdusJ‚t|jƒr| |j¡d|_dSdS)z'Finish all writing (use close instead).N)r­rBr°Úcallabler©r>r0r0r1r³s     ûzShardWriter.finishcCs| ¡|`|`|`|`dS)zClose the stream.N)r³r­rGr®r¯r>r0r0r1rB#s zShardWriter.closecCs|S)zEnter context.r0r>r0r0r1r?+szShardWriter.__enter__cOs | ¡dS)z Exit context.NrA)r<rCr¬r0r0r1rE/s zShardWriter.__exit__)r¥r¦Nr)rYrZr[r\r]r%Úfloatrrr=r±r$r³rBr?rEr0r0r0r1r9Ås.úþýüû ú%  r9)NTrTTTT)rFTFTr@)#r\r(rNr™Ú functoolsrÚtypingrrrrrrr r r"Úlhotser r Ú lhotse.lazyrÚ lhotse.utilsrrrr%r^r]r2r!r_rfr‘r¤r9r0r0r0r1Úsz2 ( ÷ÿþýüûúùø ÷ öl{Júÿþýüû ú<