|
|
|
@@ -59,11 +59,14 @@ class MonotonicLinear(nn.Module):
|
|
|
|
|
|
|
|
|
|
# log(snr) that approximates the original linear schedule
|
|
|
|
|
|
|
|
|
|
def beta_linear_log_snr(t):
|
|
|
|
|
return -torch.log(expm1(1e-4 + 10 * (t ** 2)))
|
|
|
|
|
def log(t, eps = 1e-20):
|
|
|
|
|
return torch.log(t.clamp(min = eps))
|
|
|
|
|
|
|
|
|
|
def alpha_cosine_log_snr(t):
|
|
|
|
|
raise NotImplementedError
|
|
|
|
|
def beta_linear_log_snr(t):
|
|
|
|
|
return -log(expm1(1e-4 + 10 * (t ** 2)))
|
|
|
|
|
|
|
|
|
|
def alpha_cosine_log_snr(t, s = 0.008):
|
|
|
|
|
return -log((torch.cos((t + s) / (1 + s) * torch.pi * 0.5) ** -2) - 1)
|
|
|
|
|
|
|
|
|
|
class learned_noise_schedule(nn.Module):
|
|
|
|
|
""" described in section H and then I.2 of the supplementary material for variational ddpm paper """
|
|
|
|
@@ -73,7 +76,8 @@ class learned_noise_schedule(nn.Module):
|
|
|
|
|
*,
|
|
|
|
|
log_snr_max,
|
|
|
|
|
log_snr_min,
|
|
|
|
|
hidden_dim = 1024
|
|
|
|
|
hidden_dim = 1024,
|
|
|
|
|
frac_gradient = 1.
|
|
|
|
|
):
|
|
|
|
|
super().__init__()
|
|
|
|
|
self.slope = log_snr_min - log_snr_max
|
|
|
|
@@ -90,7 +94,10 @@ class learned_noise_schedule(nn.Module):
|
|
|
|
|
Rearrange('... 1 -> ...'),
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
self.frac_gradient = frac_gradient
|
|
|
|
|
|
|
|
|
|
def forward(self, x):
|
|
|
|
|
frac_gradient = self.frac_gradient
|
|
|
|
|
device = x.device
|
|
|
|
|
|
|
|
|
|
out_zero = self.net(torch.zeros_like(x))
|
|
|
|
@@ -98,8 +105,8 @@ class learned_noise_schedule(nn.Module):
|
|
|
|
|
|
|
|
|
|
x = self.net(x)
|
|
|
|
|
|
|
|
|
|
normalized = self.slope * ((x - out_zero) / (out_one - out_zero)) + self.intercept
|
|
|
|
|
return normalized
|
|
|
|
|
normed = self.slope * ((x - out_zero) / (out_one - out_zero)) + self.intercept
|
|
|
|
|
return normed * frac_gradient + normed.detach() * (1 - frac_gradient)
|
|
|
|
|
|
|
|
|
|
class ContinuousTimeGaussianDiffusion(nn.Module):
|
|
|
|
|
def __init__(
|
|
|
|
@@ -111,7 +118,9 @@ class ContinuousTimeGaussianDiffusion(nn.Module):
|
|
|
|
|
loss_type = 'l1',
|
|
|
|
|
noise_schedule = 'linear',
|
|
|
|
|
num_sample_steps = 500,
|
|
|
|
|
clip_sample_after_noise = False
|
|
|
|
|
clip_sample_denoised = True,
|
|
|
|
|
learned_schedule_net_hidden_dim = 1024,
|
|
|
|
|
learned_noise_schedule_frac_gradient = 1. # between 0 and 1, determines what percentage of gradients go back, so one can update the learned noise schedule more slowly
|
|
|
|
|
):
|
|
|
|
|
super().__init__()
|
|
|
|
|
assert not denoise_fn.sinusoidal_cond_mlp
|
|
|
|
@@ -129,12 +138,16 @@ class ContinuousTimeGaussianDiffusion(nn.Module):
|
|
|
|
|
|
|
|
|
|
if noise_schedule == 'linear':
|
|
|
|
|
self.log_snr = beta_linear_log_snr
|
|
|
|
|
elif noise_schedule == 'cosine':
|
|
|
|
|
self.log_snr = alpha_cosine_log_snr
|
|
|
|
|
elif noise_schedule == 'learned':
|
|
|
|
|
log_snr_max, log_snr_min = [beta_linear_log_snr(torch.tensor([time])).item() for time in (0., 1.)]
|
|
|
|
|
|
|
|
|
|
self.log_snr = learned_noise_schedule(
|
|
|
|
|
log_snr_max = log_snr_max,
|
|
|
|
|
log_snr_min = log_snr_min
|
|
|
|
|
log_snr_min = log_snr_min,
|
|
|
|
|
hidden_dim = learned_schedule_net_hidden_dim,
|
|
|
|
|
frac_gradient = learned_noise_schedule_frac_gradient
|
|
|
|
|
)
|
|
|
|
|
else:
|
|
|
|
|
raise ValueError(f'unknown noise schedule {noise_schedule}')
|
|
|
|
@@ -142,10 +155,7 @@ class ContinuousTimeGaussianDiffusion(nn.Module):
|
|
|
|
|
# sampling
|
|
|
|
|
|
|
|
|
|
self.num_sample_steps = num_sample_steps
|
|
|
|
|
|
|
|
|
|
# clipping related hyperparameters
|
|
|
|
|
|
|
|
|
|
self.clip_sample_after_noise = clip_sample_after_noise
|
|
|
|
|
self.clip_sample_denoised = clip_sample_denoised
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def device(self):
|
|
|
|
@@ -164,9 +174,6 @@ class ContinuousTimeGaussianDiffusion(nn.Module):
|
|
|
|
|
# reviewer found an error in the equation in the paper (missing sigma)
|
|
|
|
|
# following - https://openreview.net/forum?id=2LdBqxc1Yv¬eId=rIQgH0zKsRt
|
|
|
|
|
|
|
|
|
|
# todo - derive x_start from the posterior mean and do dynamic thresholding
|
|
|
|
|
# assumed that is what is going on in Imagen
|
|
|
|
|
|
|
|
|
|
log_snr = self.log_snr(time)
|
|
|
|
|
log_snr_next = self.log_snr(time_next)
|
|
|
|
|
c = -expm1(log_snr - log_snr_next)
|
|
|
|
@@ -174,10 +181,21 @@ class ContinuousTimeGaussianDiffusion(nn.Module):
|
|
|
|
|
squared_alpha, squared_alpha_next = log_snr.sigmoid(), log_snr_next.sigmoid()
|
|
|
|
|
squared_sigma, squared_sigma_next = (-log_snr).sigmoid(), (-log_snr_next).sigmoid()
|
|
|
|
|
|
|
|
|
|
alpha, sigma, alpha_next = map(sqrt, (squared_alpha, squared_sigma, squared_alpha_next))
|
|
|
|
|
|
|
|
|
|
batch_log_snr = repeat(log_snr, ' -> b', b = x.shape[0])
|
|
|
|
|
pred_noise = self.denoise_fn(x, batch_log_snr)
|
|
|
|
|
|
|
|
|
|
model_mean = sqrt(squared_alpha_next / squared_alpha) * (x - c * sqrt(squared_sigma) * pred_noise)
|
|
|
|
|
if self.clip_sample_denoised:
|
|
|
|
|
x_start = (x - sigma * pred_noise) / alpha
|
|
|
|
|
|
|
|
|
|
# in Imagen, this was changed to dynamic thresholding
|
|
|
|
|
x_start.clamp_(-1., 1.)
|
|
|
|
|
|
|
|
|
|
model_mean = alpha_next * (x * (1 - c) / alpha + c * x_start)
|
|
|
|
|
else:
|
|
|
|
|
model_mean = alpha_next / alpha * (x - c * sigma * pred_noise)
|
|
|
|
|
|
|
|
|
|
posterior_variance = squared_sigma_next * c
|
|
|
|
|
|
|
|
|
|
return model_mean, posterior_variance
|
|
|
|
@@ -208,12 +226,9 @@ class ContinuousTimeGaussianDiffusion(nn.Module):
|
|
|
|
|
times_next = steps[i + 1]
|
|
|
|
|
img = self.p_sample(img, times, times_next)
|
|
|
|
|
|
|
|
|
|
if self.clip_sample_after_noise:
|
|
|
|
|
# clip after noise is added. perhaps this is sufficient for Imagen dynamic thresholding?
|
|
|
|
|
img.clamp_(-1., 1.)
|
|
|
|
|
|
|
|
|
|
img.clamp_(-1., 1.)
|
|
|
|
|
img = unnormalize_to_zero_to_one(img)
|
|
|
|
|
return img.clamp(0., 1.)
|
|
|
|
|
return img
|
|
|
|
|
|
|
|
|
|
@torch.no_grad()
|
|
|
|
|
def sample(self, batch_size = 16):
|
|
|
|
|