From ca0cf2b5cfe47a1f73d60250915351dfaab33181 Mon Sep 17 00:00:00 2001 From: "tongmu.wh" Date: Fri, 12 May 2023 17:53:56 +0800 Subject: [PATCH] =?UTF-8?q?merge=20master=20,=E5=8A=A0=E5=85=A5speaker=20c?= =?UTF-8?q?hange=20locating=20pipeline=20Link:=20https://code.alibaba-inc.?= =?UTF-8?q?com/Ali-MaaS/MaaS-lib/codereview/12601179?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- modelscope/metainfo.py | 2 + modelscope/models/audio/sv/DTDNN.py | 18 +- .../models/audio/sv/speaker_change_locator.py | 319 ++++++++++++++++++ .../audio/speaker_change_locating_pipeline.py | 105 ++++++ tests/pipelines/test_speaker_verification.py | 20 +- 5 files changed, 457 insertions(+), 7 deletions(-) create mode 100644 modelscope/models/audio/sv/speaker_change_locator.py create mode 100644 modelscope/pipelines/audio/speaker_change_locating_pipeline.py diff --git a/modelscope/metainfo.py b/modelscope/metainfo.py index 22ad8b1a..efce05b0 100644 --- a/modelscope/metainfo.py +++ b/modelscope/metainfo.py @@ -182,6 +182,7 @@ class Models(object): generic_sv = 'generic-sv' ecapa_tdnn_sv = 'ecapa-tdnn-sv' campplus_sv = 'cam++-sv' + scl_sd = 'scl-sd' rdino_tdnn_sv = 'rdino_ecapa-tdnn-sv' generic_lm = 'generic-lm' @@ -481,6 +482,7 @@ class Pipelines(object): vad_inference = 'vad-inference' speaker_verification = 'speaker-verification' speaker_verification_rdino = 'speaker-verification-rdino' + speaker_change_locating = 'speaker-change-locating' lm_inference = 'language-score-prediction' speech_timestamp_inference = 'speech-timestamp-inference' diff --git a/modelscope/models/audio/sv/DTDNN.py b/modelscope/models/audio/sv/DTDNN.py index d9e21ce8..d86d6799 100644 --- a/modelscope/models/audio/sv/DTDNN.py +++ b/modelscope/models/audio/sv/DTDNN.py @@ -76,11 +76,13 @@ class CAMPPlus(nn.Module): bn_size=4, init_channels=128, config_str='batchnorm-relu', - memory_efficient=True): + memory_efficient=True, + output_level='segment'): super(CAMPPlus, self).__init__() self.head = FCM(feat_dim=feat_dim) channels = self.head.out_channels + self.output_level = output_level self.xvector = nn.Sequential( OrderedDict([ @@ -118,10 +120,14 @@ class CAMPPlus(nn.Module): self.xvector.add_module('out_nonlinear', get_nonlinear(config_str, channels)) - self.xvector.add_module('stats', StatsPool()) - self.xvector.add_module( - 'dense', - DenseLayer(channels * 2, embedding_size, config_str='batchnorm_')) + if self.output_level == 'segment': + self.xvector.add_module('stats', StatsPool()) + self.xvector.add_module( + 'dense', + DenseLayer( + channels * 2, embedding_size, config_str='batchnorm_')) + else: + assert self.output_level == 'frame', '`output_level` should be set to \'segment\' or \'frame\'. ' for m in self.modules(): if isinstance(m, (nn.Conv1d, nn.Linear)): @@ -133,6 +139,8 @@ class CAMPPlus(nn.Module): x = x.permute(0, 2, 1) # (B,T,F) => (B,F,T) x = self.head(x) x = self.xvector(x) + if self.output_level == 'frame': + x = x.transpose(1, 2) return x diff --git a/modelscope/models/audio/sv/speaker_change_locator.py b/modelscope/models/audio/sv/speaker_change_locator.py new file mode 100644 index 00000000..c22e4c1b --- /dev/null +++ b/modelscope/models/audio/sv/speaker_change_locator.py @@ -0,0 +1,319 @@ +# Copyright (c) Alibaba, Inc. and its affiliates. + +import os +from collections import OrderedDict +from typing import Any, Dict, Union + +import numpy as np +import torch +import torch.nn as nn +import torch.nn.functional as F +import torchaudio.compliance.kaldi as Kaldi + +from modelscope.metainfo import Models +from modelscope.models import MODELS, TorchModel +from modelscope.models.audio.sv.DTDNN import CAMPPlus +from modelscope.utils.constant import Tasks + + +class MultiHeadSelfAttention(nn.Module): + + def __init__(self, n_units, h=8, dropout=0.1): + super(MultiHeadSelfAttention, self).__init__() + self.linearQ = nn.Linear(n_units, n_units) + self.linearK = nn.Linear(n_units, n_units) + self.linearV = nn.Linear(n_units, n_units) + self.linearO = nn.Linear(n_units, n_units) + self.d_k = n_units // h + self.h = h + self.dropout = nn.Dropout(p=dropout) + self.att = None + + def forward(self, x, batch_size): + # x: (BT, F) + q = self.linearQ(x).reshape(batch_size, -1, self.h, self.d_k) + k = self.linearK(x).reshape(batch_size, -1, self.h, self.d_k) + v = self.linearV(x).reshape(batch_size, -1, self.h, self.d_k) + scores = torch.matmul(q.transpose(1, 2), k.permute( + 0, 2, 3, 1)) / np.sqrt(self.d_k) + # scores: (B, h, T, T) + self.att = F.softmax(scores, dim=3) + p_att = self.dropout(self.att) + # v : (B, T, h, d_k) + # p_att : (B, h, T, T) + x = torch.matmul(p_att, v.transpose(1, 2)) + # x : (B, h, T, d_k) + x = x.transpose(1, 2).reshape(-1, self.h * self.d_k) + return self.linearO(x) + + +class PositionwiseFeedForward(nn.Module): + + def __init__(self, n_units, d_units, dropout): + super(PositionwiseFeedForward, self).__init__() + self.linear1 = nn.Linear(n_units, d_units) + self.linear2 = nn.Linear(d_units, n_units) + self.dropout = nn.Dropout(p=dropout) + + def forward(self, x): + return self.linear2(self.dropout(F.relu(self.linear1(x)))) + + +class PosEncoding(nn.Module): + + def __init__(self, max_seq_len, d_word_vec): + super(PosEncoding, self).__init__() + pos_enc = np.array([[ + pos / np.power(10000, 2.0 * (j // 2) / d_word_vec) + for j in range(d_word_vec) + ] for pos in range(max_seq_len)]) + pos_enc[:, 0::2] = np.sin(pos_enc[:, 0::2]) + pos_enc[:, 1::2] = np.cos(pos_enc[:, 1::2]) + pad_row = np.zeros([1, d_word_vec]) + pos_enc = np.concatenate([pad_row, pos_enc]).astype(np.float32) + + self.pos_enc = torch.nn.Embedding(max_seq_len + 1, d_word_vec) + self.pos_enc.weight = torch.nn.Parameter( + torch.from_numpy(pos_enc), requires_grad=False) + + def forward(self, input_len): + max_len = torch.max(input_len) + input_pos = torch.LongTensor([ + list(range(1, len + 1)) + [0] * (max_len - len) + for len in input_len + ]) + + return self.pos_enc(input_pos) + + +class TransformerEncoder(nn.Module): + + def __init__(self, + idim, + n_units=256, + n_layers=2, + e_units=512, + h=4, + dropout=0.1): + super(TransformerEncoder, self).__init__() + self.linear_in = nn.Linear(idim, n_units) + self.lnorm_in = nn.LayerNorm(n_units) + + self.n_layers = n_layers + self.dropout = nn.Dropout(p=dropout) + for i in range(n_layers): + setattr(self, '{}{:d}'.format('lnorm1_', i), nn.LayerNorm(n_units)) + setattr(self, '{}{:d}'.format('self_att_', i), + MultiHeadSelfAttention(n_units, h)) + setattr(self, '{}{:d}'.format('lnorm2_', i), nn.LayerNorm(n_units)) + setattr(self, '{}{:d}'.format('ff_', i), + PositionwiseFeedForward(n_units, e_units, dropout)) + self.lnorm_out = nn.LayerNorm(n_units) + + def forward(self, x): + # x: [B, num_anchors, T, n_in] + bs, num, tframe, dim = x.size() + x = x.reshape(bs * num, tframe, -1) # [B*num_anchors, T, dim] + # x: (B, T, F) ... batch, time, (mel)freq + B_size, T_size, _ = x.shape + # e: (BT, F) + e = self.linear_in(x.reshape(B_size * T_size, -1)) + # Encoder stack + for i in range(self.n_layers): + # layer normalization + e = getattr(self, '{}{:d}'.format('lnorm1_', i))(e) + # self-attention + s = getattr(self, '{}{:d}'.format('self_att_', i))(e, x.shape[0]) + # residual + e = e + self.dropout(s) + # layer normalization + e = getattr(self, '{}{:d}'.format('lnorm2_', i))(e) + # positionwise feed-forward + s = getattr(self, '{}{:d}'.format('ff_', i))(e) + # residual + e = e + self.dropout(s) + # final layer normalization + # output: (BT, F) + # output: (B, F, T) + output = self.lnorm_out(e).reshape(B_size, T_size, -1) + output = output.reshape(bs, num, tframe, + -1) # [B, num_anchors, T, dim] + return output + + +class TransformerEncoder_out(nn.Module): + + def __init__(self, + idim, + n_units=256, + n_layers=2, + e_units=512, + h=4, + dropout=0.1): + super(TransformerEncoder_out, self).__init__() + self.linear_in = nn.Linear(idim, n_units) + self.lnorm_in = nn.LayerNorm(n_units) + + self.n_layers = n_layers + self.dropout = nn.Dropout(p=dropout) + for i in range(n_layers): + setattr(self, '{}{:d}'.format('lnorm1_', i), nn.LayerNorm(n_units)) + setattr(self, '{}{:d}'.format('self_att_', i), + MultiHeadSelfAttention(n_units, h)) + setattr(self, '{}{:d}'.format('lnorm2_', i), nn.LayerNorm(n_units)) + setattr(self, '{}{:d}'.format('ff_', i), + PositionwiseFeedForward(n_units, e_units, dropout)) + self.lnorm_out = nn.LayerNorm(n_units) + + def forward(self, x): + # x: (B, T, F) + B_size, T_size, _ = x.shape + # e: (BT, F) + e = self.linear_in(x.reshape(B_size * T_size, -1)) + # Encoder stack + for i in range(self.n_layers): + # layer normalization + e = getattr(self, '{}{:d}'.format('lnorm1_', i))(e) + # self-attention + s = getattr(self, '{}{:d}'.format('self_att_', i))(e, x.shape[0]) + # residual + e = e + self.dropout(s) + # layer normalization + e = getattr(self, '{}{:d}'.format('lnorm2_', i))(e) + # positionwise feed-forward + s = getattr(self, '{}{:d}'.format('ff_', i))(e) + # residual + e = e + self.dropout(s) + # final layer normalization + # output: (BT, F) + # output: (B, T, F) + output = self.lnorm_out(e).reshape(B_size, T_size, -1) + return output + + +class OutLayer(nn.Module): + + def __init__(self, n_units=256, num_anchors=2): + super(OutLayer, self).__init__() + self.combine = TransformerEncoder_out(num_anchors * n_units, n_units) + self.out_linear = nn.Linear(n_units // num_anchors, 1) + + def forward(self, input): + # input: [B, num_anchors, T, dim] + bs, num, tframe, dim = input.size() + output = input.permute(0, 2, 1, + 3).reshape(bs, tframe, + -1) # [Bs, t, num_anchors*dim] + output = self.combine(output) # [Bs, t, n_units] + output = output.reshape( + bs, tframe, num, -1) # [Bs, t, num_anchors, n_units//num_anchors] + output = self.out_linear(output).squeeze(-1) # [Bs, t, num_anchors] + + return output + + +class TransformerDetector(nn.Module): + + def __init__(self, + frame_dim=512, + anchor_dim=192, + hidden_dim=256, + max_seq_len=1000): + super(TransformerDetector, self).__init__() + self.detection = TransformerEncoder( + idim=frame_dim + anchor_dim, n_units=hidden_dim) + self.output = OutLayer(n_units=hidden_dim) + self.pos_enc = PosEncoding(max_seq_len, hidden_dim) + + def forward(self, feats, anchors): + # feats: [1, t, fdim] + num_frames = feats.shape[1] + num_anchors = anchors.shape[1] + bs = feats.shape[0] + feats = feats.unsqueeze(1).repeat( + 1, num_anchors, 1, 1) # shape: [Bs, num_anchors, t, fdim] + anchors = anchors.unsqueeze(2).repeat( + 1, 1, num_frames, 1) # shape: [Bs, num_anchors, t, xdim] + sd_in = torch.cat((feats, anchors), + dim=-1) # shape: [Bs, num_anchors, t, fdim+xdim] + sd_out = self.detection(sd_in) # shape: [Bs, num_anchors, t, sd_dim] + + # pos + pos_emb = self.pos_enc(torch.tensor([num_frames] * (bs * num_anchors))) + pos_emb = pos_emb.reshape(bs, num_anchors, num_frames, -1) + sd_out += pos_emb + + # output + output = self.output(sd_out) # shape: [Bs, t, num_anchors] + + return output + + +@MODELS.register_module(Tasks.speaker_diarization, module_name=Models.scl_sd) +class SpeakerChangeLocatorTransformer(TorchModel): + r"""A speaekr change locator using the transformer architecture as the backbone. + Args: + model_dir: A model dir. + model_config: The model config. + """ + + def __init__(self, model_dir, model_config: Dict[str, Any], *args, + **kwargs): + super().__init__(model_dir, model_config, *args, **kwargs) + self.model_config = model_config + + self.feature_dim = self.model_config['fbank_dim'] + frame_size = self.model_config['frame_size'] + anchor_size = self.model_config['anchor_size'] + + self.encoder = CAMPPlus(self.feature_dim, output_level='frame') + self.backend = TransformerDetector( + frame_dim=frame_size, anchor_dim=anchor_size) + + pretrained_encoder = kwargs['pretrained_encoder'] + pretrained_backend = kwargs['pretrained_backend'] + + self.__load_check_point(pretrained_encoder, pretrained_backend) + + self.encoder.eval() + self.backend.eval() + + def forward(self, audio, anchors): + assert len(audio.shape) == 2 and audio.shape[ + 0] == 1, 'modelscope error: the shape of input audio to model needs to be [1, T]' + assert len( + anchors.shape + ) == 3 and anchors.shape[0] == 1 and anchors.shape[ + 1] == 2, 'modelscope error: the shape of input anchors to model needs to be [1, 2, D]' + # audio shape: [1, T] + feature = self.__extract_feature(audio) + frame_state = self.encoder(feature) + output = self.backend(frame_state, anchors) + output = output.squeeze(0).detach().cpu().sigmoid() + + time_scale_factor = int(np.ceil(feature.shape[1] / output.shape[0])) + output = output.unsqueeze(1).expand(-1, time_scale_factor, + -1).reshape(-1, output.shape[-1]) + return output + + def __extract_feature(self, audio): + feature = Kaldi.fbank(audio, num_mel_bins=self.feature_dim) + feature = feature - feature.mean(dim=0, keepdim=True) + feature = feature.unsqueeze(0) + return feature + + def __load_check_point(self, + pretrained_encoder, + pretrained_backend, + device=None): + if not device: + device = torch.device('cpu') + self.encoder.load_state_dict( + torch.load( + os.path.join(self.model_dir, pretrained_encoder), + map_location=device)) + + self.backend.load_state_dict( + torch.load( + os.path.join(self.model_dir, pretrained_backend), + map_location=device)) diff --git a/modelscope/pipelines/audio/speaker_change_locating_pipeline.py b/modelscope/pipelines/audio/speaker_change_locating_pipeline.py new file mode 100644 index 00000000..0bab08ac --- /dev/null +++ b/modelscope/pipelines/audio/speaker_change_locating_pipeline.py @@ -0,0 +1,105 @@ +# Copyright (c) Alibaba, Inc. and its affiliates. + +import io +from typing import Any, Dict, List, Union + +import numpy as np +import soundfile as sf +import torch + +from modelscope.fileio import File +from modelscope.metainfo import Pipelines +from modelscope.outputs import OutputKeys +from modelscope.pipelines.base import InputModel, Pipeline +from modelscope.pipelines.builder import PIPELINES +from modelscope.utils.constant import Tasks +from modelscope.utils.logger import get_logger + +logger = get_logger() + +__all__ = ['SpeakerChangeLocatingPipeline'] + + +@PIPELINES.register_module( + Tasks.speaker_diarization, module_name=Pipelines.speaker_change_locating) +class SpeakerChangeLocatingPipeline(Pipeline): + """Speaker Change Locating Inference Pipeline + use `model` to create a speaker change Locating pipeline. + + Args: + model (SpeakerChangeLocatingPipeline): A model instance, or a model local dir, or a model id in the model hub. + kwargs (dict, `optional`): + Extra kwargs passed into the pipeline's constructor. + Example: + >>> from modelscope.pipelines import pipeline + >>> from modelscope.utils.constant import Tasks + >>> p = pipeline( + >>> task=Tasks.speaker_diarization, model='damo/speech_campplus-transformer_scl_zh-cn_16k-common') + >>> print(p(audio)) + + """ + + def __init__(self, model: InputModel, **kwargs): + """use `model` to create a speaker change Locating pipeline for prediction + Args: + model (str): a valid offical model id + """ + super().__init__(model=model, **kwargs) + self.model_config = self.model.model_config + self.config = self.model.model_config + self.anchor_size = self.config['anchor_size'] + + def __call__(self, audio: str, embds: List = None) -> Dict[str, Any]: + if embds is not None: + assert len(embds) == 2 + assert isinstance(embds[0], np.ndarray) and isinstance( + embds[1], np.ndarray) + assert embds[0].shape == ( + self.anchor_size, ) and embds[1].shape == (self.anchor_size, ) + else: + embd1 = np.zeros(self.anchor_size // 2) + embd2 = np.ones(self.anchor_size - self.anchor_size // 2) + embd3 = np.ones(self.anchor_size // 2) + embd4 = np.zeros(self.anchor_size - self.anchor_size // 2) + embds = [ + np.stack([embd1, embd2], axis=1).flatten(), + np.stack([embd3, embd4], axis=1).flatten(), + ] + anchors = torch.from_numpy(np.stack(embds, + axis=0)).float().unsqueeze(0) + + output = self.preprocess(audio) + output = self.forward(output, anchors) + output = self.postprocess(output) + + return output + + def forward(self, input: torch.Tensor, anchors: torch.Tensor): + output = self.model(input, anchors) + return output + + def postprocess(self, input: torch.Tensor) -> Dict[str, Any]: + predict = np.where(np.diff(input.argmax(-1).numpy())) + try: + predict = predict[0][0] * 0.01 + 0.02 + predict = round(predict, 2) + return {OutputKeys.TEXT: f'The change point is at {predict}s.'} + except Exception: + return {OutputKeys.TEXT: 'No change point is found.'} + + def preprocess(self, input: str) -> torch.Tensor: + if isinstance(input, str): + file_bytes = File.read(input) + data, fs = sf.read(io.BytesIO(file_bytes), dtype='float32') + if len(data.shape) == 2: + data = data[:, 0] + if fs != self.model_config['sample_rate']: + raise ValueError( + 'modelscope error: Only support %d sample rate files' + % self.model_cfg['sample_rate']) + data = torch.from_numpy(data).unsqueeze(0) + else: + raise ValueError( + 'modelscope error: The input type is restricted to audio file address' + % i) + return data diff --git a/tests/pipelines/test_speaker_verification.py b/tests/pipelines/test_speaker_verification.py index 236b952b..41335fc8 100644 --- a/tests/pipelines/test_speaker_verification.py +++ b/tests/pipelines/test_speaker_verification.py @@ -2,7 +2,7 @@ import os.path import unittest -from typing import Any, Dict, List +from typing import Any, Dict, List, Union from modelscope.outputs import OutputKeys from modelscope.pipelines import pipeline @@ -16,20 +16,25 @@ logger = get_logger() SPEAKER1_A_EN_16K_WAV = 'data/test/audios/speaker1_a_en_16k.wav' SPEAKER1_B_EN_16K_WAV = 'data/test/audios/speaker1_b_en_16k.wav' SPEAKER2_A_EN_16K_WAV = 'data/test/audios/speaker2_a_en_16k.wav' +SCL_EXAMPLE_WAV = 'data/test/audios/scl_example1.wav' class SpeakerVerificationTest(unittest.TestCase, DemoCompatibilityCheck): ecapatdnn_voxceleb_16k_model_id = 'damo/speech_ecapa-tdnn_sv_en_voxceleb_16k' campplus_voxceleb_16k_model_id = 'damo/speech_campplus_sv_en_voxceleb_16k' rdino_voxceleb_16k_model_id = 'damo/speech_rdino_ecapa_tdnn_sv_en_voxceleb_16k' + speaker_change_locating_cn_model_id = 'damo/speech_campplus-transformer_scl_zh-cn_16k-common' def setUp(self) -> None: self.task = Tasks.speaker_verification def run_pipeline(self, model_id: str, - audios: List[str], + audios: Union[List[str], str], + task: str = None, model_revision=None) -> Dict[str, Any]: + if task is not None: + self.task = task p = pipeline( task=self.task, model=model_id, model_revision=model_revision) result = p(audios) @@ -66,6 +71,17 @@ class SpeakerVerificationTest(unittest.TestCase, DemoCompatibilityCheck): print(result) self.assertTrue(OutputKeys.SCORE in result) + @unittest.skipUnless(test_level() >= 0, 'skip test in current test level') + def test_run_with_speaker_change_locating_cn_16k(self): + logger.info( + 'Run speaker change locating for campplus-transformer model') + result = self.run_pipeline( + model_id=self.speaker_change_locating_cn_model_id, + task=Tasks.speaker_diarization, + audios=SCL_EXAMPLE_WAV) + print(result) + self.assertTrue(OutputKeys.TEXT in result) + @unittest.skip('demo compatibility test is only enabled on a needed-basis') def test_demo_compatibility(self): self.compatibility_check()