mirror of
https://github.com/modelscope/modelscope.git
synced 2026-09-01 19:49:03 +02:00
merge master ,加入speaker change locating pipeline
Link: https://code.alibaba-inc.com/Ali-MaaS/MaaS-lib/codereview/12601179
This commit is contained in:
tongmu.wh
@@ -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'
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
319
modelscope/models/audio/sv/speaker_change_locator.py
Normal file
319
modelscope/models/audio/sv/speaker_change_locator.py
Normal file
@@ -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))
|
||||
105
modelscope/pipelines/audio/speaker_change_locating_pipeline.py
Normal file
105
modelscope/pipelines/audio/speaker_change_locating_pipeline.py
Normal file
@@ -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
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user