o oiRE@sdZddlZddlmZddlmmZddlmZddl m Z ddl m Z ddlmZGddde Z Gd d d eZGd d d ejZGd ddejjZGdddejZGdddejZGdddejZGdddejjZGdddejjZdS)zQA popular speaker recognition and diarization model. Authors * Hwidong Na 2020 N)length_to_mask)Conv1d)Linear) BatchNorm1dc eZdZdZfddZZS)rz=1D convolution. Skip transpose is used to improve efficiency.ctj|ddi|dSNskip_transposeTsuper__init__selfargskwargs __class__a/home/ubuntu/SoloSpeech/.venv/lib/python3.10/site-packages/speechbrain/lobes/models/ECAPA_TDNN.pyr zConv1d.__init____name__ __module__ __qualname____doc__r __classcell__rrrrrrcr)rzE1D batch normalization. Skip transpose is used to improve efficiency.crrr r rrrr rzBatchNorm1d.__init__rrrrrrrrcs0eZdZdZejdffdd ZddZZS) TDNNBlockaAn implementation of TDNN. Arguments --------- in_channels : int Number of input channels. out_channels : int The number of output channels. kernel_size : int The kernel size of the TDNN blocks. dilation : int The dilation of the TDNN block. activation : torch class A class for constructing the activation layers. groups : int The groups size of the TDNN blocks. Example ------- >>> inp_tensor = torch.rand([8, 120, 64]).transpose(1, 2) >>> layer = TDNNBlock(64, 64, kernel_size=3, dilation=1) >>> out_tensor = layer(inp_tensor).transpose(1, 2) >>> out_tensor.shape torch.Size([8, 120, 64]) cs6tt|||||d|_||_t|d|_dS)N) in_channels out_channels kernel_sizedilationgroups input_size)r r rconv activationrnorm)rrr r!r"r'r#rrrr ;s zTDNNBlock.__init__cCs||||S:Processes the input tensor x and returns an output tensor.)r(r'r&)rxrrrforwardOszTDNNBlock.forward) rrrrnnReLUr r,rrrrrr s  rcs,eZdZdZ d fdd ZddZZS) Res2NetBlockaAn implementation of Res2NetBlock w/ dilation. Arguments --------- in_channels : int The number of channels expected in the input. out_channels : int The number of output channels. scale : int The scale of the Res2Net block. kernel_size: int The kernel size of the Res2Net block. dilation : int The dilation of the Res2Net block. Example ------- >>> inp_tensor = torch.rand([8, 120, 64]).transpose(1, 2) >>> layer = Res2NetBlock(64, 64, scale=4, dilation=3) >>> out_tensor = layer(inp_tensor).transpose(1, 2) >>> out_tensor.shape torch.Size([8, 120, 64]) rcslt||dks J||dksJ||||tfddt|dD|_||_dS)Nrcsg|] }tdqS))r!r")r).0ir"hidden_channel in_channelr!rr xsz)Res2NetBlock.__init__..r)r r r- ModuleListrangeblocksscale)rrr r;r!r"rr4rr ms   zRes2NetBlock.__init__cCsg}ttj||jddD])\}}|dkr|}n|dkr&|j|d|}n |j|d||}||q tj|dd}|S)r*rdimr) enumeratetorchchunkr;r:appendcat)rr+yr3x_iy_irrrr,s zRes2NetBlock.forward)r0r1rrrrrr r,rrrrrr/Ts r/cs*eZdZdZfddZdddZZS)SEBlocka3An implementation of squeeze-and-excitation block. Arguments --------- in_channels : int The number of input channels. se_channels : int The number of output channels after squeeze. out_channels : int The number of output channels. Example ------- >>> inp_tensor = torch.rand([8, 120, 64]).transpose(1, 2) >>> se_layer = SEBlock(64, 16, 64) >>> lengths = torch.rand((8,)) >>> out_tensor = se_layer(inp_tensor, lengths).transpose(1, 2) >>> out_tensor.shape torch.Size([8, 120, 64]) csJtt||dd|_tjjdd|_t||dd|_tj |_ dS)Nrrr r!T)inplace) r r rconv1r?r-r.reluconv2Sigmoidsigmoid)rr se_channelsr rrrr s zSEBlock.__init__NcCs|jd}|dur+t||||jd}|d}|jddd}||jddd|}n|jddd}|||}|| |}||S)r*Nmax_lendevicerTr=keepdim) shaperrS unsqueezesummeanrKrJrNrL)rr+lengthsLmasktotalsrrrr,s  zSEBlock.forwardNrFrrrrrGs  rGcs,eZdZdZd fdd Zd ddZZS) AttentiveStatisticsPoolingaThis class implements an attentive statistic pooling layer for each channel. It returns the concatenated mean and std of the input tensor. Arguments --------- channels: int The number of input channels. attention_channels: int The number of attention channels. global_context: bool Whether to use global context. Example ------- >>> inp_tensor = torch.rand([8, 120, 64]).transpose(1, 2) >>> asp_layer = AttentiveStatisticsPooling(64) >>> lengths = torch.rand((8,)) >>> out_tensor = asp_layer(inp_tensor, lengths).transpose(1, 2) >>> out_tensor.shape torch.Size([8, 1, 128]) Tcs^td|_||_|rt|d|dd|_nt||dd|_t|_t ||dd|_ dS)Ng-q=r1rrH) r r epsglobal_contextrtdnnr-Tanhtanhrr&)rchannelsattention_channelsrdrrrr s   z#AttentiveStatisticsPooling.__init__Nc Cs(|jd}d|jfdd}|durtj|jd|jd}t||||jd}|d }|jr_|jdd d  }||||\}}|d d d |}|d d d |}tj |||gd d } n|} | | || } | |dkt d } tj| dd } ||| \}}tj ||fd d } | d} | S)aeCalculates mean and std for a batch (input tensor). Arguments --------- x : torch.Tensor Tensor of shape [N, C, L]. lengths : torch.Tensor The corresponding relative lengths of the inputs. Returns ------- pooled_stats : torch.Tensor mean and std of batch rPrTcSs@|||}t||||d||}||fS)NrT)rYr?sqrtrXpowclamp)r+mr=rcrZstdrrr_compute_statisticss "z?AttentiveStatisticsPooling.forward.._compute_statisticsNrrSrQrTrUr<z-inf)rWrcr?onesrSrrXrdrYfloatrepeatrBr&rgre masked_fillFsoftmax) rr+r[r\ror]r^rZrnattn pooled_statsrrrr,s(   z"AttentiveStatisticsPooling.forward)rbTr`rFrrrrrasracs<eZdZdZddddejjdffdd Zd dd ZZ S) SERes2NetBlockaAn implementation of building block in ECAPA-TDNN, i.e., TDNN-Res2Net-TDNN-SEBlock. Arguments --------- in_channels: int Expected size of input channels. out_channels: int The number of output channels. res2net_scale: int The scale of the Res2Net block. se_channels : int The number of output channels after squeeze. kernel_size: int The kernel size of the TDNN blocks. dilation: int The dilation of the Res2Net block. activation : torch class A class for constructing the activation layers. groups: int Number of blocked connections from input channels to output channels. Example ------- >>> x = torch.rand(8, 120, 64).transpose(1, 2) >>> conv = SERes2NetBlock(64, 64, res2net_scale=4) >>> out = conv(x).transpose(1, 2) >>> out.shape torch.Size([8, 120, 64]) r0rbrc st||_t||dd||d|_t||||||_t||dd||d|_t||||_ d|_ ||kr?t ||dd|_ dSdS)Nr)r!r"r'r#rH) r r r rtdnn1r/ res2net_blocktdnn2rGse_blockshortcutr) rrr res2net_scalerOr!r"r'r#rrrr Fs<   zSERes2NetBlock.__init__NcCsF|}|jr ||}||}||}||}|||}||Sr))r~rzr{r|r})rr+r[residualrrrr,ps     zSERes2NetBlock.forwardr` rrrrr?r-r.r r,rrrrrry&s#*ryc sVeZdZdZddejjgdgdgddddd gd f fd d ZdddZZ S) ECAPA_TDNNa=An implementation of the speaker embedding model in a paper. "ECAPA-TDNN: Emphasized Channel Attention, Propagation and Aggregation in TDNN Based Speaker Verification" (https://arxiv.org/abs/2005.07143). Arguments --------- input_size : int Expected size of the input dimension. device : str Device used, e.g., "cpu" or "cuda". lin_neurons : int Number of neurons in linear layers. activation : torch class A class for constructing the activation layers. channels : list of ints Output channels for TDNN/SERes2Net layer. kernel_sizes : list of ints List of kernel sizes for each layer. dilations : list of ints List of dilations for kernels in each layer. attention_channels: int The number of attention channels. res2net_scale : int The scale of the Res2Net block. se_channels : int The number of output channels after squeeze. global_context: bool Whether to use global context. groups : list of ints List of groups for kernels in each layer. Example ------- >>> input_feats = torch.rand([5, 120, 80]) >>> compute_embedding = ECAPA_TDNN(80, lin_neurons=192) >>> outputs = compute_embedding(input_feats) >>> outputs.shape torch.Size([5, 1, 192]) cpu)rrri)r1r1r1r)rrTr1rrbr0T)rrrrrc  s6tt|t|ksJt|t|ksJ||_t|_|jt||d|d|d|| dt dt|dD]} |jt || d|| | | || || || | dq?t|dt|d|d|d|d|| dd|_ t |d|| d|_ t|ddd |_t|dd|dd |_dS) Nrr)rrOr!r"r'r#rTrP)r#)rirdr$rH)r r lenrhr-r8r:rArr9rymfaraasprasp_bnrfc)rr%rS lin_neuronsr'rh kernel_sizes dilationsrirrOrdr#r3rrrr s^       zECAPA_TDNN.__init__Nc Cs|dd}g}|jD]}z|||d}Wn ty"||}Ynw||q tj|dddd}||}|j||d}||}| |}|dd}|S)aEReturns the embedding vector. Arguments --------- x : torch.Tensor Tensor of shape (batch, time, channel). lengths : torch.Tensor Corresponding relative lengths of inputs. Returns ------- x : torch.Tensor Embedding vector. rrT)r[Nr<) transposer: TypeErrorrAr?rBrrrr)rr+r[xllayerrrrr,s         zECAPA_TDNN.forwardr`rrrrrr~s+Ircs2eZdZdZ    d fdd Zdd ZZS) ClassifieraThis class implements the cosine similarity on the top of features. Arguments --------- input_size : int Expected size of input dimension. device : str Device used, e.g., "cpu" or "cuda". lin_blocks : int Number of linear layers. lin_neurons : int Number of neurons in linear layers. out_neurons : int Number of classes. Example ------- >>> classify = Classifier(input_size=2, lin_neurons=2, out_neurons=2) >>> outputs = torch.tensor([ [1., -1.], [-9., 1.], [0.9, 0.1], [0.1, 0.9] ]) >>> outputs = outputs.unsqueeze(1) >>> cos = classify(outputs) >>> (cos < -1.0).long().sum() tensor(0) >>> (cos > 1.0).long().sum() tensor(0) rrrcsntt|_t|D]}|jt|dt||dg|}qt t j |||d|_ tj |j dS)Nr$)r% n_neuronsrp)r r r-r8r:r9extend _BatchNorm1dr Parameterr? FloatTensorweightinitxavier_uniform_)rr%rS lin_blocksr out_neurons block_indexrrrr 5s    zClassifier.__init__cCs>|jD]}||}qtt|dt|j}|dS)zReturns the output probabilities over speakers. Arguments --------- x : torch.Tensor Torch tensor. Returns ------- out : torch.Tensor Output probabilities over speakers. r)r:rulinear normalizesqueezerrX)rr+rrrrr,Os   zClassifier.forward)rrrrrFrrrrrsr)rr?torch.nnr-torch.nn.functional functionalruspeechbrain.dataio.dataiorspeechbrain.nnet.CNNr_Conv1dspeechbrain.nnet.linearrspeechbrain.nnet.normalizationrrModulerr/rGraryrrrrrrs$     4?3`X