o SiR @s6ddlZddlmZmZddlmZddlmZmZm Z m Z m Z m Z m Z mZddlZddlmmZddlmZmZddlmZddlmZmZmZmZmZddlm Z m!Z!m"Z"m#Z#m$Z$e d ed Z%Gd d d Z&Gd dde&Z'Gddde&Z(Gddde&Z)edddefde*de e%de efddZ+dS)N)ExecutorThreadPoolExecutor) lru_cache)CallableDictListOptionalTupleTypeTypeVarUnion)CutSetFeatureExtractor)compute_supervisions_frame_mask) collate_audiocollate_featurescollate_matricescollate_vectorsread_audio_from_cuts) LOG_EPSILONcompute_num_framesifnonesupervision_to_framessupervision_to_samples ExecutorType)boundc@s~eZdZdZdefdedeeddfddZd e de e j e j ffd d Zd e deee j ffd d Zd e de j fddZdS)BatchIOa Converts a :class:`CutSet` into a collated batch of audio representations. These representations can be e.g. audio samples or features. They might also be single or multi channel. All InputStrategies support the ``executor`` parameter in the constructor. It allows to pass a ``ThreadPoolExecutor`` or a ``ProcessPoolExecutor`` to parallelize reading audio/features from wherever they are stored. Note that this approach is incompatible with specifying the ``num_workers`` to ``torch.utils.data.DataLoader``, but in some instances may be faster. .. note:: This is a base class that only defines the interface. .. automethod:: __call__ r num_workers executor_typereturnNcCs||_||_dSN)r_executor_type)selfrrr#S/home/ubuntu/.local/lib/python3.10/site-packages/lhotse/dataset/input_strategies.py__init__.s zBatchIO.__init__cutscCt)zcReturns a tensor with collated input signals, and a tensor of length of each signal before padding.NotImplementedErrorr"r&r#r#r$__call__6szBatchIO.__call__cCr')ap Returns a dict that specifies the start and end bounds for each supervision, as a 1-D int tensor. Depending on the strategy, the dict should look like: .. code-block:: { "sequence_idx": tensor(shape=(S,)), "start_frame": tensor(shape=(S,)), "num_frames": tensor(shape=(S,)), } or .. code-block:: { "sequence_idx": tensor(shape=(S,)), "start_sample": tensor(shape=(S,)), "num_samples": tensor(shape=(S,)) } Where ``S`` is the total number of supervisions encountered in the :class:`CutSet`. Note that ``S`` might be different than the number of cuts (``B``). ``sequence_idx`` means the index of the corresponding feature matrix (or cut) in a batch. r(r*r#r#r$supervision_intervals:szBatchIO.supervision_intervalscCr')a Returns a collated batch of masks, marking the supervised regions in cuts. They are zero-padded to the longest cut. Depending on the strategy implementation, it is expected to be a tensor of shape ``(B, NF)`` or ``(B, NS)``, where ``B`` denotes the number of cuts, ``NF`` the number of frames and ``NS`` the total number of samples. ``NF`` and ``NS`` are determined by the longest cut in a batch. r(r*r#r#r$supervision_masksYs zBatchIO.supervision_masks)__name__ __module__ __qualname____doc__rintr rr%r r torchTensor IntTensorr+rstrr,r-r#r#r#r$rs rc @seZdZdZ ddedeedeej ej ffddZ ddedeede eej ffdd Z dded eedeedej fd d Z d S)PrecomputedFeaturesa :class:`InputStrategy` that reads pre-computed features, whose manifests are attached to cuts, from disk. It automatically pads the feature matrices so that every example has the same number of frames as the longest cut in a mini-batch. This is needed to put all examples into a single tensor. The padding value is a low log-energy, around log(1e-10). .. automethod:: __call__ rightr& pad_directionrcCst||t|j|jddS)a Reads the pre-computed features from disk/other storage. The returned shape is ``(B, T, F) => (batch_size, num_frames, num_features)``. :return: a tensor with collated features, and a tensor of ``num_frames`` of each cut before padding.r)r9executor)r _get_executorrr!)r"r&r9r#r#r$r+ss  zPrecomputedFeatures.__call__cs|dvr td|tdd|Dtdd|D\}}|dkr6fdd|D}d dt||D}d dt|D}tj|tjd tj|tjd tj|tjd d S) h Returns a dict that specifies the start and end bounds for each supervision, as a 1-D int tensor, in terms of frames: .. code-block:: { "sequence_idx": tensor(shape=(S,)), "start_frame": tensor(shape=(S,)), "num_frames": tensor(shape=(S,)) } Where ``S`` is the total number of supervisions encountered in the :class:`CutSet`. Note that ``S`` might be different than the number of cuts (``B``). ``sequence_idx`` means the index of the corresponding feature matrix (or cut) in a batch. leftr8-pad_direction must be 'left' or 'right', got css|]}|jVqdSr ) num_frames.0cutr#r#r$ sz.css2|]}|jD]}t||j|j|jdVqqdS) max_framesN) supervisionsr frame_shift sampling_raterArCrDsupr#r#r$rEs r?cs"g|] }|jD]}|jqqSr#)rHrA)rCrD_rFr#r$ s  z=PrecomputedFeatures.supervision_intervals..cSsg|]\}}||qSr#r#)rCsor#r#r$rNscS g|] \}}|jD]}|q qSr#rH)rCicrMr#r#r$rN dtype sequence_idx start_framerA) ValueErrormaxzip enumerater3tensorint32)r"r&r9 start_frames nums_framesoffsetsrYr#rFr$r,s( z)PrecomputedFeatures.supervision_intervalsNuse_alignment_if_existscs4|dvr td|fdd|D}t||dS)a0Returns the mask for supervised frames. :param use_alignment_if_exists: optional str, key for alignment type to use for generating the mask. If not exists, fall back on supervision time spans. :param pad_direction: where to apply the padding (``right`` or ``left``). r>r@cg|]}|jdqSrd)supervisions_feature_maskrBrgr#r$rN z9PrecomputedFeatures.supervision_masks..)r9)r[r)r"r&rdr9masksr#rgr$r-s   z%PrecomputedFeatures.supervision_masks)r8)Nr8)r.r/r0r1r rr6r r3r4r+rr,r-r#r#r#r$r7fs:   7r7c seZdZdZddedfdededeededd f fd d Z dd e d e e de eejejfeejeje fffddZd e dee ejffddZ dd e de e dejfddZZS) AudioSamplesa :class:`InputStrategy` that reads single-channel recordings, whose manifests are attached to cuts, from disk (or other audio source). It automatically zero-pads the recordings so that every example has the same number of audio samples as the longest cut in a mini-batch. This is needed to put all examples into a single tensor. .. automethod:: __call__ rFrfault_tolerantruse_batch_loaderrNcsDtj||d||_d|_||_|jr ddlm}||_dSdS)a. AudioSamples constructor. :param num_workers: when larger than 0, we will spawn an executor (of type specified by ``executor_type``) to read the audio data in parallel. Thread executor can be used with PyTorch's DataLoader, whereas Process executor would fail (but could be faster for other applications). :param fault_tolerant: when ``True``, the cuts for which audio loading failed will be skipped. It will make ``__call__`` return an additional item, which is the CutSet for which we successfully read the audio. It may be a subset of the input CutSet. :param executor_type: the type of executor used for parallel audio reads (only relevant when ``num_workers>0``). :param use_batch_loader: When ``True``, enables batch loading of audio data from AIStore. This allows all audio samples in the batch to be fetched in a single request for increased efficiency. Requires the input CutSet to be eager (not lazy). rrNr)AISBatchLoader)superr%rlais_batch_loaderrm lhotse.aisro)r"rrlrrmro __class__r#r$r%s  zAudioSamples.__init__r&recording_fieldcCs8|jr |jdur ||}t|t|j|jd|j|dS)a Reads the audio samples from recordings on disk/other storage. The returned shape is ``(B, T) => (batch_size, num_samples)``. :param recording_field: when specified, we will try to load recordings from a custom field with this name (i.e., ``cut.load_()`` instead of default ``cut.load_audio()``). :return: a tensor with collated audio samples, and a tensor of ``num_samples`` of each cut before padding. .. note:: When AIStore batch loading is enabled (`use_batch_loader=True`), the audio data will be fetched from AIStore using a single batch request before collation. The input CutSet must be eager (not lazy). Nr:)r;rlru)rmrqrr<rr!rl)r"r&rur#r#r$r+s zAudioSamples.__call__cCsXtdd|D\}}ddt|D}tj|tjdtj|tjdtj|tjddS)al Returns a dict that specifies the start and end bounds for each supervision, as a 1-D int tensor, in terms of samples: .. code-block:: { "sequence_idx": tensor(shape=(S,)), "start_sample": tensor(shape=(S,)), "num_samples": tensor(shape=(S,)) } Where ``S`` is the total number of supervisions encountered in the :class:`CutSet`. Note that ``S`` might be different than the number of cuts (``B``). ``sequence_idx`` means the index of the corresponding feature matrix (or cut) in a batch. css(|]}|jD] }t||jVqqdSr )rHrrJrKr#r#r$rE-s z5AudioSamples.supervision_intervals..cSrQr#rRrCrSrTrOr#r#r$rN3rUz6AudioSamples.supervision_intervals..rV)rY start_sample num_samplesr]r^r3r_r`)r"r& start_samples nums_samplesrYr#r#r$r,sz"AudioSamples.supervision_intervalsrdcstfdd|DS)Returns the mask for supervised samples. :param use_alignment_if_exists: optional str, key for alignment type to use for generating the mask. If not exists, fall back on supervision time spans. crerf)supervisions_audio_maskrBrgr#r$rNCriz2AudioSamples.supervision_masks..rr"r&rdr#rgr$r-:s  zAudioSamples.supervision_masksr )r.r/r0r1rr2boolr rr%r rr6r r r3r4r+rr,r- __classcell__r#r#rsr$rksH "  !rkcseZdZdZdddddefdedeeej gej fde d e d e d e d e e d dffdd Z ddedeed eeej ej feej ej efffddZded eeej ffddZ ddedeed ej fddZZS)OnTheFlyFeaturesa :class:`InputStrategy` that reads single-channel recordings, whose manifests are attached to cuts, from disk (or other audio source). Then, it uses a :class:`FeatureExtractor` to compute their features on-the-fly. It automatically pads the feature matrices so that every example has the same number of frames as the longest cut in a mini-batch. This is needed to put all examples into a single tensor. The padding value is a low log-energy, around log(1e-10). .. note: The batch feature extraction performed here is not as efficient as it could be, but it allows to use arbitrary feature extraction method that may work on a single recording at a time. .. automethod:: __call__ NrTF extractorwave_transformsruse_batch_extractrl return_audiorrcs8tj||d||_t|g|_||_||_||_dS)a OnTheFlyFeatures' constructor. :param extractor: the feature extractor used on-the-fly (individually on each waveform). :param wave_transforms: an optional list of transforms applied on the batch of audio waveforms collated into a single tensor, right before the feature extraction. :param num_workers: when larger than 0, we will spawn an executor (of type specified by ``executor_type``) to read the audio data in parallel. Thread executor can be used with PyTorch's DataLoader, whereas Process executor would fail (but could be faster for other applications). :param use_batch_extract: when ``True``, we will call :meth:`~lhotse.features.base.FeatureExtractor.extract_batch` to compute the features as it is possibly faster. It has a restriction that all cuts must have the same sampling rate. If that is not the case, set this to ``False``. :param fault_tolerant: when ``True``, the cuts for which audio loading failed will be skipped. It will make ``__call__`` return an additional item, which is the CutSet for which we successfully read the audio. It may be a subset of the input CutSet. :param return_audio: When ``True``, calling this object will additionally return collated audio tensor and audio lengths tensor. :param executor_type: the type of executor used for parallel audio reads (only relevant when ``num_workers>0``). rnN)rpr%rrrrrlr)r"rrrrrlrrrsr#r$r%_s !  zOnTheFlyFeatures.__init__r&ruc s|tt|j|jd|j|d\}|jD]}tt|D] }|||||<qq|jrDt fddDs8J|j j |dj d}n6g}t D]/\}}||}z |j ||j } Wntd|jd||t| qJt|td } tjd d |Dtjd } | | f} |jrd d |D}tjdd |Dtjd } t|dd }| || f} |jr| f} | S)a Reads the audio samples from recordings on disk/other storage and computes their features. The returned shape is ``(B, T, F) => (batch_size, num_frames, num_features)``. :param recording_field: when specified, we will try to load recordings from a custom field with this name (i.e., ``cut.load_()`` instead of default ``cut.load_audio()``). :return: a tuple of objcets: ``(feats, feat_lens, [audios, audio_lens], [cuts])``. Tensors ``audios`` and ``audio_lens`` are returned when ``return_audio=True``. CutSet ``cuts`` is returned when ``fault_tolerant=True``. r:)r;suppress_errorsruc3s |] }|jdjkVqdS)rNrJ)rCrTr&r#r$rEsz,OnTheFlyFeatures.__call__..rrz4Error while extracting the features for cut with ID z -- details: ) padding_valuecSg|]}|jdqSrshape)rCfr#r#r$rNz-OnTheFlyFeatures.__call__..rVcSsg|]}|dqSr)squeezerCar#r#r$rNrcSrrrrr#r#r$rNr)rr<rr!rlrrangelenrallr extract_batchrJr^numpyextractloggingerroridappendr3 from_numpyrrr_int64rr)r"r&ruaudiostfnmidxfeatures_singlerDsamplesfeaturesfeatures_batch feature_lensout audio_lensr#rr$r+sN       zOnTheFlyFeatures.__call__cs\tfdd|D\}}ddt|D}tj|tjdtj|tjdtj|tjddS)r=c3s.|]}|jD] }t|jj|jVqqdSr )rHrrrIrJrKr"r#r$rEs  z9OnTheFlyFeatures.supervision_intervals..cSrQr#rRrvr#r#r$rNrUz:OnTheFlyFeatures.supervision_intervals..rVrXry)r"r&rarbrYr#rr$r,s  z&OnTheFlyFeatures.supervision_intervalsrdcstfdd|DS)r|csg|] }t|jjdqS))rIrd)rrrIrBr"rdr#r$rNsz6OnTheFlyFeatures.supervision_masks..r~rr#rr$r-s  z"OnTheFlyFeatures.supervision_masksr )r.r/r0r1rrrrr3r4r2rr rr%r rr6r r r+rr,r-rr#r#rsr$rLsX )  D"r)maxsize max_workersrrcCs|dkrdS||dS)aA This function caches a thread/process pool in the global state of a given process. It's useful for keeping a process pool alive across different invocations within the same process for efficiency. We intend it to be used for efficient data reads withing a task executed in a parent process pool. rN)rr#)rrr#r#r$r<s  r<),rconcurrent.futuresrr functoolsrtypingrrrrr r r r r3torch.nn.functionalnn functionalFlhotser r lhotse.cutrlhotse.dataset.collationrrrrr lhotse.utilsrrrrrrrr7rkrr2r<r#r#r#r$s2 (  Ij|5