o 2wi2 @sddlZddlmZddlmZmZmZmZmZm Z m Z m Z m Z ddl mZddlmZddlmZmZmZmZmZddlmZddlmZdd lmZmZmZGd d d eZd ed e dee e e!efddffddZ"dS)NPath) CallableDict GeneratorListLiteralOptionalSequenceTupleUnion)Cut) resolve_seed)DillableLazyIteratorChainLazyJsonlIteratorLazyManifestIteratorcount_newlines_fast)extension_contains) TarIterator)Pathlikeexactly_one_not_nullifnonec@seZdZdZ       d"deeeeefdeede d e d e d e e e d e d fdeee egefddfddZd#deeeeeffddZdefddZdedefddZdedefddZddZde fddZd$d d!ZdS)%LazySharIteratora LazySharIterator reads cuts and their corresponding data from multiple shards, also recognized as the Lhotse Shar format. Each shard is numbered and represented as a collection of one text manifest and one or more binary tarfiles. Each tarfile contains a single type of data, e.g., recordings, features, or custom fields. Given an example directory named ``some_dir``, its expected layout is ``some_dir/cuts.000000.jsonl.gz``, ``some_dir/recording.000000.tar``, ``some_dir/features.000000.tar``, and then the same names but numbered with ``000001``, etc. There may also be other files if the cuts have custom data attached to them. The main idea behind Lhotse Shar format is to optimize dataloading with sequential reads, while keeping the data composition more flexible than e.g. WebDataset tar archives do. To achieve this, Lhotse Shar keeps each data type in a separate archive, along a single CutSet JSONL manifest. This way, the metadata can be investigated without iterating through the binary data. The format also allows iteration over a subset of fields, or extension of existing data with new fields. As you iterate over cuts from ``LazySharIterator``, it keeps a file handle open for the JSONL manifest and all of the tar files that correspond to the current shard. The tar files are read item by item together, and their binary data is attached to the cuts. It can be normally accessed using methods such as ``cut.load_audio()``. We can simply load a directory created by :class:`~lhotse.shar.writers.shar.SharWriter`. Example:: >>> cuts = LazySharIterator(in_dir="some_dir") ... for cut in cuts: ... print("Cut", cut.id, "has duration of", cut.duration) ... audio = cut.load_audio() ... fbank = cut.load_features() :class:`.LazySharIterator` can also be initialized from a dict, where the keys indicate fields to be read, and the values point to actual shard locations. This is useful when only a subset of data is needed, or it is stored in different directories. Example:: >>> cuts = LazySharIterator({ ... "cuts": ["some_dir/cuts.000000.jsonl.gz"], ... "recording": ["another_dir/recording.000000.tar"], ... "features": ["yet_another_dir/features.000000.tar"], ... }) ... for cut in cuts: ... print("Cut", cut.id, "has duration of", cut.duration) ... audio = cut.load_audio() ... fbank = cut.load_features() We also support providing shell commands as shard sources, inspired by WebDataset. Example:: >>> cuts = LazySharIterator({ ... "cuts": ["pipe:curl https://my.page/cuts.000000.jsonl.gz"], ... "recording": ["pipe:curl https://my.page/recording.000000.tar"], ... }) ... for cut in cuts: ... print("Cut", cut.id, "has duration of", cut.duration) ... audio = cut.load_audio() Finally, we allow specifying URLs or cloud storage URIs for the shard sources. We defer to ``smart_open`` library to handle those. Example:: >>> cuts = LazySharIterator({ ... "cuts": ["s3://my-bucket/cuts.000000.jsonl.gz"], ... "recording": ["s3://my-bucket/recording.000000.tar"], ... }) ... for cut in cuts: ... print("Cut", cut.id, "has duration of", cut.duration) ... audio = cut.load_audio() :param fields: a dict whose keys specify which fields to load, and values are lists of shards (either paths or shell commands). The field "cuts" pointing to CutSet shards always has to be present. :param in_dir: path to a directory created with ``SharWriter`` with all the shards in a single place. Can be used instead of ``fields``. :param split_for_dataloading: bool, by default ``False`` which does nothing. Setting it to ``True`` is intended for PyTorch training with multiple dataloader workers and possibly multiple DDP nodes. It results in each node+worker combination receiving a unique subset of shards from which to read data to avoid data duplication. This is mutually exclusive with ``seed='randomized'``. :param shuffle_shards: bool, by default ``False``. When ``True``, the shards are shuffled (in case of multi-node training, the shuffling is the same on each node given the same seed). :param seed: When ``shuffle_shards`` is ``True``, we use this number to seed the RNG. Seed can be set to ``'randomized'`` in which case we expect that the user provided :func:`lhotse.dataset.dataloading.worker_init_fn` as DataLoader's ``worker_init_fn`` argument. It will cause the iterator to shuffle shards differently on each node and dataloading worker in PyTorch training. This is mutually exclusive with ``split_for_dataloading=True``. Seed can be set to ``'trng'`` which, like ``'randomized'``, shuffles the shards differently on each iteration, but is not possible to control (and is not reproducible). ``trng`` mode is mostly useful when the user has limited control over the training loop and may not be able to guarantee internal Shar epoch is being incremented, but needs randomness on each iteration (e.g. useful with PyTorch Lightning). :param stateful_shuffle: bool, by default ``False``. When ``True``, every time this object is fully iterated, it increments an internal epoch counter and triggers shard reshuffling with RNG seeded by ``seed`` + ``epoch``. Doesn't have any effect when ``shuffle_shards`` is ``False``. :param cut_map_fns: optional sequence of callables that accept cuts and return cuts. It's expected to have the same length as the number of shards, so each function corresponds to a specific shard. It can be used to attach shard-specific custom attributes to cuts. See also: :class:`~lhotse.shar.writers.shar.SharWriter` NFT*fieldsin_dirsplit_for_dataloadingshuffle_shardsstateful_shuffleseed randomizedtrng cut_map_fnsreturnc st||s Jd|r|dksJd|_|_|_|_d_d_|dur/|n|t j d_ j D]%}t j |j ksdJdj d|dt j |d j |q?fd d t j D_t|dgj _dS) NzITo read Lhotse Shar format, provide either 'in_dir' or 'fields' argument.r!zyError: seed='randomized' and split_for_dataloading=True are mutually exclusive options as they would result in data loss.rcutsz Expected z shards available for field 'z ' but found z: cs"g|] fddjDqS)csi|] }|j|qSstreams).0field)self shard_idxr&U/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/lhotse/shar/readers/lazy.py sz8LazySharIterator.__init__...r')r)r+)r,r- sz-LazySharIterator.__init__..)rrrrr epoch_len_init_from_dir_init_from_inputslenr( num_shardsrrangeshardsrr#) r+rrrrrr r#r*r&r/r-__init__s6     , zLazySharIterator.__init__cCs4d|vsJdt||_|jd||_dS)Nr%zOTo initialize Shar reader, please provide the value for key 'cuts' in 'fields'.)setkeysrremover()r+rr&r&r-r4s   z"LazySharIterator._init_from_inputscst||_t|jd}tdd|D|_d|jvsJ|jddtdd|Di|_|jD]tfdd|D|j<q3dS)N*css |] }|jddVqdS.rN)stemsplitr)pr&r&r- sz2LazySharIterator._init_from_dir..r%css2|]}|jdddkrtd|r|VqdS)r?rr%z.jsonlN)namerArrBr&r&r-rDsc3s(|]}|jddkr|VqdSr>)rErArBr*r&r-rDs) rrlistglobr:rr<sortedr()r+r all_pathsr&rFr-r3s     zLazySharIterator._init_from_dirr8cCs&ddlm}m}|jr|||S|S)N) split_by_nodesplit_by_worker)utilsrLrMr)r+r8rLrMr&r&r-_maybe_split_for_dataloadings z-LazySharIterator._maybe_split_for_dataloadingcCs<|jr|}t|j}|jr||j7}t|||SN) rcopyrr rr1randomRandomshuffle)r+r8r r&r&r-_maybe_shuffle_shardss  z&LazySharIterator._maybe_shuffle_shardsc cs@|j|j}}||}||}|dur ||}||}t||D]q\}}t|d}dd|D}dd|D}t|g|RD]J^}} t|| D]*\} \} } | duraqVt | j | j |j kszJd|j d| d| t || | qV|d|_|j|_|dur||}|VqKq%|jd7_dS) Nr%cSsi|] \}}|dkr||qS)r%r&r)r*pathr&r&r-r.sz-LazySharIterator.__iter__..cSs4i|]\}}|td|rt|ntt||dqS)z.tarrF)rr_jsonl_tar_adaptorrrVr&r&r-r.s zMismatched IDs: cut ID is 'z' but found data with name 'z ' fsor field rK)r8r#rUrOzipritemsvaluesr;strparentr@idsetattr shard_originr1 shar_epoch) r+r8map_fnsshard cut_map_fnr% field_paths field_iterscut field_datar*maybe_manifest data_pathr&r&r-__iter__sB      zLazySharIterator.__iter__cCs*|jdurtdd|jdD|_|jS)Ncss|]}t|VqdSrP)rrBr&r&r-rDsz+LazySharIterator.__len__..r%)r2sumr(r/r&r&r-__len__s zLazySharIterator.__len__rcCs t||SrP)r)r+otherr&r&r-__add__s zLazySharIterator.__add__)NNFFTrNrP)r$r)__name__ __module__ __qualname____doc__r rr\r rboolr intrrr r9r4r3rrOrUrkrmror&r&r&r-rsBq , /r jsonl_iterr*r$ccsB|D]}t|dd}||vrd}n||}||fVqdS)z_ Used to adapt the iteration output of LazyJsonlIterator to mimic that of TarIterator. cut_idz.dummyNr)rvr*item pseudo_pathr&r&r-rX!s rX)#rRpathlibrtypingrrrrrr r r r lhotse.cutr lhotse.dataset.dataloadingr lhotse.lazyrrrrrlhotse.serializationrlhotse.shar.readers.tarr lhotse.utilsrrrrr\dictrXr&r&r&r-s& ,