| 49 | |
| 50 | |
| 51 | class StreamConv2d(nn.Module): |
| 52 | def __init__(self, |
| 53 | in_channels: int, |
| 54 | out_channels: int, |
| 55 | kernel_size: Union[int, Tuple[int, int]], |
| 56 | stride: Union[int, Tuple[int, int]] = 1, |
| 57 | padding: Union[str, int, Tuple[int, int]] = 0, |
| 58 | dilation: Union[int, Tuple[int, int]] = 1, |
| 59 | groups: int = 1, |
| 60 | bias: bool = True, |
| 61 | *args, **kargs): |
| 62 | super().__init__(*args, **kargs) |
| 63 | """ |
| 64 | kernel_size = [T_size, F_size] by defalut |
| 65 | """ |
| 66 | if type(padding) is int: |
| 67 | self.T_pad = padding |
| 68 | self.F_pad = padding |
| 69 | elif type(padding) in [list, tuple]: |
| 70 | self.T_pad, self.F_pad = padding |
| 71 | else: |
| 72 | raise ValueError('Invalid padding size.') |
| 73 | |
| 74 | assert self.T_pad == 0, "To meet the demands of causal streaming requirements" |
| 75 | |
| 76 | self.Conv2d = nn.Conv2d(in_channels = in_channels, |
| 77 | out_channels = out_channels, |
| 78 | kernel_size = kernel_size, |
| 79 | stride = stride, |
| 80 | padding = padding, |
| 81 | dilation = dilation, |
| 82 | groups = groups, |
| 83 | bias = bias) |
| 84 | |
| 85 | def forward(self, x, cache): |
| 86 | """ |
| 87 | x: [bs, C, 1, F] |
| 88 | cache: [bs, C, T_size-1, F] |
| 89 | """ |
| 90 | inp = torch.cat([cache, x], dim=2) |
| 91 | outp = self.Conv2d(inp) |
| 92 | out_cache = inp[:,:, 1:] |
| 93 | return outp, out_cache |
| 94 | |
| 95 | ## Version 1 |
| 96 | ## The inference of this implementation is slow, according to https://github.com/Xiaobin-Rong/gtcrn/issues/37 |