o %ݫiX@sdZddlZddlZddlmZddlmmZddl m Z ddl m Z m Z mZeejZGdddejZGdddejZGd d d ejZGd d d ejZGd ddejZGdddejZdS)zKLibrary for the Resource-Efficient Sepformer. Authors * Cem Subakan 2022 N) select_norm)PositionalEncodingTransformerEncoderget_lookahead_maskcs2eZdZdZ    d fdd Zdd ZZS) MemLSTMathe Mem-LSTM of SkiM -- Note: This is taken from the SkiM implementation in ESPNet toolkit and modified for compatibility with SpeechBrain. Arguments --------- hidden_size: int Dimension of the hidden state. dropout: float dropout ratio. Default is 0. bidirectional: bool Whether the LSTM layers are bidirectional. Default is False. mem_type: str 'hc', 'h', 'c', or 'id' This controls whether the hidden (or cell) state of SegLSTM will be processed by MemLSTM. In 'id' mode, both the hidden and cell states will be identically returned. norm_type: str 'gln', 'cln' This selects the type of normalization cln is for causal implementation Example ------- >>> x = (torch.randn(1, 5, 64), torch.randn(1, 5, 64)) >>> block = MemLSTM(64) >>> x = block(x, 5) >>> x[0].shape torch.Size([1, 5, 64]) Fhcclnc st||_||_t|d||_||_|dvs"Jd||dvr?t|j|jd|jd||d|_t ||jdt d|_ |d vr^t|j|jd|jd||d|_ t ||jdt d|_ dSdS) N)rhcidz4only support 'hc', 'h', 'c' and 'id', current type: )rr LSTM) input_sizehidden_channels num_layersoutsizernn_typedropout bidirectionalnormdimshapeeps)rr )super__init__ hidden_sizerintrmem_type SBRNNBlockh_netrEPSh_normc_netc_norm)selfrrrr norm_type __class__X/home/ubuntu/.local/lib/python3.10/site-packages/speechbrain/lobes/models/resepformer.pyr9sF     zMemLSTM.__init__c Cs|jdkr|}n|\}}|j\}}}||} |dd| |||}|dd| |||}|jdkrb||||dddddd}||| |dddddd}n=|jdkr||||dddddd}t |}n|jdkrt |}||| |dddddd}|| |||dd}|| |||dd}||f}|j sg} |D]$} t | } | dddd ddf| ddddddf<| | qt| }|S) a-The forward function for the memory RNN Arguments --------- hc : tuple (h, c), tuple of hidden and cell states from SegLSTM shape of h and c: (d, B*S, H) where d is the number of directions B is the batchsize S is the number chunks H is the latent dimensionality S : int S is the number of chunks Returns ------- ret_val : torch.Tensor The output of memory RNN r r rrr r N)r r transpose contiguousviewr$r"permuter&r%torch zeros_likerappendtuple) r'rSret_valr r dBSHBcausal_ret_valxx_r+r+r,forwardksF        0 zMemLSTM.forward)rFrr __name__ __module__ __qualname____doc__rr@ __classcell__r+r+r)r,rs$2rc0eZdZdZ   d fdd ZddZZS) SegLSTMaMthe Segment-LSTM of SkiM Note: This is taken from the SkiM implementation in ESPNet toolkit and modified for compatibility with SpeechBrain. Arguments --------- input_size: int, dimension of the input feature. The input should have shape (batch, seq_len, input_size). hidden_size: int, dimension of the hidden state. dropout: float, dropout ratio. Default is 0. bidirectional: bool, whether the LSTM layers are bidirectional. Default is False. norm_type: str One of gln, cln. This selects the type of normalization cln is for causal implementation. Example ------- >>> x = torch.randn(3, 20, 64) >>> hc = None >>> seglstm = SegLSTM(64, 64) >>> y = seglstm(x, hc) >>> y[0].shape torch.Size([3, 20, 64]) rFcLNcsrt||_||_t|d|_tj||dd|d|_tj |d|_ t ||j||_ t ||dtd|_dS)Nr T) batch_firstr)prr)rrrrr num_directionnnrlstmDropoutrLinearprojrr#r)r'rrrrr(r)r+r,rs  zSegLSTM.__init__c Cs|j\}}}|dur&|j}t|||j|j}t|||j|j}n|\}}||||f\} \}}|| } | | d| jd |j} | | ddd ddd} || } | ||ffS)aThe forward function of the Segment LSTM Arguments --------- input : torch.Tensor shape [B*S, T, H] where B is the batchsize S is the number of chunks T is the chunks size H is the latent dimensionality hc : tuple tuple of hidden and cell states from SegLSTM shape of h and c: (d, B*S, H) where d is the number of directions B is the batchsize S is the number chunks H is the latent dimensionality Returns ------- output: torch.Tensor Output of Segment LSTM (h, c): tuple Same as hc input Nr.r-rr )rrLr3zerosrtodevicerNrrQr0r1rr2) r'inputrr<Tr;r9r r output output_normr+r+r,r@s   zSegLSTM.forward)rFrIrAr+r+r)r,rHs#rHcrG) r!aRNNBlock with output layer. Arguments --------- input_size : int Dimensionality of the input features. hidden_channels : int Dimensionality of the latent layer of the rnn. num_layers : int Number of the rnn layers. outsize : int Number of dimensions at the output of the linear layer rnn_type : str Type of the the rnn cell. dropout : float Dropout rate bidirectional : bool If True, bidirectional. Example ------- >>> x = torch.randn(10, 100, 64) >>> rnn = SBRNNBlock(64, 100, 1, 128, bidirectional=True) >>> x = rnn(x) >>> x.shape torch.Size([10, 100, 128]) rrTc sFttt||||||d|_|rd|n|}t|||_dS)N)rrrrr-)rrgetattrSBRNNmdlrMrPout) r'rrrrrrr rnn_outsizer)r+r,r0s zSBRNNBlock.__init__cCs||d}||}|S)aJReturns the transformed output. Arguments --------- x : torch.Tensor [B, L, N] where, B = Batchsize, N = number of filters L = time points Returns ------- out : torch.Tensor The transformed output. r)r[r\)r'r>rnn_outr\r+r+r,r@Fs zSBRNNBlock.forward)rrTrAr+r+r)r,r!s"r!csDeZdZdZ             dfd d Zd d ZZS)SBTransformerBlock_wnormandskipaA wrapper for the SpeechBrain implementation of the transformer encoder. Arguments --------- num_layers : int Number of layers. d_model : int Dimensionality of the representation. nhead : int Number of attention heads. d_ffn : int Dimensionality of positional feed forward. input_shape : tuple Shape of input. kdim : int Dimension of the key (Optional). vdim : int Dimension of the value (Optional). dropout : float Dropout rate. activation : str Activation function. use_positional_encoding : bool If true we use a positional encoding. norm_before : bool Use normalization before transformations. attention_type : str Type of attention, default "regularMHA" causal : bool Whether to mask future information, default False use_norm : bool Whether to include norm in the block. use_skip : bool Whether to add skip connections in the block. norm_type : str One of "cln", "gln" Example ------- >>> x = torch.randn(10, 100, 64) >>> block = SBTransformerBlock_wnormandskip(1, 64, 8) >>> x = block(x) >>> x.shape torch.Size([10, 100, 64]) N皙?reluF regularMHATglncst| |_| dkrtj} n | dkrtj} ntd| |_t||||||||| | | | d |_ ||_ ||_ |rAt ||dt d|_| rLt|dd|_dSdS) Nrbgeluzunknown activation) rnheadd_ffn input_shaped_modelkdimvdimr activationnormalize_beforecausalattention_typerri)rmax_len)rruse_positional_encodingrMReLUGELU ValueErrorrnrr[use_normuse_skiprr#rrpos_enc)r'rrirfrgrhrjrkrrlrq norm_beforerornrurvr(r)r+r,rsB  z(SBTransformerBlock_wnormandskip.__init__cCs|jrt|nd}|jr||}|j|||dd}n |j||dd}|jr8||dddddd}|jr?||}|S)aXReturns the transformed output. Arguments --------- x : torch.Tensor Tensor shape [B, L, N], where, B = Batchsize, L = time points N = number of filters Returns ------- out : torch.Tensor The transformed output. N)src_maskrr-r ) rnrrqrwr[rurr2rv)r'r>ryrwr\r+r+r,r@s z'SBTransformerBlock_wnormandskip.forward) r`NNNrarbFFrcFTTrdrAr+r+r)r,r_[s"3;r_csBeZdZdZ        dfd d Zd d Zd dZZS)#ResourceEfficientSeparationPipelineaqResource Efficient Separation Pipeline Used for RE-SepFormer and SkiM Note: This implementation is a generalization of the ESPNET implementation of SkiM Arguments --------- input_size: int Dimension of the input feature. Input shape should be (batch, length, input_size) hidden_size: int Dimension of the hidden state. output_size: int Dimension of the output size. dropout: float Dropout ratio. Default is 0. num_blocks: int Number of basic SkiM blocks segment_size: int Segmentation size for splitting long features bidirectional: bool Whether the RNN layers are bidirectional. mem_type: str 'hc', 'h', 'c', 'id' or None. This controls whether the hidden (or cell) state of SegLSTM will be processed by MemLSTM. In 'id' mode, both the hidden and cell states will be identically returned. When mem_type is None, the MemLSTM will be removed. norm_type: str One of gln or cln cln is for causal implementation. seg_model: class The model that processes the within segment elements mem_model: class The memory model that ensures continuity between the segments Example ------- >>> x = torch.randn(10, 100, 64) >>> seg_mdl = SBTransformerBlock_wnormandskip(1, 64, 8) >>> mem_mdl = SBTransformerBlock_wnormandskip(1, 64, 8) >>> resepf_pipeline = ResourceEfficientSeparationPipeline(64, 64, 128, seg_model=seg_mdl, mem_model=mem_mdl) >>> out = resepf_pipeline.forward(x) >>> out.shape torch.Size([10, 100, 128]) rr-TavrdNc st||_||_||_||_||_||_||_| |_ |dvs(Jd|t g|_ t |D] } |j t| q2|jdur[t g|_t |dD] } |jt| qOt t t ||d|_dS)Nrr r r r|Nz@only support 'hc', 'h', 'c', 'id', 'av' and None, current type: r )rrr output_sizer segment_sizer num_blocksr r(rM ModuleList seg_modelranger5copydeepcopy mem_model SequentialPReLUConv1d output_fc) r'rrr~rrrrr r(rrir)r+r,rs,      z,ResourceEfficientSeparationPipeline.__init__c Cs|j\}}}|j|d\}}||d|j|}|j\}}}}||jks%J|||||}|jdkrDtj|jdd|jd|jd} nd} t |j D]T} t |j dj } | dkrc|j | || }n| d krr|j | || \}} ntd | |j dkr|jdkr|dd} |j| | ddd } qK|j| | |} qK|||||ddd|ddf}||dd dd }|S) aThe forward function of the ResourceEfficientSeparationPipeline This takes in a tensor of size [B, (S*K), D] Arguments --------- input : torch.Tensor Tensor shape [B, (S*K), D], where, B = Batchsize, S = Number of chunks K = Chunksize D = number of features Returns ------- output : torch.Tensor The separated tensor. )rUr.r|rr )rTNr_rHzUnsupported segment model classr-)r _padfeaturer1rreshaper r3rRrTrrtyperrBrtmean unsqueezerr2rr/) r'rUr<rVDrestr7KrWrrseg_model_typer+r+r,r@Cs6   (z+ResourceEfficientSeparationPipeline.forwardcCsD|j\}}}|j||j}|dkrtjj|ddd|f}||fS)az Arguments --------- input : Tensor of size [B, T, D] where B is Batchsize T is the chunk length D is the feature dimensionality Returns ------- input : torch.Tensor Padded input rest : torch.Tensor Amount of padding r)rrr3rM functionalpad)r'rUr<rVrrr+r+r,r|s z/ResourceEfficientSeparationPipeline._padfeature)rr-r{Tr|rdNN)rBrCrDrErr@rrFr+r+r)r,rzs4.9rzcsleZdZdZ        dd ed ed ededededededeffdd Zde j fddZ Z S)ResourceEfficientSeparatorawResource Efficient Source Separator This is the class that implements RE-SepFormer Arguments --------- input_dim: int Input feature dimension causal: bool Whether the system is causal. num_spk: int Number of target speakers. nonlinear: class the nonlinear function for mask estimation, select from 'relu', 'tanh', 'sigmoid' layer: int number of blocks. Default is 2 for RE-SepFormer. unit: int Dimensionality of the hidden state. segment_size: int Chunk size for splitting long features dropout: float dropout ratio. Default is 0. mem_type: str 'hc', 'h', 'c', 'id', 'av' or None. This controls whether a memory representation will be used to ensure continuity between segments. In 'av' mode, the summary state is is calculated by simply averaging over the time dimension of each segment In 'id' mode, both the hidden and cell states will be identically returned. When mem_type is None, the memory model will be removed. seg_model: class The model that processes the within segment elements mem_model: class The memory model that ensures continuity between the segments Example ------- >>> x = torch.randn(10, 64, 100) >>> seg_mdl = SBTransformerBlock_wnormandskip(1, 64, 8) >>> mem_mdl = SBTransformerBlock_wnormandskip(1, 64, 8) >>> resepformer = ResourceEfficientSeparator(64, num_spk=3, mem_type='av', seg_model=seg_mdl, mem_model=mem_mdl) >>> out = resepformer.forward(x) >>> out.shape torch.Size([3, 10, 64, 100]) Tr-rbrr{rrN input_dimrnnum_spk nonlinearlayerunitrrr c st||_||_| dvrtd| t||||||| |r$dnd|| | | d |_|dvr8td|tj tj tj d||_ dS)Nr}zNot supporting mem_type={}r rd) rrr~rrrr(rr rr)sigmoidrbtanhzNot supporting nonlinear={})rrrrrtformatrzmodelr3rMSigmoidrrTanhr) r'rrnrrrrrrr rrr)r+r,rs4   z#ResourceEfficientSeparator.__init__inptcCs`|ddd}|j\}}}||}|||||j}||jdd}tdd|D}|S)zForward Arguments --------- inpt : torch.Tensor Encoded feature [B, T, N] Returns ------- mask_tensor : torch.Tensor rr-r r)rcSsg|] }|dddqS)rr-r )r2).0mr+r+r, sz6ResourceEfficientSeparator.forward..) r2rrrrrunbindr3stack)r'rr<rVN processedmasks mask_tensorr+r+r,r@s  z"ResourceEfficientSeparator.forward) Tr-rbrrr{rrNN) rBrCrDrErboolstrfloatrr3Tensorr@rFr+r+r)r,rs@0  .r)rErr3torch.nnrMspeechbrain.nnet.RNNnnetRNNrZ"speechbrain.lobes.models.dual_pathr0speechbrain.lobes.models.transformer.Transformerrrrfinfoget_default_dtyperr#ModulerrHr!r_rzrr+r+r+r,s"  jH 0