o SiO5@sddlZddlZddlmZddlmZmZmZmZm Z m Z m Z m Z m Z ddlZddlmZddlmZmZddlmZddlmZmZddlmZdd lmZdd lmZdd lm Z e e e!fZ"Gd d d ej#j$j%eZ&GdddZ'dS)N)Path) CallableDict GeneratorIterableListOptionalSequenceTupleUnion) compose_left)CutSetSeconds)deserialize_cut)get_rankget_world_size)SamplingDiagnostics)Dillable)decode_json_line)PathlikecseZdZdZ     ddeeeeeefdedede e de ede ed e e d e e d dffd d Z de egefd dfddZd efddZded dfddZd eeddffddZd efddZZS)StatelessSamplera] An infinite and stateless cut sampler that selects data at random from one or more cut manifests. The main idea is to make training resumption easy while guaranteeing the data seen each time by the model is shuffled differently. It discards the notion of an "epoch" and it never finishes iteration. It makes no strong guarantees about avoiding data duplication, but in practice you would rarely see duplicated data. The recommended way to use this sampler is by placing it into a dataloader worker subprocess with Lhotse's :class:``~lhotse.dataset.iterable_dataset.IterableDatasetWrapper``, so that each worker has its own sampler replica that uses a slightly different random seed:: >>> import torch >>> import lhotse >>> dloader = torch.utils.data.DataLoader( ... lhotse.dataset.iterable_dataset.IterableDatasetWrapper( ... dataset=lhotse.dataset.K2SpeechRecognitionDataset(...), ... sampler=StatelessSampler(...), ... ), ... batch_size=None, ... num_workers=4, ... ) This sampler's design was originally proposed by Dan Povey. For details see: https://github.com/lhotse-speech/lhotse/issues/1096 Example 1: Get a non-bucketing :class:``.StatelessSampler``:: >>> sampler = StatelessSampler( ... cuts_paths=["data/cuts_a.jsonl", "data/cuts_b.jsonl"], ... index_path="data/files.idx", ... max_duration=600.0, ... ) Example 2: Get a bucketing :class:``.StatelessSampler``:: >>> sampler = StatelessSampler( ... cuts_paths=["data/cuts_a.jsonl", "data/cuts_b.jsonl"], ... index_path="data/files.idx", ... max_duration=600.0, ... num_buckets=50, ... quadratic_duration=30.0, ... ) Example 3: Get a bucketing :class:``.StatelessSampler`` with scaled weights for each cutset:: >>> sampler = StatelessSampler( ... cuts_paths=[ ... ("data/cuts_a.jsonl", 2.0), ... ("data/cuts_b.jsonl", 1.0), ... ], ... index_path="data/files.idx", ... max_duration=600.0, ... num_buckets=50, ... quadratic_duration=30.0, ... ) .. note:: This sampler works only with uncompressed jsonl manifests, as it creates extra index files with line byte offsets to quickly find and sample JSON lines. This means this sampler will not work with Webdataset and Lhotse Shar data format. :param cuts_paths: Path, or list of paths, or list of tuples of (path, scale) to cutset files. :param index_path: Path to a file that contains the index of all cutsets and their line count (will be auto-created the first time this object is initialized). :param base_seed: Int, user-provided part of the seed used to initialize the RNG for sampling (each node and worker are still going to produce different results). When continuing the training it should be a function of the number of training steps to ensure the model doesn't see identical mini-batches again. :param max_duration: Maximum total number of audio seconds in a mini-batch (dynamic batch size). :param max_cuts: Maximum number of examples in a mini-batch (static batch size). :param num_buckets: If set, enables bucketing (each mini-batch has examples of a similar duration). :param duration_bins: A list of floats (seconds); when provided, we'll skip the initial estimation of bucket duration bins (useful to speed-up the launching of experiments). :param quadratic_duration: If set, adds a penalty term for longer duration cuts. Works well with models that have quadratic time complexity to keep GPU utilization similar when using bucketing. Suggested values are between 30 and 45. N cuts_paths index_path base_seed max_durationmax_cuts num_buckets duration_binsquadratic_durationreturnc stjddg|_g|_t|ttfr#|jt||jdnpt|}g|_t|dttfrR|D]} t| ttfsBJd|jt| |jdq5nA|D]>} t | dksdJd| d| \} } t| ttfsvJd| t| t t fsJd | |jt| |j| qT||_ ||_ ||_||_||_||_||_td d |j |jfDsJd t|_t|j|j |_d dt|jj|jD|_g|_t|_t|_ dS)N) data_sourceg?rz4Mixing paths with and without scales is not allowed.z Expected (path, scale) but got: z< [note: mixing paths with and without scales is not allowed]z$Path must be a string or Path, got: z$Scale must be an int or float, got: css|]}|duVqdSN).0vr#r#U/home/ubuntu/.local/lib/python3.10/site-packages/lhotse/dataset/sampling/stateless.py s z,StatelessSampler.__init__..z7At least one of max_duration or max_cuts has to be set.cSsg|]\}}||qSr#r#)r$lcscaler#r#r& sz-StatelessSampler.__init__..)!super__init__pathsscales isinstancerstrappendlistlenintfloatrrrrrrranyr diagnostics ManifestIndexindexzip line_countsvaluesscaled_line_counts _transformsrrankr world_size) selfrrrrrrrrptplr) __class__r#r&r,msn     zStatelessSampler.__init__fncCs|j||S)zAApply ``fn`` to each mini-batch of ``CutSet`` before yielding it.)r>r1)rArFr#r#r&maps zStatelessSampler.mapcCsiS)zHStub state_dict method that returns nothing - this sampler is stateless.r#rAr#r#r& state_dictzStatelessSampler.state_dictrIcCsdS)zJStub load_state_dict method that does nothing - this sampler is stateless.Nr#)rArIr#r#r&load_state_dictrJz StatelessSampler.load_state_dictc #sddlm}m}tjj}|durdn|j}|dj}j |}t |t dtjd|dj d|djd |d fd d }jdusWjdurl||jjjjd d jddd }n||jjd d ddd}|tj|j_|EdHdS)Nr)DynamicBucketingSamplerDynamicCutSampleri[z$] Initialized sampler RNG with seed z (== base_seed=z + my_id=z ) [ddp_rank=z worker_id=]c 3sd} jjd}jj|}t|d}||||d}}|}||| ||}Wdn1sCwYt |}t |} | j d|| _ | V|d7}q)a  Infinite generator of cuts. Each cut is samples in two steps: - first we select a cutset file, weighted by line count (num cuts) - then we randomly select a line from that file using uniform distribution rTN_it) choicesr-r=r9 line_offsets randranger3openseekreadrrid) npathrS begin_idxbeginendflinedatacutrngrAr#r&_inners"   z)StatelessSampler.__iter__.._innerFrP) rrrrshuffle drop_lastrr@r?)rrrerfr@r?)lhotse.datasetrLrMtorchutilsr`get_worker_inforXr?rrandomRandomlogginginfotype__name__rrrrrrGr r>r7) rArLrM worker_info worker_idmy_idseedrd inner_samplerr#rbr&__iter__sP   "  zStatelessSampler.__iter__cCs |jS)zJReturns a string describing the statistics of the sampling process so far.)r7 get_reportrHr#r#r&rws zStatelessSampler.get_report)NNNNN)rp __module__ __qualname____doc__r rrPathlikeAndScaler4rrrr,rr rGrrIrKrrvr0rw __classcell__r#r#rDr&rs>S  GCrc @sbeZdZdZ ddeedededdfdd Zd ede e fd d Z d ed ede e fddZ dS)r8a; An index of line count and line offset for each cutset manifest. When created for the first time, it writes a .jsonl.idx file for each .jsonl file that contains byte offsets for each line. It also writes a file at ``index_path`` that has the line count and path for each manifest. When this object is instantiated again (e.g. when resuming training), it will just load the contents of existing files from disk. Objects of this class expose two members: ``line_counts: Dict[Path, int]`` and ``line_offsets: Dict[Path, List[int]]`` to simplify working with manifests. :param manifest_paths: A list of paths to cut sets. :param index_path: A path where we should write the line count index (if it doesn't exist). :param force: When true, we'll ignore existing files and reindex the cutsets. Fmanifest_pathsrforcerNc Csi|_i|_tt|D]1}|jdksJd||d}|r*|s*||}n|||}t ||j|<||j|<q |rC|rp| d}|j D]\}}t |d||dqNWddS1siwYdSdS)Nz.jsonlzIWe only support uncompressed .jsonl files in this sampler, but received: z .jsonl.idxw file) r;rSrGrsuffix with_suffixis_file_load_processr3rUitemsprint) rAr}rr~rB offset_pathoffsetsindex_fr(r#r#r&r,s&        "zManifestIndex.__init__ file_indexcCs<|}ttt|}Wd|S1swY|Sr")rUtuplerGr4)rArr^rr#r#r&r7s  zManifestIndex._loadmanifestc Csdg}|K}|d)}td|d|}|r/||t|d|d|}|sWdn1s9wYWdt|SWdt|S1sUwYt|S)Nrrr)rUrreadliner1tellr)rArrrcuts_frr_r#r#r&r<s  ( zManifestIndex._process)F) rprxryrzr rboolr,rr r4rrr#r#r#r&r8 s  r8)(rmrkpathlibrtypingrrrrrrr r r rhcytoolzr lhotser rlhotse.cut.setrlhotse.dataset.dataloadingrrlhotse.dataset.sampling.baser lhotse.lazyrlhotse.serializationr lhotse.utilsrr5r{rir`Samplerrr8r#r#r#r&s" ,       l