Calculate forward propagation. Args: xs_pad (torch.Tensor): Input tensor (#batch, L, input_size). ilens (torch.Tensor): Input length (#batch). prev_states (torch.Tensor): Not to be used now. Returns: torch.Tensor: Output tensor (#batch,
(
self,
xs_pad: torch.Tensor,
ilens: torch.Tensor,
channel_size: torch.Tensor,
prev_states: torch.Tensor = None,
)
| 349 | return self._output_size |
| 350 | |
| 351 | def forward( |
| 352 | self, |
| 353 | xs_pad: torch.Tensor, |
| 354 | ilens: torch.Tensor, |
| 355 | channel_size: torch.Tensor, |
| 356 | prev_states: torch.Tensor = None, |
| 357 | ) -> Tuple[torch.Tensor, torch.Tensor, Optional[torch.Tensor]]: |
| 358 | """Calculate forward propagation. |
| 359 | Args: |
| 360 | xs_pad (torch.Tensor): Input tensor (#batch, L, input_size). |
| 361 | ilens (torch.Tensor): Input length (#batch). |
| 362 | prev_states (torch.Tensor): Not to be used now. |
| 363 | Returns: |
| 364 | torch.Tensor: Output tensor (#batch, L, output_size). |
| 365 | torch.Tensor: Output length (#batch). |
| 366 | torch.Tensor: Not to be used now. |
| 367 | """ |
| 368 | masks = (~make_pad_mask(ilens)[:, None, :]).to(xs_pad.device) |
| 369 | if ( |
| 370 | isinstance(self.embed, Conv2dSubsampling) |
| 371 | or isinstance(self.embed, Conv2dSubsampling6) |
| 372 | or isinstance(self.embed, Conv2dSubsampling8) |
| 373 | ): |
| 374 | short_status, limit_size = check_short_utt(self.embed, xs_pad.size(1)) |
| 375 | if short_status: |
| 376 | raise TooShortUttError( |
| 377 | f"has {xs_pad.size(1)} frames and is too short for subsampling " |
| 378 | + f"(it needs more than {limit_size} frames), return empty results", |
| 379 | xs_pad.size(1), |
| 380 | limit_size, |
| 381 | ) |
| 382 | xs_pad, masks = self.embed(xs_pad, masks) |
| 383 | else: |
| 384 | xs_pad = self.embed(xs_pad) |
| 385 | xs_pad, masks, channel_size = self.encoders(xs_pad, masks, channel_size) |
| 386 | if isinstance(xs_pad, tuple): |
| 387 | xs_pad = xs_pad[0] |
| 388 | |
| 389 | t_leng = xs_pad.size(1) |
| 390 | d_dim = xs_pad.size(2) |
| 391 | xs_pad = xs_pad.reshape(-1, channel_size, t_leng, d_dim) |
| 392 | if channel_size < 8: |
| 393 | repeat_num = math.ceil(8 / channel_size) |
| 394 | xs_pad = xs_pad.repeat(1, repeat_num, 1, 1)[:, 0:8, :, :] |
| 395 | xs_pad = self.conv1(xs_pad) |
| 396 | xs_pad = self.conv2(xs_pad) |
| 397 | xs_pad = self.conv3(xs_pad) |
| 398 | xs_pad = self.conv4(xs_pad) |
| 399 | xs_pad = xs_pad.squeeze().reshape(-1, t_leng, d_dim) |
| 400 | mask_tmp = masks.size(1) |
| 401 | masks = masks.reshape(-1, channel_size, mask_tmp, t_leng)[:, 0, :, :] |
| 402 | |
| 403 | if self.normalize_before: |
| 404 | xs_pad = self.after_norm(xs_pad) |
| 405 | |
| 406 | olens = masks.squeeze(1).sum(1) |
| 407 | return xs_pad, olens, None |
| 408 |
nothing calls this directly
no test coverage detected