o %ݫiG@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__W/home/ubuntu/.local/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__rrrrrrrrcs2eZdZdZejddffdd 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. dropout : float Rate of channel dropout during training. 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]) csDtt|||||d|_||_t|d|_tj|d|_ dS)N) in_channels out_channels kernel_sizedilationgroups input_size)p) r r rconv activationrnormnn Dropout1ddropout)rr r!r"r#r)r$r-rrrr =s  zTDNNBlock.__init__c Cs|||||S:Processes the input tensor x and returns an output tensor.)r-r*r)r()rxrrrforwardSszTDNNBlock.forward) rrrrr+ReLUr r1rrrrrr s"rc2eZdZdZ    d fdd Zdd ZZS) 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. dropout : float Rate of channel dropout during training. 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]) rrcsnt||dks J||dksJ||||tfddt|dD|_||_dS)Nrc sg|] }tdqS))r"r#r-)r).0ir#r-hidden_channel in_channelr"rr sz)Res2NetBlock.__init__..r)r r r+ ModuleListrangeblocksscale)rr r!r@r"r#r-rr9rr ss   zRes2NetBlock.__init__cCsg}ttj||jddD])\}}|dkr|}n|dkr&|j|d|}n |j|d||}||q tj|dd}|S)r/rdimr) enumeratetorchchunkr@r?appendcat)rr0yr8x_iy_irrrr1s zRes2NetBlock.forward)r5r6rrrrrrr r1rrrrrr4Xsr4cs*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)Nrr r!r"T)inplace) r r rconv1rDr+r2reluconv2Sigmoidsigmoid)rr se_channelsr!rrrr s zSEBlock.__init__NcCs|jd}|dur+t||||jd}|d}|jddd}||jddd|}n|jddd}|||}|| |}||S)r/Nmax_lendevicerTrBkeepdim) shaperrX unsqueezesummeanrPrOrSrQ)rr0lengthsLmasktotalsrrrr1s  zSEBlock.forwardNrKrrrrrLs  rLcs,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=r6rrM) r r epsglobal_contextrtdnnr+Tanhtanhrr()rchannelsattention_channelsrirrrr 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 rUrYcSs@|||}t||||d||}||fS)NrY)r^rDsqrtr]powclamp)r0mrBrhr_stdrrr_compute_statistics s "z?AttentiveStatisticsPooling.forward.._compute_statisticsNrrXrVrTrZrAz-inf)r\rhrDonesrXrr]rir^floatrepeatrGr(rlrj masked_fillFsoftmax) rr0r`rartrbrcr_rsattn pooled_statsrrrr1s(   z"AttentiveStatisticsPooling.forward)rgTrerKrrrrrfsrfcs>eZdZdZddddejjddffdd Zd d d 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. dropout: float Rate of channel dropout during training. 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]) r5rgrrc st||_t||dd||| d|_t||||||_t||dd||| d|_t||||_ d|_ ||krAt ||dd|_ dSdS)Nr)r"r#r)r$r-rM) r r r!rtdnn1r4 res2net_blocktdnn2rLse_blockshortcutr) rr r! res2net_scalerTr"r#r)r$r-rrrr Us@    zSERes2NetBlock.__init__NcCsF|}|jr ||}||}||}||}|||}||Sr.)rrrrr)rr0r`residualrrrr1s     zSERes2NetBlock.forwardre rrrrrDr+r2r r1rrrrrr~3s%-r~c sXeZdZdZddejjgdgdgddddd gd d f fd d ZdddZZ S) ECAPA_TDNNaAn 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. dropout : float Rate of channel dropout during training. 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)r6r6r6r)rrYr6rrgr5T)rrrrrrcs<tt|t|ksJt|t|ksJ||_t|_|jt||d|d|d|| d| t dt|dD]}|jt ||d||| | |||||| || d q@t|dt|d|d|d|d|| d| d|_ t |d|| d|_ t|ddd |_t|dd|dd |_dS) Nrr)rrTr"r#r)r$r-rYrU)r$r-)rnrir%rM)r r lenrmr+r=r?rFrr>r~mfarfasprasp_bnrfc)rr&rX lin_neuronsr)rm kernel_sizes dilationsrnrrTrir$r-r8rrrr sd       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. rrY)r`NrA) transposer? TypeErrorrFrDrGrrrr)rr0r`xllayerrrrr1s         zECAPA_TDNN.forwardrerrrrrrs -Mrcr3) 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_neuronsru)r r r+r=r?r>extend _BatchNorm1dr ParameterrD FloatTensorweightinitxavier_uniform_)rr&rX lin_blocksr out_neurons block_indexrrrr Ms    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?rzlinear normalizesqueezerr])rr0rrrrr1gs   zClassifier.forward)rrrrrKrrrrr1sr)rrDtorch.nnr+torch.nn.functional functionalrzspeechbrain.dataio.dataiorspeechbrain.nnet.CNNr_Conv1dspeechbrain.nnet.linearrspeechbrain.nnet.normalizationrrModulerr4rLrfr~rrrrrrs$     8H3`]"