| | import torch |
| | import torch.nn as nn |
| | import torch.nn.functional as F |
| |
|
| |
|
| | def nonlinearity(x): |
| | |
| | return x * torch.sigmoid(x) |
| |
|
| |
|
| | class Normalize(nn.Module): |
| | def __init__(self, channels, eps=1e-5): |
| | super().__init__() |
| | self.channels = channels |
| | self.eps = eps |
| |
|
| | self.gamma = nn.Parameter(torch.ones(channels)) |
| | self.beta = nn.Parameter(torch.zeros(channels)) |
| | self.proj = nn.Linear(channels, channels) |
| |
|
| | def forward(self, x): |
| | x = x.transpose(1, 2) |
| | x = F.layer_norm(x, (self.channels,), self.gamma, self.beta, self.eps) |
| | x = self.proj(x) |
| | return x.transpose(1, 2) |
| |
|
| |
|
| | class Upsample(nn.Module): |
| | def __init__(self, in_channels, with_conv): |
| | super().__init__() |
| | self.with_conv = with_conv |
| | if self.with_conv: |
| | self.conv = torch.nn.Conv1d(in_channels, |
| | in_channels, |
| | kernel_size=3, |
| | stride=1, |
| | padding=1) |
| |
|
| | def forward(self, x): |
| | x = torch.nn.functional.interpolate(x, scale_factor=2.0, mode="nearest") |
| | if self.with_conv: |
| | x = self.conv(x) |
| | return x |
| |
|
| |
|
| | class Downsample(nn.Module): |
| | def __init__(self, in_channels, with_conv): |
| | super().__init__() |
| | self.with_conv = with_conv |
| | if self.with_conv: |
| | |
| | self.conv = torch.nn.Conv1d(in_channels, |
| | in_channels, |
| | kernel_size=4, |
| | stride=2, |
| | padding=1) |
| |
|
| | def forward(self, x): |
| | if self.with_conv: |
| | x = self.conv(x) |
| | else: |
| | x = torch.nn.functional.avg_pool1d(x, kernel_size=2, stride=2) |
| | return x |
| |
|
| |
|
| | class ResnetBlock(nn.Module): |
| | def __init__(self, *, in_channels, out_channels=None, conv_shortcut=False, |
| | temb_channels=512): |
| | super().__init__() |
| | self.in_channels = in_channels |
| | out_channels = in_channels if out_channels is None else out_channels |
| | self.out_channels = out_channels |
| | self.use_conv_shortcut = conv_shortcut |
| |
|
| | self.norm1 = Normalize(in_channels) |
| | self.conv1 = torch.nn.Conv1d(in_channels, |
| | out_channels, |
| | kernel_size=3, |
| | stride=1, |
| | padding=1) |
| | if temb_channels > 0: |
| | self.temb_proj = torch.nn.Linear(temb_channels, |
| | out_channels) |
| | self.norm2 = Normalize(out_channels) |
| | self.conv2 = torch.nn.Conv1d(out_channels, |
| | out_channels, |
| | kernel_size=3, |
| | stride=1, |
| | padding=1) |
| | if self.in_channels != self.out_channels: |
| | if self.use_conv_shortcut: |
| | self.conv_shortcut = torch.nn.Conv1d(in_channels, |
| | out_channels, |
| | kernel_size=3, |
| | stride=1, |
| | padding=1) |
| | else: |
| | self.nin_shortcut = torch.nn.Conv1d(in_channels, |
| | out_channels, |
| | kernel_size=1, |
| | stride=1, |
| | padding=0) |
| |
|
| | def forward(self, x, _, x_mask): |
| | x = x * x_mask |
| | h = x |
| | h = self.norm1(h) * x_mask |
| | h = nonlinearity(h) * x_mask |
| | h = self.conv1(h) * x_mask |
| |
|
| | h = self.norm2(h) * x_mask |
| | h = nonlinearity(h) * x_mask |
| | h = self.conv2(h) * x_mask |
| |
|
| | if self.in_channels != self.out_channels: |
| | if self.use_conv_shortcut: |
| | x = self.conv_shortcut(x) * x_mask |
| | else: |
| | x = self.nin_shortcut(x) * x_mask |
| |
|
| | return (x + h) * x_mask |
| |
|
| |
|
| | class AttnBlock(nn.Module): |
| | def __init__(self, in_channels): |
| | super().__init__() |
| | self.in_channels = in_channels |
| |
|
| | self.norm = Normalize(in_channels) |
| | self.q = torch.nn.Conv1d(in_channels, |
| | in_channels, |
| | kernel_size=1, |
| | stride=1, |
| | padding=0) |
| | self.k = torch.nn.Conv1d(in_channels, |
| | in_channels, |
| | kernel_size=1, |
| | stride=1, |
| | padding=0) |
| | self.v = torch.nn.Conv1d(in_channels, |
| | in_channels, |
| | kernel_size=1, |
| | stride=1, |
| | padding=0) |
| | self.proj_out = torch.nn.Conv1d(in_channels, |
| | in_channels, |
| | kernel_size=1, |
| | stride=1, |
| | padding=0) |
| |
|
| | def forward(self, x, x_mask): |
| | h_ = x * x_mask |
| | h_ = self.norm(h_) * x_mask |
| | q = self.q(h_) * x_mask |
| | k = self.k(h_) * x_mask |
| | v = self.v(h_) * x_mask |
| |
|
| | |
| | b, c, h = q.shape |
| | w = 1 |
| | q = q.reshape(b, c, h * w) |
| | q = q.permute(0, 2, 1) |
| | k = k.reshape(b, c, h * w) |
| | w_ = torch.bmm(q, k) |
| | w_ = w_ * (int(c) ** (-0.5)) |
| | w_ = w_ + ((1 - x_mask) * -1e8) + ((1 - x_mask) * -1e8).transpose(1, 2) |
| | w_ = torch.nn.functional.softmax(w_, dim=2) |
| |
|
| | |
| | v = v.reshape(b, c, h * w) |
| | w_ = w_.permute(0, 2, 1) |
| | h_ = torch.bmm(v, w_) |
| | h_ = h_.reshape(b, c, h) |
| |
|
| | h_ = self.proj_out(h_) * x_mask |
| |
|
| | return (x + h_) * x_mask |
| |
|
| |
|
| | class Encoder(nn.Module): |
| | def __init__(self, *, ch, out_ch, ch_mult=(1, 2, 4, 8), num_res_blocks, |
| | resamp_with_conv=False, in_channels): |
| | super().__init__() |
| | self.ch = ch |
| | self.temb_ch = 0 |
| | self.num_resolutions = len(ch_mult) |
| | self.num_res_blocks = num_res_blocks |
| | self.in_channels = in_channels |
| |
|
| | |
| | self.conv_in = torch.nn.Conv1d(in_channels, |
| | self.ch, |
| | kernel_size=3, |
| | stride=1, |
| | padding=1) |
| |
|
| | in_ch_mult = (1,) + tuple(ch_mult) |
| | self.down = nn.ModuleList() |
| | for i_level in range(self.num_resolutions): |
| | block = nn.ModuleList() |
| | attn = nn.ModuleList() |
| | block_in = ch * in_ch_mult[i_level] |
| | block_out = ch * ch_mult[i_level] |
| | for i_block in range(self.num_res_blocks): |
| | block.append(ResnetBlock(in_channels=block_in, |
| | out_channels=block_out, |
| | temb_channels=self.temb_ch)) |
| | block_in = block_out |
| | if i_level == self.num_resolutions - 1: |
| | attn.append(AttnBlock(block_in)) |
| | down = nn.Module() |
| | down.block = block |
| | down.attn = attn |
| | if i_level != self.num_resolutions - 1: |
| | down.downsample = Downsample(block_in, resamp_with_conv) |
| | self.down.append(down) |
| |
|
| | |
| | self.mid = nn.Module() |
| | self.mid.block_1 = ResnetBlock(in_channels=block_in, |
| | out_channels=block_in, |
| | temb_channels=self.temb_ch) |
| | self.mid.attn_1 = AttnBlock(block_in) |
| | self.mid.block_2 = ResnetBlock(in_channels=block_in, |
| | out_channels=block_in, |
| | temb_channels=self.temb_ch) |
| |
|
| | |
| | self.norm_out = Normalize(block_in) |
| | self.conv_out = torch.nn.Conv1d(block_in, |
| | out_ch, |
| | kernel_size=3, |
| | stride=1, |
| | padding=1) |
| |
|
| | def forward(self, x, x_mask): |
| | if x_mask is None: |
| | x_mask = torch.ones_like(x_mask[:, :, :1]) |
| | x = x.permute(0, 2, 1) |
| | x_mask = x_mask.permute(0, 2, 1) |
| |
|
| | temb = None |
| | |
| | hs = [self.conv_in(x) * x_mask] |
| | for i_level in range(self.num_resolutions): |
| | x_mask_ = x_mask[:, :, ::2 ** i_level] |
| | for i_block in range(self.num_res_blocks): |
| | h = self.down[i_level].block[i_block](hs[-1], temb, x_mask_) * x_mask_ |
| | if len(self.down[i_level].attn) > 0: |
| | h = self.down[i_level].attn[i_block](h, x_mask_) * x_mask_ |
| | hs.append(h) |
| | if i_level != self.num_resolutions - 1: |
| | hs.append(self.down[i_level].downsample(hs[-1]) * x_mask_[:, :, ::2]) |
| |
|
| | x_mask_ = x_mask[:, :, ::2 ** (self.num_resolutions - 1)] |
| | |
| | h = hs[-1] * x_mask_ |
| | h = self.mid.block_1(h, temb, x_mask_) * x_mask_ |
| | h = self.mid.attn_1(h, x_mask_) * x_mask_ |
| | h = self.mid.block_2(h, temb, x_mask_) * x_mask_ |
| |
|
| | |
| | h = self.norm_out(h) * x_mask_ |
| | h = nonlinearity(h) * x_mask_ |
| | h = self.conv_out(h) * x_mask_ |
| | h = h.permute(0, 2, 1) |
| | return h |
| |
|
| |
|
| | class Decoder(nn.Module): |
| | def __init__(self, *, ch, out_ch, ch_mult=(1, 2, 4, 8), num_res_blocks, |
| | resamp_with_conv=True, in_channels, give_pre_end=False): |
| | super().__init__() |
| | self.ch = ch |
| | self.temb_ch = 0 |
| | self.num_resolutions = len(ch_mult) |
| | self.num_res_blocks = num_res_blocks |
| | self.in_channels = in_channels |
| | self.give_pre_end = give_pre_end |
| |
|
| | |
| | block_in = ch * ch_mult[self.num_resolutions - 1] |
| |
|
| | |
| | self.conv_in = torch.nn.Conv1d(in_channels, |
| | block_in, |
| | kernel_size=3, |
| | stride=1, |
| | padding=1) |
| |
|
| | |
| | self.mid = nn.Module() |
| | self.mid.block_1 = ResnetBlock(in_channels=block_in, |
| | out_channels=block_in, |
| | temb_channels=self.temb_ch) |
| | self.mid.attn_1 = AttnBlock(block_in) |
| | self.mid.block_2 = ResnetBlock(in_channels=block_in, |
| | out_channels=block_in, |
| | temb_channels=self.temb_ch) |
| |
|
| | |
| | self.up = nn.ModuleList() |
| | for i_level in reversed(range(self.num_resolutions)): |
| | block = nn.ModuleList() |
| | attn = nn.ModuleList() |
| | block_out = ch * ch_mult[i_level] |
| | for i_block in range(self.num_res_blocks + 1): |
| | block.append(ResnetBlock(in_channels=block_in, |
| | out_channels=block_out, |
| | temb_channels=self.temb_ch)) |
| | block_in = block_out |
| | if i_level == self.num_resolutions - 1: |
| | attn.append(AttnBlock(block_in)) |
| | up = nn.Module() |
| | up.block = block |
| | up.attn = attn |
| | if i_level != 0: |
| | up.upsample = Upsample(block_in, resamp_with_conv) |
| | self.up.insert(0, up) |
| |
|
| | |
| | self.norm_out = Normalize(block_in) |
| | self.conv_out = torch.nn.Conv1d(block_in, |
| | out_ch, |
| | kernel_size=3, |
| | stride=1, |
| | padding=1) |
| |
|
| | def forward(self, z, x_mask): |
| | if x_mask is None: |
| | x_mask = torch.ones_like(z[:, :, :1]).repeat(1, 8, 1) |
| | z = z.permute(0, 2, 1) |
| | x_mask = x_mask.permute(0, 2, 1) |
| |
|
| | |
| | temb = None |
| |
|
| | |
| | h = self.conv_in(z) |
| |
|
| | |
| | i_level = self.num_resolutions - 1 |
| | x_mask_ = x_mask[:, :, ::2 ** i_level] |
| | h = self.mid.block_1(h, temb, x_mask_) |
| | h = self.mid.attn_1(h, x_mask_) |
| | h = self.mid.block_2(h, temb, x_mask_) |
| |
|
| | |
| | for i_level in reversed(range(self.num_resolutions)): |
| | x_mask_ = x_mask[:, :, ::2 ** i_level] |
| | for i_block in range(self.num_res_blocks + 1): |
| | h = self.up[i_level].block[i_block](h, temb, x_mask_) |
| | if len(self.up[i_level].attn) > 0: |
| | h = self.up[i_level].attn[i_block](h, x_mask_) |
| | if i_level != 0: |
| | h = self.up[i_level].upsample(h) |
| |
|
| | |
| | if self.give_pre_end: |
| | return h |
| |
|
| | h = self.norm_out(h) |
| | h = nonlinearity(h) |
| | h = self.conv_out(h) * x_mask |
| | h = h.permute(0, 2, 1) |
| | return h |
| |
|