o eii5@sdZddlZddlZddlZddlmZddlmmZ ddl m Z ddl m Z ddlmZddlmZddlmZddlmZdd lmZGd d d ejZGd d d ejZGdddejZddZddZddZdS)zComponents necessary to build a wav2vec 2.0 architecture following the original paper: https://arxiv.org/abs/2006.11477. Authors * Rudolf A Braun 2022 * Guillermo Cambara 2022 * Titouan Parcollet 2022 N)length_to_mask)ConvolutionFrontEnd)PositionalEncoding)Conv1d) LayerNorm)GumbelVectorQuantizer)batch_pad_rightcsReZdZdZgdgdgdddffdd Zdd d Zd ejfd dZZ S)W2VLatentExtractoraConvolution based feature extractor from raw audio. Channel numbers increasing is based on https://arxiv.org/abs/2109.06870 Arguments --------- out_channels : list of ints Out channels of convolutional layers. kernel_sizes : list of ints Kernels of convolutional layers. strides : list of ints Strides of convolutional layers. dropout : float Dropout of CNN. conv_init : str Type of initialization to use, default "kaiming" Example ------- >>> extractor = W2VLatentExtractor() >>> inputs = torch.rand(10, 5000) >>> outputs = extractor(inputs) >>> outputs.shape torch.Size([10, 14, 512]) )r r r r r r ) r r r r r )rrrrrgkaimingcstt|t|krt|ksJJt|}||_||_|d|_td|d|||dg|dg|ttj t |dd|d|_ t |d|_ dS)N)Ni>rFvalid) num_blocksnum_layers_per_block out_channels kernel_sizesstrides dilations residuals conv_module activationnormdropout conv_biaspadding conv_init) super__init__lenrrout_dimrrnnGELUr extractorr)selfrrrrr r __class__^/home/ubuntu/transcripts/venv/lib/python3.10/site-packages/speechbrain/lobes/models/wav2vec.pyr"4s0 ( zW2VLatentExtractor.__init__TcCs8|r t||jdd}|d}||}||S)z$Calculates latents from audio input.rNr)F layer_normshape unsqueezer'r)r(xnormalize_signallatentsr+r+r,forwardYs    zW2VLatentExtractor.forward input_lengthscCs8dd}t|j|jD] \}}||||}q |tjS)z2Calculates output lengths for given input lengths.cSst|||dS)Nr)torchfloor) input_length kernel_sizestrider+r+r,_conv_out_lengthdsz?W2VLatentExtractor.get_output_lengths.._conv_out_length)ziprrtor6long)r(r5r;r9r:r+r+r,get_output_lengthsas z%W2VLatentExtractor.get_output_lengths)T) __name__ __module__ __qualname____doc__r"r4r6 LongTensorr? __classcell__r+r+r)r,r s %r cs4eZdZdZddeddffdd Zdd ZZS) W2VTargetQuantiseraWraps ``nnet.quantiser.GumbelVectorQuantizer``, see for documentation on arguments. Arguments --------- in_dim : int Input dimension (channels). out_dim : int Output dimension quantiser : class Default GumbelVectorQuantizer num_vars : int Number of quantized vectors per group. temperature_decay : tuple Temperature for training. this should be a tuple of 3 elements: (start, stop, decay factor). Example ------- >>> quantiser = W2VTargetQuantiser() >>> inputs = torch.rand(10, 12, 512) >>> output, meta = quantiser(inputs) >>> output.shape torch.Size([10, 12, 256]) r i@)g@g?g;?cs.t||||d||_t|||_dS)Nr)r!r" quantiserr%Linearproj)r(in_dimr$rHnum_varstemperature_decayr)r+r,r"s  zW2VTargetQuantiser.__init__c Cs\||}||d}|d}|d}|d}|d}|||}|||||d}||fS)z0Returns quantised targets plus meta information.r1code_perplexity prob_perplexrLtemp)diversity_loss code_perplexrOrLrP)rHrJ) r(r1targetsrRrOrLrPrQmetar+r+r,r4s  zW2VTargetQuantiser.forward)r@rArBrCrr"r4rEr+r+r)r,rFlsrFcs0eZdZdZedffdd ZdddZZS) EncoderWrapperaA wrapper that adds positional information, masks the input and then runs the latent encoder. Arguments --------- in_dim : int Last dimension of input tensor. embedding_dim : int Dimension to project input to and that the latent encoder will use. latent_encoder : torch.nn.module Initialized latent encoder object. positional_encoding : torch.nn.module Uninitialized nn.module for adding positional information, will use ``embedding_dim``. dropout_encoder_input : float Dropout on encoder input. Example ------- >>> from speechbrain.lobes.models.transformer.Transformer import TransformerEncoder >>> encoder = TransformerEncoder(d_model=768, num_layers=4, nhead=4, d_ffn=1024) >>> wrapper = EncoderWrapper(1024, 768, encoder) >>> inputs = torch.rand(10, 12, 1024) >>> outputs = wrapper(inputs) >>> outputs["embeddings"].shape torch.Size([10, 12, 768]) g?csRtt|||_||_|||_t||_tj t | dd|_ dS)NT) requires_grad)r!r"r%rIinput_projectorlatent_encoderpositional_encodingDropoutdropout_encoder_input Parameterr6 FloatTensoruniform_mask_emb)r(rK embedding_dimrXrYr[r)r+r,r"s    zEncoderWrapper.__init__Nc Csi}|d}||}||}|dur.|j|j||<|}||d<|||d<|dur@t ||}t |t d}|| |}|j ||d\}} ||d<|S)a  Arguments --------- latents : torch.Tensor, shape (B, T, C) Batch of latent representations (AKA frames) output from latent extractor. wav_lens : torch.Tensor, shape (B,) The actual (unpadded) relative lengths for each sample of the batch (0This creates the boolean mask for a target shape which respects the sample lengths and will half roughly ``mask_prob`` entries set to ``True``. Arguments --------- shape : list of ints, like (N, M) Shape of boolean mask to return. sample_lens: list of ints Absolute lengths of per sample lengths. mask_prob : float Percentage to mask. mask_length: int Length of contiguous subsequence to mask. Returns ------- mask : numpy.ndarray Boolean mask with shape of input argument ``shape``. rF)replacecs&g|]}tD]}||qqSr+)range).0joffset mask_indices mask_lengthr+r, )s z compute_mask..T)minintfloatrandomrtnpchoiceasarrayr#appenduniquefull enumeratearangedelete)r/ sample_lens mask_probrzbspadded_sample_lenmin_sample_lennum_mask mask_idcsi sample_lenrnnum_mask_totalmask_idcnum_mask_missingr extra_indcsr+rxr, compute_masks<         rc Cs|j\}}}|d}t,t|dd|}tjd||||fd}|||kd7<Wdn1s=wY|t|d|}|d|}||d}||||| dddd}|S)aSamples negatives from target tensor y. Arguments --------- y : torch.Tensor Tensor of shape (B, T, C) num_neg : int Number of negatives to sample. Returns ------- negs : torch.Tensor Negatives in shape (N, B, T, C) rrr)lowhighrgNrr ) r/r6no_gradrr0expandflattenrandintviewpermute) ynum_negBrpCrrS neg_indcsnegsr+r+r,sample_negativesDs   rcCsgg}}g}|D]#}||d|d}|||t|d} || q t|} t|\} } t|} t| | f|||}t| t| tj|tj dfS)aThis creates a batch from a list of samples and also creates the boolean mask that will be used to mask the inputs of the latent encoder. To create the mask we need to know the output shape after the latent extractor, therefore the argument `get_out_len_fn`. One could also create masks per sample (when loading the audio file) and then collate them but at that time one doesn't know the length of the shortest sample in the batch (which determines the number of masked frames) so it's better this way. Arguments --------- samples_lst : list List of samples returned by the audio_pipeline. get_out_len_fn : function Function that calculates length of sample after it passes through feature extractor. mask_prob : float Approximate percentage of frames to mask. mask_length : int Number of contiguous frames that will be masked. Returns ------- wavs_padded : torch.Tensor, shape (B, T) Audio arrays with right-sided padding. wav_lens : torch.Tensor, shape (B,) For each sample the percentage of the array that is not padding. mask : torch.Tensor, shape (B, T) Boolean mask to mask frames. idsigrrc) rr6 as_tensorrgitemr#rmaxrrk) samples_lstget_out_len_fnrrzwav_lstlatent_length_lstidssampler latent_lengthr wavs_paddedrlbatch_time_lenrnr+r+r,w2v_mask_collate_fnbs.    r)rCrnumpyrr6torch.nnr%torch.nn.functional functionalr-speechbrain.dataio.dataior$speechbrain.lobes.models.convolutionr0speechbrain.lobes.models.transformer.Transformerrspeechbrain.nnet.CNNrspeechbrain.nnet.normalizationrspeechbrain.nnet.quantisersrspeechbrain.utils.data_utilsrModuler rFrUrrrr+r+r+r,s&         R;ZC