o i.@sddlZddlmZddlmmZzddlmZWnYddlm Z m Z m Z ddl m Z ddlmZddZGdd d ejZGd d d ejZGd d d ejZGdddejZGdddejZdS)N)RotaryEmbedding)GlobalLayerNormCumulativeLayerNorm ScaleNorm)ScaledSinuEmbeddingFLASH_ShareA_FFConvMcCsL|dkr t||ddS|dkrt|ddS|dkr!tjd|ddSt|S) z0Just a wrapper to select the normalization type.glnT)elementwise_affineclnlng:0yE>eps)rrnn GroupNorm BatchNorm1d)normdimshaper_/home/ubuntu/.local/lib/python3.10/site-packages/funasr/models/mossformer/mossformer_encoder.py select_norms  rcs<eZdZddddddddfd d Zd d d dZZS)MossformerBlock@F皙? scalenormT) group_size query_key_dimexpansion_factorcausal attn_dropout norm_type shift_tokensc s~t|dvs Jd|dkrtn|dkrtj|_ttddtf ddt |D|_ dS) N)r layernormz/norm_type must be one of scalenorm or layernormrr& rc s&g|]}td qS)) rrr r!r"dropoutrotary_pos_emb norm_klassr%r).0_ r#r"rr!rr+r r*r%rr <s z,MossformerBlock.__init__..) super__init__rr LayerNormrrmin ModuleListrangelayers) selfrdepthrr r!r"r#r$r% __class__r.rr1s   zMossformerBlock.__init__NmaskcCs(d}|jD] }|||d}|d}q|S)Nrr;r )r6)r7xr<iiflashrrrforwardLs    zMossformerBlock.forward)__name__ __module__ __qualname__r1r@ __classcell__rrr9rrs-rcs6eZdZdZ      d fdd Zd d ZZS) MossFormer_MaskNetaThe MossFormer module for computing output masks. Arguments --------- in_channels : int Number of channels at the output of the encoder. out_channels : int Number of channels that would be inputted to the intra and inter blocks. num_blocks : int Number of layers of Dual Computation Block. norm : str Normalization type. num_spks : int Number of sources (speakers). skip_around_intra : bool Skip connection around intra. use_global_pos_enc : bool Global positional encodings. max_length : int Maximum sequence length. Example --------- >>> mossformer_block = MossFormerM(1, 64, 8) >>> mossformer_masknet = MossFormer_MaskNet(64, 64, intra_block, num_spks=2) >>> x = torch.randn(10, 64, 2000) >>> x = mossformer_masknet(x) >>> x.shape torch.Size([2, 10, 64, 2000]) r T Nc stt|||_||_t||d|_tj||ddd|_ ||_ |j r)t ||_ t ||||d|_tj|||dd|_tj||ddd|_t|_t|_tt||dt|_tt||dt|_dS)Nr F)bias)skip_around_intra) kernel_size)r0rEr1num_spks num_blocksrrrConv1dconv1d_encoderuse_global_pos_encrpos_encComputation_Blockmdl conv1d_out conv1_decoderPReLUpreluReLU activation SequentialTanhoutputSigmoid output_gate) r7 in_channels out_channelsrNrrMrKrQ max_lengthr9rrr1ts(     zMossFormer_MaskNet.__init__c Cs||}||}|jr$|}|dd}||}|dd}||}||}||}||}|j\}}}| ||j d|}| || |}| |}|j\}}}| ||j ||}||}|dd}|S)aReturns the output tensor. Arguments --------- x : torch.Tensor Input tensor of dimension [B, N, S]. Returns ------- out : torch.Tensor Output tensor of dimension [spks, B, N, S] where, spks = Number of speakers B = Batchsize, N = number of filters S = the number of time frames r r)rrPrQ transposerRrTrXrUrviewrMr]r_rVrZ) r7r=baseembBr-SNLrrrr@s(             zMossFormer_MaskNet.forward)rFr rGTTrHrArBrC__doc__r1r@rDrrr9rrETs#$rEcs*eZdZdZd fdd ZddZZS) MossFormerEncoderaConvolutional Encoder Layer. Arguments --------- kernel_size : int Length of filters. in_channels : int Number of input channels. out_channels : int Number of output channels. Example ------- >>> x = torch.randn(2, 1000) >>> encoder = Encoder(kernel_size=4, out_channels=64) >>> h = encoder(x) >>> h.shape torch.Size([2, 64, 499]) rG@r cs4tt|tj||||dddd|_||_dS)NrGr F)r`rarLstridegroupsrJ)r0rnr1rrOconv1dr`)r7rLrar`r9rrr1s zMossFormerEncoder.__init__cCs0|jdkr tj|dd}||}t|}|S)aReturn the encoded output. Arguments --------- x : torch.Tensor Input tensor with dimensionality [B, L]. Return ------ x : torch.Tensor Encoded tensor with dimensionality [B, N, T_out]. where B = Batchsize L = Number of timepoints N = Number of filters T_out = Number of timepoints at the output of the encoder r r()r`torch unsqueezerrFrelu)r7r=rrrr@s   zMossFormerEncoder.forward)rGror rlrrr9rrns rncs6eZdZdZ      d fdd Zd d ZZS) MossFormerMaThis class implements the transformer encoder. Arguments --------- num_blocks : int Number of mossformer blocks to include. d_model : int The dimension of the input embedding. attn_dropout : float Dropout for the self-attention (Optional). group_size: int the chunk size query_key_dim: int the attention vector dimension expansion_factor: int the expansion factor for the linear projection in conv module causal: bool true for causal / false for non causal Example ------- >>> import torch >>> x = torch.rand((8, 60, 512)) >>> net = TransformerEncoder_MossFormerM(num_blocks=8, d_model=512) >>> output, _ = net(x) >>> output.shape torch.Size([8, 60, 512]) NFrrrrc s6tt|||||||d|_tj|dd|_dS)N)rr8rr r!r"r#gư>r)r0r1r mossformerMrr2r)r7rNd_modelr"rr r!r#r9rrr13s  zMossFormerM.__init__cCs||}||}|S)a Arguments ---------- src : torch.Tensor Tensor shape [B, L, N], where, B = Batchsize, L = time points N = number of filters The sequence to the encoder layer (required). src_mask : tensor The mask for the src sequence (optional). src_key_padding_mask : tensor The mask for the src keys per batch (optional). )rxr)r7srcr]rrrr@Js  zMossFormerM.forward)NFrrrrrlrrr9rrws rwcs.eZdZdZ  dfdd ZddZZS) rSaComputation block for dual-path processing. Arguments --------- out_channels : int Dimensionality of inter/intra model. norm : str Normalization type. skip_around_intra : bool Skip connection around the intra layer. Example --------- >>> comp_block = Computation_Block(64) >>> x = torch.randn(10, 64, 100) >>> x = comp_block(x) >>> x.shape torch.Size([10, 64, 100]) r TcsFtt|t||d|_||_||_|dur!t||d|_dSdS)N)rNryrI) r0rSr1rw intra_mdlrKrr intra_norm)r7rNrarrKr9rrr1wszComputation_Block.__init__cCsd|j\}}}|ddd}||}|ddd}|jdur'||}|jr.||}|}|S)aoReturns the output tensor. Arguments --------- x : torch.Tensor Input tensor of dimension [B, N, S]. Return --------- out: torch.Tensor Output tensor of dimension [B, N, S]. where, B = Batchsize, N = number of filters S = sequence time index rrGr N)rpermute contiguousr{rr|rK)r7r=rhrjriintraoutrrrr@s    zComputation_Block.forward)r Trlrrr9rrSbs rS)rstorch.nnrtorch.nn.functional functionalrurotary_embedding_torchr$funasr.models.transformer.layer_normrrr#funasr.models.transformer.embeddingr#funasr.models.mossformer.mossformerrrModulerrErnrwrSrrrrs"    6<M