| | |
| | | model, punc_train_args = PUNCTask.build_model_from_file( |
| | | punc_train_config, punc_model_file, 'cpu' |
| | | ) |
| | | elif mode.startswith('punc_VadRealtime'): |
| | | from funasr.tasks.punctuation import PunctuationTask as PUNCTask |
| | | punc_train_config = os.path.join(model_dir, 'config.yaml') |
| | | punc_model_file = os.path.join(model_dir, 'punc.pb') |
| | | model, punc_train_args = PUNCTask.build_model_from_file( |
| | | punc_train_config, punc_model_file, 'cpu' |
| | | ) |
| | | self._export(model, tag_name) |
| | | |
| | | |
| | |
| | | from funasr.modules.positionwise_feed_forward import PositionwiseFeedForward |
| | | from funasr.export.models.modules.feedforward import PositionwiseFeedForward as PositionwiseFeedForward_export |
| | | |
| | | def subsequent_mask(size, device="cpu", dtype=torch.bool): |
| | | """Create mask for subsequent steps (size, size). |
| | | |
| | | :param int size: size of mask |
| | | :param str device: "cpu" or "cuda" or torch.Tensor.device |
| | | :param torch.dtype dtype: result dtype |
| | | :rtype: torch.Tensor |
| | | >>> subsequent_mask(3) |
| | | [[1, 0, 0], |
| | | [1, 1, 0], |
| | | [1, 1, 1]] |
| | | """ |
| | | ret = torch.ones(size, size, device=device, dtype=dtype) |
| | | return torch.tril(ret, out=ret) |
| | | |
| | | class SANMEncoder(nn.Module): |
| | | def __init__( |
| | | self, |
| | |
| | | |
| | | def prepare_mask(self, mask): |
| | | mask_3d_btd = mask[:, :, None] |
| | | sub_masks = subsequent_mask(mask.size(-1)) |
| | | if len(mask.shape) == 2: |
| | | mask_4d_bhlt = 1 - mask[:, None, None, :] |
| | | mask_4d_bhlt = 1 - sub_masks[:, None, None, :] |
| | | elif len(mask.shape) == 3: |
| | | mask_4d_bhlt = 1 - mask[:, None, :] |
| | | mask_4d_bhlt = 1 - sub_masks[:, None, :] |
| | | mask_4d_bhlt = mask_4d_bhlt * -10000.0 |
| | | |
| | | return mask_3d_btd, mask_4d_bhlt |
| | |
| | | def forward(self, |
| | | speech: torch.Tensor, |
| | | speech_lengths: torch.Tensor, |
| | | vad_mask: torch.Tensor, |
| | | ): |
| | | speech = speech * self._output_size ** 0.5 |
| | | mask = self.make_pad_mask(speech_lengths) |
| | |
| | | encoder_outs = self.model.encoders0(xs_pad, mask) |
| | | xs_pad, masks = encoder_outs[0], encoder_outs[1] |
| | | |
| | | encoder_outs = self.model.encoders(xs_pad, mask) |
| | | xs_pad, masks = encoder_outs[0], encoder_outs[1] |
| | | # encoder_outs = self.model.encoders(xs_pad, mask) |
| | | for layer_idx, encoder_layer in enumerate(self.model.encoders): |
| | | if layer_idx == len(self.model.encoders) - 1: |
| | | mask = (mask[0], vad_mask) |
| | | encoder_outs = encoder_layer(xs_pad, mask) |
| | | xs_pad, masks = encoder_outs[0], encoder_outs[1] |
| | | |
| | | xs_pad = self.model.after_norm(xs_pad) |
| | | |
| | |
| | | **kwargs, |
| | | ): |
| | | super().__init__() |
| | | |
| | | onnx = False |
| | | if "onnx" in kwargs: |
| | | onnx = kwargs["onnx"] |
| | | |
| | | self.embed = model.embed |
| | | if isinstance(model.encoder, SANMVadEncoder): |
| | |
| | | |
| | | def get_dummy_inputs(self): |
| | | length = 120 |
| | | text_indexes = torch.randint(0, self.embed.num_embeddings, (2, length)) |
| | | text_lengths = torch.tensor([length-20, length], dtype=torch.int32) |
| | | return (text_indexes, text_lengths) |
| | | text_indexes = torch.randint(0, self.embed.num_embeddings, (1, length)) |
| | | text_lengths = torch.tensor([length], dtype=torch.int32) |
| | | vad_mask = torch.ones(length, length)[None, None, :, :] |
| | | return (text_indexes, text_lengths, vad_mask) |
| | | |
| | | def get_input_names(self): |
| | | return ['input', 'text_lengths'] |
| | | return ['input', 'text_lengths', 'vad_mask'] |
| | | |
| | | def get_output_names(self): |
| | | return ['logits'] |
| | |
| | | def get_dynamic_axes(self): |
| | | return { |
| | | 'input': { |
| | | 0: 'batch_size', |
| | | 1: 'feats_length' |
| | | }, |
| | | 'text_lengths': { |
| | | 0: 'batch_size', |
| | | }, |
| | | 'logits': { |
| | | 0: 'batch_size', |
| | | 1: 'logits_length' |
| | | }, |
| | | } |