mirror of
https://github.com/wassname/denoising-diffusion-pytorch.git
synced 2026-09-10 12:01:08 +08:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Phil Wang
|
b50c9f6f13 |
@@ -1,4 +1,4 @@
|
||||
<img src="./images/denoising-diffusion.png" width="500px"></img>
|
||||
<img src="./denoising-diffusion.png" width="500px"></img>
|
||||
|
||||
## Denoising Diffusion Probabilistic Model, in Pytorch
|
||||
|
||||
@@ -6,15 +6,7 @@ Implementation of <a href="https://arxiv.org/abs/2006.11239">Denoising Diffusion
|
||||
|
||||
This implementation was transcribed from the official Tensorflow version <a href="https://github.com/hojonathanho/diffusion">here</a>
|
||||
|
||||
Youtube AI Educators - <a href="https://www.youtube.com/watch?v=W-O7AZNzbzQ">Yannic Kilcher</a> | <a href="https://www.youtube.com/watch?v=344w5h24-h8">AI Coffeebreak with Letitia</a> | <a href="https://www.youtube.com/watch?v=HoKDTa5jHvg">Outlier</a>
|
||||
|
||||
<a href="https://github.com/yiyixuxu/denoising-diffusion-flax">Flax implementation</a> from <a href="https://github.com/yiyixuxu">YiYi Xu</a>
|
||||
|
||||
<a href="https://huggingface.co/blog/annotated-diffusion">Annotated code</a> by Research Scientists / Engineers from <a href="https://huggingface.co/">🤗 Huggingface</a>
|
||||
|
||||
Update: Turns out none of the technicalities really matters at all | <a href="https://arxiv.org/abs/2208.09392">"Cold Diffusion" paper</a>
|
||||
|
||||
<img src="./images/sample.png" width="500px"><img>
|
||||
<img src="./sample.png" width="500px"><img>
|
||||
|
||||
[](https://badge.fury.io/py/denoising-diffusion-pytorch)
|
||||
|
||||
@@ -64,16 +56,15 @@ model = Unet(
|
||||
diffusion = GaussianDiffusion(
|
||||
model,
|
||||
image_size = 128,
|
||||
timesteps = 1000, # number of steps
|
||||
sampling_timesteps = 250, # number of sampling timesteps (using ddim for faster inference [see citation for ddim paper])
|
||||
loss_type = 'l1' # L1 or L2
|
||||
timesteps = 1000, # number of steps
|
||||
loss_type = 'l1' # L1 or L2
|
||||
).cuda()
|
||||
|
||||
trainer = Trainer(
|
||||
diffusion,
|
||||
'path/to/your/images',
|
||||
train_batch_size = 32,
|
||||
train_lr = 8e-5,
|
||||
train_lr = 1e-4,
|
||||
train_num_steps = 700000, # total training steps
|
||||
gradient_accumulate_every = 2, # gradient accumulation steps
|
||||
ema_decay = 0.995, # exponential moving average decay
|
||||
@@ -85,53 +76,6 @@ trainer.train()
|
||||
|
||||
Samples and model checkpoints will be logged to `./results` periodically
|
||||
|
||||
## Multi-GPU Training
|
||||
|
||||
The `Trainer` class is now equipped with <a href="https://huggingface.co/docs/accelerate/accelerator">🤗 Accelerator</a>. You can easily do multi-gpu training in two steps using their `accelerate` CLI
|
||||
|
||||
At the project root directory, where the training script is, run
|
||||
|
||||
```python
|
||||
$ accelerate config
|
||||
```
|
||||
|
||||
Then, in the same directory
|
||||
|
||||
```python
|
||||
$ accelerate launch train.py
|
||||
```
|
||||
|
||||
## Miscellaenous
|
||||
|
||||
By popular request, a 1D Unet + Gaussian Diffusion implementation. You will have to do the training code yourself
|
||||
|
||||
```python
|
||||
import torch
|
||||
from denoising_diffusion_pytorch import Unet1D, GaussianDiffusion1D
|
||||
|
||||
model = Unet1D(
|
||||
dim = 64,
|
||||
dim_mults = (1, 2, 4, 8),
|
||||
channels = 32
|
||||
)
|
||||
|
||||
diffusion = GaussianDiffusion1D(
|
||||
model,
|
||||
seq_length = 128,
|
||||
timesteps = 1000,
|
||||
objective = 'pred_v'
|
||||
)
|
||||
|
||||
training_seq = torch.randn(8, 32, 128) # features are normalized from 0 to 1
|
||||
loss = diffusion(training_seq)
|
||||
loss.backward()
|
||||
|
||||
# after a lot of training
|
||||
|
||||
sampled_seq = diffusion.sample(batch_size = 4)
|
||||
sampled_seq.shape # (4, 32, 128)
|
||||
```
|
||||
|
||||
## Citations
|
||||
|
||||
```bibtex
|
||||
@@ -175,64 +119,3 @@ sampled_seq.shape # (4, 32, 128)
|
||||
url = {https://openreview.net/forum?id=2LdBqxc1Yv}
|
||||
}
|
||||
```
|
||||
|
||||
```bibtex
|
||||
@article{Choi2022PerceptionPT,
|
||||
title = {Perception Prioritized Training of Diffusion Models},
|
||||
author = {Jooyoung Choi and Jungbeom Lee and Chaehun Shin and Sungwon Kim and Hyunwoo J. Kim and Sung-Hoon Yoon},
|
||||
journal = {ArXiv},
|
||||
year = {2022},
|
||||
volume = {abs/2204.00227}
|
||||
}
|
||||
```
|
||||
|
||||
```bibtex
|
||||
@article{Karras2022ElucidatingTD,
|
||||
title = {Elucidating the Design Space of Diffusion-Based Generative Models},
|
||||
author = {Tero Karras and Miika Aittala and Timo Aila and Samuli Laine},
|
||||
journal = {ArXiv},
|
||||
year = {2022},
|
||||
volume = {abs/2206.00364}
|
||||
}
|
||||
```
|
||||
|
||||
```bibtex
|
||||
@article{Song2021DenoisingDI,
|
||||
title = {Denoising Diffusion Implicit Models},
|
||||
author = {Jiaming Song and Chenlin Meng and Stefano Ermon},
|
||||
journal = {ArXiv},
|
||||
year = {2021},
|
||||
volume = {abs/2010.02502}
|
||||
}
|
||||
```
|
||||
|
||||
```bibtex
|
||||
@misc{chen2022analog,
|
||||
title = {Analog Bits: Generating Discrete Data using Diffusion Models with Self-Conditioning},
|
||||
author = {Ting Chen and Ruixiang Zhang and Geoffrey Hinton},
|
||||
year = {2022},
|
||||
eprint = {2208.04202},
|
||||
archivePrefix = {arXiv},
|
||||
primaryClass = {cs.CV}
|
||||
}
|
||||
```
|
||||
|
||||
```bibtex
|
||||
@article{Qiao2019WeightS,
|
||||
title = {Weight Standardization},
|
||||
author = {Siyuan Qiao and Huiyu Wang and Chenxi Liu and Wei Shen and Alan Loddon Yuille},
|
||||
journal = {ArXiv},
|
||||
year = {2019},
|
||||
volume = {abs/1903.10520}
|
||||
}
|
||||
```
|
||||
|
||||
```bibtex
|
||||
@article{Salimans2022ProgressiveDF,
|
||||
title = {Progressive Distillation for Fast Sampling of Diffusion Models},
|
||||
author = {Tim Salimans and Jonathan Ho},
|
||||
journal = {ArXiv},
|
||||
year = {2022},
|
||||
volume = {abs/2202.00512}
|
||||
}
|
||||
```
|
||||
|
||||
|
Before Width: | Height: | Size: 40 KiB After Width: | Height: | Size: 40 KiB |
@@ -3,8 +3,3 @@ from denoising_diffusion_pytorch.denoising_diffusion_pytorch import GaussianDiff
|
||||
from denoising_diffusion_pytorch.learned_gaussian_diffusion import LearnedGaussianDiffusion
|
||||
from denoising_diffusion_pytorch.continuous_time_gaussian_diffusion import ContinuousTimeGaussianDiffusion
|
||||
from denoising_diffusion_pytorch.weighted_objective_gaussian_diffusion import WeightedObjectiveGaussianDiffusion
|
||||
from denoising_diffusion_pytorch.elucidated_diffusion import ElucidatedDiffusion
|
||||
from denoising_diffusion_pytorch.v_param_continuous_time_gaussian_diffusion import VParamContinuousTimeGaussianDiffusion
|
||||
|
||||
from denoising_diffusion_pytorch.denoising_diffusion_pytorch_1d import GaussianDiffusion1D, Unet1D
|
||||
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import math
|
||||
import torch
|
||||
from torch import sqrt
|
||||
from torch import nn, einsum
|
||||
@@ -6,8 +5,7 @@ import torch.nn.functional as F
|
||||
from torch.special import expm1
|
||||
|
||||
from tqdm import tqdm
|
||||
from einops import rearrange, repeat, reduce
|
||||
from einops.layers.torch import Rearrange
|
||||
from einops import rearrange, repeat
|
||||
|
||||
# helpers
|
||||
|
||||
@@ -35,24 +33,6 @@ def right_pad_dims_to(x, t):
|
||||
return t
|
||||
return t.view(*t.shape, *((1,) * padding_dims))
|
||||
|
||||
# neural net helpers
|
||||
|
||||
class Residual(nn.Module):
|
||||
def __init__(self, fn):
|
||||
super().__init__()
|
||||
self.fn = fn
|
||||
|
||||
def forward(self, x):
|
||||
return x + self.fn(x)
|
||||
|
||||
class MonotonicLinear(nn.Module):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__()
|
||||
self.net = nn.Linear(*args, **kwargs)
|
||||
|
||||
def forward(self, x):
|
||||
return F.linear(x, self.net.weight.abs(), self.net.bias.abs())
|
||||
|
||||
# continuous schedules
|
||||
|
||||
# equations are taken from https://openreview.net/attachment?id=2LdBqxc1Yv&name=supplementary_material
|
||||
@@ -60,76 +40,33 @@ class MonotonicLinear(nn.Module):
|
||||
|
||||
# log(snr) that approximates the original linear schedule
|
||||
|
||||
def log(t, eps = 1e-20):
|
||||
return torch.log(t.clamp(min = eps))
|
||||
|
||||
def beta_linear_log_snr(t):
|
||||
return -log(expm1(1e-4 + 10 * (t ** 2)))
|
||||
return -torch.log(expm1(1e-4 + 10 * (t ** 2)))
|
||||
|
||||
def alpha_cosine_log_snr(t, s = 0.008):
|
||||
return -log((torch.cos((t + s) / (1 + s) * math.pi * 0.5) ** -2) - 1, eps = 1e-5)
|
||||
def alpha_cosine_log_snr(t):
|
||||
raise NotImplementedError
|
||||
|
||||
class learned_noise_schedule(nn.Module):
|
||||
""" described in section H and then I.2 of the supplementary material for variational ddpm paper """
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
log_snr_max,
|
||||
log_snr_min,
|
||||
hidden_dim = 1024,
|
||||
frac_gradient = 1.
|
||||
):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.slope = log_snr_min - log_snr_max
|
||||
self.intercept = log_snr_max
|
||||
|
||||
self.net = nn.Sequential(
|
||||
Rearrange('... -> ... 1'),
|
||||
MonotonicLinear(1, 1),
|
||||
Residual(nn.Sequential(
|
||||
MonotonicLinear(1, hidden_dim),
|
||||
nn.Sigmoid(),
|
||||
MonotonicLinear(hidden_dim, 1)
|
||||
)),
|
||||
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))
|
||||
out_one = self.net(torch.ones_like(x))
|
||||
|
||||
x = self.net(x)
|
||||
|
||||
normed = self.slope * ((x - out_zero) / (out_one - out_zero)) + self.intercept
|
||||
return normed * frac_gradient + normed.detach() * (1 - frac_gradient)
|
||||
raise NotImplementedError
|
||||
# learned noise schedule, using learned monotonic MLP (weights kept positive) in the paper
|
||||
|
||||
class ContinuousTimeGaussianDiffusion(nn.Module):
|
||||
def __init__(
|
||||
self,
|
||||
model,
|
||||
denoise_fn,
|
||||
*,
|
||||
image_size,
|
||||
channels = 3,
|
||||
cond_scale = 500,
|
||||
loss_type = 'l1',
|
||||
noise_schedule = 'linear',
|
||||
num_sample_steps = 500,
|
||||
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
|
||||
p2_loss_weight_gamma = 0., # p2 loss weight, from https://arxiv.org/abs/2204.00227 - 0 is equivalent to weight of 1 across time
|
||||
p2_loss_weight_k = 1
|
||||
num_sample_steps = 500
|
||||
):
|
||||
super().__init__()
|
||||
assert model.random_or_learned_sinusoidal_cond
|
||||
assert not model.self_condition, 'not supported yet'
|
||||
|
||||
self.model = model
|
||||
self.denoise_fn = denoise_fn
|
||||
|
||||
# image dimensions
|
||||
|
||||
@@ -138,40 +75,21 @@ class ContinuousTimeGaussianDiffusion(nn.Module):
|
||||
|
||||
# continuous noise schedule related stuff
|
||||
|
||||
self.cond_scale = cond_scale # the log(snr) will be scaled by this value
|
||||
self.loss_type = loss_type
|
||||
|
||||
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,
|
||||
hidden_dim = learned_schedule_net_hidden_dim,
|
||||
frac_gradient = learned_noise_schedule_frac_gradient
|
||||
)
|
||||
else:
|
||||
raise ValueError(f'unknown noise schedule {noise_schedule}')
|
||||
|
||||
# sampling
|
||||
|
||||
self.num_sample_steps = num_sample_steps
|
||||
self.clip_sample_denoised = clip_sample_denoised
|
||||
|
||||
# p2 loss weight
|
||||
# proposed https://arxiv.org/abs/2204.00227
|
||||
|
||||
assert p2_loss_weight_gamma <= 2, 'in paper, they noticed any gamma greater than 2 is harmful'
|
||||
|
||||
self.p2_loss_weight_gamma = p2_loss_weight_gamma # recommended to be 0.5 or 1
|
||||
self.p2_loss_weight_k = p2_loss_weight_k
|
||||
|
||||
@property
|
||||
def device(self):
|
||||
return next(self.model.parameters()).device
|
||||
return next(self.denoise_fn.parameters()).device
|
||||
|
||||
@property
|
||||
def loss_fn(self):
|
||||
@@ -186,6 +104,14 @@ 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
|
||||
|
||||
batch = x.shape[0]
|
||||
batch_time = repeat(time, ' -> b', b = batch)
|
||||
|
||||
pred_noise = self.denoise_fn(x, batch_time * self.cond_scale)
|
||||
|
||||
log_snr = self.log_snr(time)
|
||||
log_snr_next = self.log_snr(time_next)
|
||||
c = -expm1(log_snr - log_snr_next)
|
||||
@@ -193,21 +119,7 @@ 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.model(x, batch_log_snr)
|
||||
|
||||
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)
|
||||
|
||||
model_mean = sqrt(squared_alpha_next / squared_alpha) * (x - c * sqrt(squared_sigma) * pred_noise)
|
||||
posterior_variance = squared_sigma_next * c
|
||||
|
||||
return model_mean, posterior_variance
|
||||
@@ -215,16 +127,18 @@ class ContinuousTimeGaussianDiffusion(nn.Module):
|
||||
# sampling related functions
|
||||
|
||||
@torch.no_grad()
|
||||
def p_sample(self, x, time, time_next):
|
||||
def p_sample(self, x, time, time_next, eps = 2e-4):
|
||||
batch, *_, device = *x.shape, x.device
|
||||
|
||||
model_mean, model_variance = self.p_mean_variance(x = x, time = time, time_next = time_next)
|
||||
|
||||
if time_next == 0:
|
||||
return model_mean
|
||||
|
||||
noise = torch.randn_like(x)
|
||||
return model_mean + sqrt(model_variance) * noise
|
||||
|
||||
# no noise when time is below some epsilon
|
||||
# not sure how important this is
|
||||
|
||||
time = repeat(time, ' -> b', b = batch)
|
||||
nonzero_mask = (1 - (time < eps).float()).reshape(batch, *((1,) * (len(x.shape) - 1)))
|
||||
return model_mean + nonzero_mask * sqrt(model_variance) * noise
|
||||
|
||||
@torch.no_grad()
|
||||
def p_sample_loop(self, shape):
|
||||
@@ -238,7 +152,6 @@ class ContinuousTimeGaussianDiffusion(nn.Module):
|
||||
times_next = steps[i + 1]
|
||||
img = self.p_sample(img, times, times_next)
|
||||
|
||||
img.clamp_(-1., 1.)
|
||||
img = unnormalize_to_zero_to_one(img)
|
||||
return img
|
||||
|
||||
@@ -267,17 +180,9 @@ class ContinuousTimeGaussianDiffusion(nn.Module):
|
||||
noise = default(noise, lambda: torch.randn_like(x_start))
|
||||
|
||||
x, log_snr = self.q_sample(x_start = x_start, times = times, noise = noise)
|
||||
model_out = self.model(x, log_snr)
|
||||
|
||||
losses = self.loss_fn(model_out, noise, reduction = 'none')
|
||||
losses = reduce(losses, 'b ... -> b', 'mean')
|
||||
|
||||
if self.p2_loss_weight_gamma >= 0:
|
||||
# following eq 8. in https://arxiv.org/abs/2204.00227
|
||||
loss_weight = (self.p2_loss_weight_k + log_snr.exp()) ** -self.p2_loss_weight_gamma
|
||||
losses = losses * loss_weight
|
||||
|
||||
return losses.mean()
|
||||
model_out = self.denoise_fn(x, log_snr * self.cond_scale)
|
||||
return self.loss_fn(model_out, noise)
|
||||
|
||||
def forward(self, img, *args, **kwargs):
|
||||
b, c, h, w, device, img_size, = *img.shape, img.device, self.image_size
|
||||
|
||||
@@ -1,31 +1,21 @@
|
||||
import math
|
||||
import copy
|
||||
from pathlib import Path
|
||||
from random import random
|
||||
from functools import partial
|
||||
from collections import namedtuple
|
||||
from multiprocessing import cpu_count
|
||||
|
||||
import torch
|
||||
from torch import nn, einsum
|
||||
import torch.nn.functional as F
|
||||
from torch.utils.data import Dataset, DataLoader
|
||||
from inspect import isfunction
|
||||
from functools import partial
|
||||
|
||||
from torch.utils import data
|
||||
from torch.cuda.amp import autocast, GradScaler
|
||||
|
||||
from pathlib import Path
|
||||
from torch.optim import Adam
|
||||
from torchvision import transforms as T, utils
|
||||
|
||||
from einops import rearrange, reduce
|
||||
from einops.layers.torch import Rearrange
|
||||
|
||||
from torchvision import transforms, utils
|
||||
from PIL import Image
|
||||
from tqdm.auto import tqdm
|
||||
from ema_pytorch import EMA
|
||||
|
||||
from accelerate import Accelerator
|
||||
|
||||
# constants
|
||||
|
||||
ModelPrediction = namedtuple('ModelPrediction', ['pred_noise', 'pred_x_start'])
|
||||
from tqdm import tqdm
|
||||
from einops import rearrange
|
||||
|
||||
# helpers functions
|
||||
|
||||
@@ -35,19 +25,13 @@ def exists(x):
|
||||
def default(val, d):
|
||||
if exists(val):
|
||||
return val
|
||||
return d() if callable(d) else d
|
||||
|
||||
def identity(t, *args, **kwargs):
|
||||
return t
|
||||
return d() if isfunction(d) else d
|
||||
|
||||
def cycle(dl):
|
||||
while True:
|
||||
for data in dl:
|
||||
yield data
|
||||
|
||||
def has_int_squareroot(num):
|
||||
return (math.sqrt(num) ** 2) == num
|
||||
|
||||
def num_to_groups(num, divisor):
|
||||
groups = num // divisor
|
||||
remainder = num % divisor
|
||||
@@ -56,13 +40,6 @@ def num_to_groups(num, divisor):
|
||||
arr.append(remainder)
|
||||
return arr
|
||||
|
||||
def convert_image_to_fn(img_type, image):
|
||||
if image.mode != img_type:
|
||||
return image.convert(img_type)
|
||||
return image
|
||||
|
||||
# normalization functions
|
||||
|
||||
def normalize_to_neg_one_to_one(img):
|
||||
return img * 2 - 1
|
||||
|
||||
@@ -71,6 +48,21 @@ def unnormalize_to_zero_to_one(t):
|
||||
|
||||
# small helper modules
|
||||
|
||||
class EMA():
|
||||
def __init__(self, beta):
|
||||
super().__init__()
|
||||
self.beta = beta
|
||||
|
||||
def update_model_average(self, ma_model, current_model):
|
||||
for current_params, ma_params in zip(current_model.parameters(), ma_model.parameters()):
|
||||
old_weight, up_weight = ma_params.data, current_params.data
|
||||
ma_params.data = self.update_average(old_weight, up_weight)
|
||||
|
||||
def update_average(self, old, new):
|
||||
if old is None:
|
||||
return new
|
||||
return old * self.beta + (1 - self.beta) * new
|
||||
|
||||
class Residual(nn.Module):
|
||||
def __init__(self, fn):
|
||||
super().__init__()
|
||||
@@ -79,53 +71,6 @@ class Residual(nn.Module):
|
||||
def forward(self, x, *args, **kwargs):
|
||||
return self.fn(x, *args, **kwargs) + x
|
||||
|
||||
def Upsample(dim, dim_out = None):
|
||||
return nn.Sequential(
|
||||
nn.Upsample(scale_factor = 2, mode = 'nearest'),
|
||||
nn.Conv2d(dim, default(dim_out, dim), 3, padding = 1)
|
||||
)
|
||||
|
||||
def Downsample(dim, dim_out = None):
|
||||
return nn.Conv2d(dim, default(dim_out, dim), 4, 2, 1)
|
||||
|
||||
class WeightStandardizedConv2d(nn.Conv2d):
|
||||
"""
|
||||
https://arxiv.org/abs/1903.10520
|
||||
weight standardization purportedly works synergistically with group normalization
|
||||
"""
|
||||
def forward(self, x):
|
||||
eps = 1e-5 if x.dtype == torch.float32 else 1e-3
|
||||
|
||||
weight = self.weight
|
||||
mean = reduce(weight, 'o ... -> o 1 1 1', 'mean')
|
||||
var = reduce(weight, 'o ... -> o 1 1 1', partial(torch.var, unbiased = False))
|
||||
normalized_weight = (weight - mean) * (var + eps).rsqrt()
|
||||
|
||||
return F.conv2d(x, normalized_weight, self.bias, self.stride, self.padding, self.dilation, self.groups)
|
||||
|
||||
class LayerNorm(nn.Module):
|
||||
def __init__(self, dim):
|
||||
super().__init__()
|
||||
self.g = nn.Parameter(torch.ones(1, dim, 1, 1))
|
||||
|
||||
def forward(self, x):
|
||||
eps = 1e-5 if x.dtype == torch.float32 else 1e-3
|
||||
var = torch.var(x, dim = 1, unbiased = False, keepdim = True)
|
||||
mean = torch.mean(x, dim = 1, keepdim = True)
|
||||
return (x - mean) * (var + eps).rsqrt() * self.g
|
||||
|
||||
class PreNorm(nn.Module):
|
||||
def __init__(self, dim, fn):
|
||||
super().__init__()
|
||||
self.fn = fn
|
||||
self.norm = LayerNorm(dim)
|
||||
|
||||
def forward(self, x):
|
||||
x = self.norm(x)
|
||||
return self.fn(x)
|
||||
|
||||
# sinusoidal positional embeds
|
||||
|
||||
class SinusoidalPosEmb(nn.Module):
|
||||
def __init__(self, dim):
|
||||
super().__init__()
|
||||
@@ -140,29 +85,40 @@ class SinusoidalPosEmb(nn.Module):
|
||||
emb = torch.cat((emb.sin(), emb.cos()), dim=-1)
|
||||
return emb
|
||||
|
||||
class RandomOrLearnedSinusoidalPosEmb(nn.Module):
|
||||
""" following @crowsonkb 's lead with random (learned optional) sinusoidal pos emb """
|
||||
""" https://github.com/crowsonkb/v-diffusion-jax/blob/master/diffusion/models/danbooru_128.py#L8 """
|
||||
def Upsample(dim):
|
||||
return nn.ConvTranspose2d(dim, dim, 4, 2, 1)
|
||||
|
||||
def __init__(self, dim, is_random = False):
|
||||
def Downsample(dim):
|
||||
return nn.Conv2d(dim, dim, 4, 2, 1)
|
||||
|
||||
class LayerNorm(nn.Module):
|
||||
def __init__(self, dim, eps = 1e-5):
|
||||
super().__init__()
|
||||
assert (dim % 2) == 0
|
||||
half_dim = dim // 2
|
||||
self.weights = nn.Parameter(torch.randn(half_dim), requires_grad = not is_random)
|
||||
self.eps = eps
|
||||
self.g = nn.Parameter(torch.ones(1, dim, 1, 1))
|
||||
self.b = nn.Parameter(torch.zeros(1, dim, 1, 1))
|
||||
|
||||
def forward(self, x):
|
||||
x = rearrange(x, 'b -> b 1')
|
||||
freqs = x * rearrange(self.weights, 'd -> 1 d') * 2 * math.pi
|
||||
fouriered = torch.cat((freqs.sin(), freqs.cos()), dim = -1)
|
||||
fouriered = torch.cat((x, fouriered), dim = -1)
|
||||
return fouriered
|
||||
var = torch.var(x, dim = 1, unbiased = False, keepdim = True)
|
||||
mean = torch.mean(x, dim = 1, keepdim = True)
|
||||
return (x - mean) / (var + self.eps).sqrt() * self.g + self.b
|
||||
|
||||
class PreNorm(nn.Module):
|
||||
def __init__(self, dim, fn):
|
||||
super().__init__()
|
||||
self.fn = fn
|
||||
self.norm = LayerNorm(dim)
|
||||
|
||||
def forward(self, x):
|
||||
x = self.norm(x)
|
||||
return self.fn(x)
|
||||
|
||||
# building block modules
|
||||
|
||||
class Block(nn.Module):
|
||||
def __init__(self, dim, dim_out, groups = 8):
|
||||
super().__init__()
|
||||
self.proj = WeightStandardizedConv2d(dim, dim_out, 3, padding = 1)
|
||||
self.proj = nn.Conv2d(dim, dim_out, 3, padding = 1)
|
||||
self.norm = nn.GroupNorm(groups, dim_out)
|
||||
self.act = nn.SiLU()
|
||||
|
||||
@@ -200,7 +156,6 @@ class ResnetBlock(nn.Module):
|
||||
h = self.block1(x, scale_shift = scale_shift)
|
||||
|
||||
h = self.block2(h)
|
||||
|
||||
return h + self.res_conv(x)
|
||||
|
||||
class LinearAttention(nn.Module):
|
||||
@@ -225,8 +180,6 @@ class LinearAttention(nn.Module):
|
||||
k = k.softmax(dim = -1)
|
||||
|
||||
q = q * self.scale
|
||||
v = v / (h * w)
|
||||
|
||||
context = torch.einsum('b h d n, b h e n -> b h d e', k, v)
|
||||
|
||||
out = torch.einsum('b h d e, b h d n -> b h e n', context, q)
|
||||
@@ -239,7 +192,6 @@ class Attention(nn.Module):
|
||||
self.scale = dim_head ** -0.5
|
||||
self.heads = heads
|
||||
hidden_dim = dim_head * heads
|
||||
|
||||
self.to_qkv = nn.Conv2d(dim, hidden_dim * 3, 1, bias = False)
|
||||
self.to_out = nn.Conv2d(hidden_dim, dim, 1)
|
||||
|
||||
@@ -247,13 +199,13 @@ class Attention(nn.Module):
|
||||
b, c, h, w = x.shape
|
||||
qkv = self.to_qkv(x).chunk(3, dim = 1)
|
||||
q, k, v = map(lambda t: rearrange(t, 'b (h c) x y -> b h c (x y)', h = self.heads), qkv)
|
||||
|
||||
q = q * self.scale
|
||||
|
||||
sim = einsum('b h d i, b h d j -> b h i j', q, k)
|
||||
sim = sim - sim.amax(dim = -1, keepdim = True).detach()
|
||||
attn = sim.softmax(dim = -1)
|
||||
out = einsum('b h i j, b h d j -> b h i d', attn, v)
|
||||
|
||||
out = einsum('b h i j, b h d j -> b h i d', attn, v)
|
||||
out = rearrange(out, 'b h (x y) d -> b (h d) x y', x = h, y = w)
|
||||
return self.to_out(out)
|
||||
|
||||
@@ -267,23 +219,18 @@ class Unet(nn.Module):
|
||||
out_dim = None,
|
||||
dim_mults=(1, 2, 4, 8),
|
||||
channels = 3,
|
||||
self_condition = False,
|
||||
with_time_emb = True,
|
||||
resnet_block_groups = 8,
|
||||
learned_variance = False,
|
||||
learned_sinusoidal_cond = False,
|
||||
random_fourier_features = False,
|
||||
learned_sinusoidal_dim = 16
|
||||
learned_variance = False
|
||||
):
|
||||
super().__init__()
|
||||
|
||||
# determine dimensions
|
||||
|
||||
self.channels = channels
|
||||
self.self_condition = self_condition
|
||||
input_channels = channels * (2 if self_condition else 1)
|
||||
|
||||
init_dim = default(init_dim, dim)
|
||||
self.init_conv = nn.Conv2d(input_channels, init_dim, 7, padding = 3)
|
||||
init_dim = default(init_dim, dim // 3 * 2)
|
||||
self.init_conv = nn.Conv2d(channels, init_dim, 7, padding = 3)
|
||||
|
||||
dims = [init_dim, *map(lambda m: dim * m, dim_mults)]
|
||||
in_out = list(zip(dims[:-1], dims[1:]))
|
||||
@@ -292,23 +239,17 @@ class Unet(nn.Module):
|
||||
|
||||
# time embeddings
|
||||
|
||||
time_dim = dim * 4
|
||||
|
||||
self.random_or_learned_sinusoidal_cond = learned_sinusoidal_cond or random_fourier_features
|
||||
|
||||
if self.random_or_learned_sinusoidal_cond:
|
||||
sinu_pos_emb = RandomOrLearnedSinusoidalPosEmb(learned_sinusoidal_dim, random_fourier_features)
|
||||
fourier_dim = learned_sinusoidal_dim + 1
|
||||
if with_time_emb:
|
||||
time_dim = dim * 4
|
||||
self.time_mlp = nn.Sequential(
|
||||
SinusoidalPosEmb(dim),
|
||||
nn.Linear(dim, time_dim),
|
||||
nn.GELU(),
|
||||
nn.Linear(time_dim, time_dim)
|
||||
)
|
||||
else:
|
||||
sinu_pos_emb = SinusoidalPosEmb(dim)
|
||||
fourier_dim = dim
|
||||
|
||||
self.time_mlp = nn.Sequential(
|
||||
sinu_pos_emb,
|
||||
nn.Linear(fourier_dim, time_dim),
|
||||
nn.GELU(),
|
||||
nn.Linear(time_dim, time_dim)
|
||||
)
|
||||
time_dim = None
|
||||
self.time_mlp = None
|
||||
|
||||
# layers
|
||||
|
||||
@@ -320,10 +261,10 @@ class Unet(nn.Module):
|
||||
is_last = ind >= (num_resolutions - 1)
|
||||
|
||||
self.downs.append(nn.ModuleList([
|
||||
block_klass(dim_in, dim_in, time_emb_dim = time_dim),
|
||||
block_klass(dim_in, dim_in, time_emb_dim = time_dim),
|
||||
Residual(PreNorm(dim_in, LinearAttention(dim_in))),
|
||||
Downsample(dim_in, dim_out) if not is_last else nn.Conv2d(dim_in, dim_out, 3, padding = 1)
|
||||
block_klass(dim_in, dim_out, time_emb_dim = time_dim),
|
||||
block_klass(dim_out, dim_out, time_emb_dim = time_dim),
|
||||
Residual(PreNorm(dim_out, LinearAttention(dim_out))),
|
||||
Downsample(dim_out) if not is_last else nn.Identity()
|
||||
]))
|
||||
|
||||
mid_dim = dims[-1]
|
||||
@@ -331,42 +272,36 @@ class Unet(nn.Module):
|
||||
self.mid_attn = Residual(PreNorm(mid_dim, Attention(mid_dim)))
|
||||
self.mid_block2 = block_klass(mid_dim, mid_dim, time_emb_dim = time_dim)
|
||||
|
||||
for ind, (dim_in, dim_out) in enumerate(reversed(in_out)):
|
||||
is_last = ind == (len(in_out) - 1)
|
||||
for ind, (dim_in, dim_out) in enumerate(reversed(in_out[1:])):
|
||||
is_last = ind >= (num_resolutions - 1)
|
||||
|
||||
self.ups.append(nn.ModuleList([
|
||||
block_klass(dim_out + dim_in, dim_out, time_emb_dim = time_dim),
|
||||
block_klass(dim_out + dim_in, dim_out, time_emb_dim = time_dim),
|
||||
Residual(PreNorm(dim_out, LinearAttention(dim_out))),
|
||||
Upsample(dim_out, dim_in) if not is_last else nn.Conv2d(dim_out, dim_in, 3, padding = 1)
|
||||
block_klass(dim_out * 2, dim_in, time_emb_dim = time_dim),
|
||||
block_klass(dim_in, dim_in, time_emb_dim = time_dim),
|
||||
Residual(PreNorm(dim_in, LinearAttention(dim_in))),
|
||||
Upsample(dim_in) if not is_last else nn.Identity()
|
||||
]))
|
||||
|
||||
default_out_dim = channels * (1 if not learned_variance else 2)
|
||||
self.out_dim = default(out_dim, default_out_dim)
|
||||
|
||||
self.final_res_block = block_klass(dim * 2, dim, time_emb_dim = time_dim)
|
||||
self.final_conv = nn.Conv2d(dim, self.out_dim, 1)
|
||||
|
||||
def forward(self, x, time, x_self_cond = None):
|
||||
if self.self_condition:
|
||||
x_self_cond = default(x_self_cond, lambda: torch.zeros_like(x))
|
||||
x = torch.cat((x_self_cond, x), dim = 1)
|
||||
self.final_conv = nn.Sequential(
|
||||
block_klass(dim, dim),
|
||||
nn.Conv2d(dim, self.out_dim, 1)
|
||||
)
|
||||
|
||||
def forward(self, x, time):
|
||||
x = self.init_conv(x)
|
||||
r = x.clone()
|
||||
|
||||
t = self.time_mlp(time)
|
||||
t = self.time_mlp(time) if exists(self.time_mlp) else None
|
||||
|
||||
h = []
|
||||
|
||||
for block1, block2, attn, downsample in self.downs:
|
||||
x = block1(x, t)
|
||||
h.append(x)
|
||||
|
||||
x = block2(x, t)
|
||||
x = attn(x)
|
||||
h.append(x)
|
||||
|
||||
x = downsample(x)
|
||||
|
||||
x = self.mid_block1(x, t)
|
||||
@@ -374,18 +309,12 @@ class Unet(nn.Module):
|
||||
x = self.mid_block2(x, t)
|
||||
|
||||
for block1, block2, attn, upsample in self.ups:
|
||||
x = torch.cat((x, h.pop()), dim = 1)
|
||||
x = torch.cat((x, h.pop()), dim=1)
|
||||
x = block1(x, t)
|
||||
|
||||
x = torch.cat((x, h.pop()), dim = 1)
|
||||
x = block2(x, t)
|
||||
x = attn(x)
|
||||
|
||||
x = upsample(x)
|
||||
|
||||
x = torch.cat((x, r), dim = 1)
|
||||
|
||||
x = self.final_res_block(x, t)
|
||||
return self.final_conv(x)
|
||||
|
||||
# gaussian diffusion trainer class
|
||||
@@ -408,7 +337,7 @@ def cosine_beta_schedule(timesteps, s = 0.008):
|
||||
"""
|
||||
steps = timesteps + 1
|
||||
x = torch.linspace(0, timesteps, steps, dtype = torch.float64)
|
||||
alphas_cumprod = torch.cos(((x / timesteps) + s) / (1 + s) * math.pi * 0.5) ** 2
|
||||
alphas_cumprod = torch.cos(((x / timesteps) + s) / (1 + s) * torch.pi * 0.5) ** 2
|
||||
alphas_cumprod = alphas_cumprod / alphas_cumprod[0]
|
||||
betas = 1 - (alphas_cumprod[1:] / alphas_cumprod[:-1])
|
||||
return torch.clip(betas, 0, 0.999)
|
||||
@@ -416,32 +345,23 @@ def cosine_beta_schedule(timesteps, s = 0.008):
|
||||
class GaussianDiffusion(nn.Module):
|
||||
def __init__(
|
||||
self,
|
||||
model,
|
||||
denoise_fn,
|
||||
*,
|
||||
image_size,
|
||||
channels = 3,
|
||||
timesteps = 1000,
|
||||
sampling_timesteps = None,
|
||||
loss_type = 'l1',
|
||||
objective = 'pred_noise',
|
||||
beta_schedule = 'cosine',
|
||||
p2_loss_weight_gamma = 0., # p2 loss weight, from https://arxiv.org/abs/2204.00227 - 0 is equivalent to weight of 1 across time - 1. is recommended
|
||||
p2_loss_weight_k = 1,
|
||||
ddim_sampling_eta = 1.
|
||||
beta_schedule = 'cosine'
|
||||
):
|
||||
super().__init__()
|
||||
assert not (type(self) == GaussianDiffusion and model.channels != model.out_dim)
|
||||
assert not model.random_or_learned_sinusoidal_cond
|
||||
|
||||
self.model = model
|
||||
self.channels = self.model.channels
|
||||
self.self_condition = self.model.self_condition
|
||||
assert not (type(self) == GaussianDiffusion and denoise_fn.channels != denoise_fn.out_dim)
|
||||
|
||||
self.channels = channels
|
||||
self.image_size = image_size
|
||||
|
||||
self.denoise_fn = denoise_fn
|
||||
self.objective = objective
|
||||
|
||||
assert objective in {'pred_noise', 'pred_x0', 'pred_v'}, 'objective must be either pred_noise (predict noise) or pred_x0 (predict image start) or pred_v (predict v [v-parameterization as defined in appendix D of progressive distillation paper, used in imagen-video successfully])'
|
||||
|
||||
if beta_schedule == 'linear':
|
||||
betas = linear_beta_schedule(timesteps)
|
||||
elif beta_schedule == 'cosine':
|
||||
@@ -450,21 +370,13 @@ class GaussianDiffusion(nn.Module):
|
||||
raise ValueError(f'unknown beta schedule {beta_schedule}')
|
||||
|
||||
alphas = 1. - betas
|
||||
alphas_cumprod = torch.cumprod(alphas, dim=0)
|
||||
alphas_cumprod = torch.cumprod(alphas, axis=0)
|
||||
alphas_cumprod_prev = F.pad(alphas_cumprod[:-1], (1, 0), value = 1.)
|
||||
|
||||
timesteps, = betas.shape
|
||||
self.num_timesteps = int(timesteps)
|
||||
self.loss_type = loss_type
|
||||
|
||||
# sampling related parameters
|
||||
|
||||
self.sampling_timesteps = default(sampling_timesteps, timesteps) # default num sampling timesteps to number of timesteps at training
|
||||
|
||||
assert self.sampling_timesteps <= timesteps
|
||||
self.is_ddim_sampling = self.sampling_timesteps < timesteps
|
||||
self.ddim_sampling_eta = ddim_sampling_eta
|
||||
|
||||
# helper function to register buffer from float64 to float32
|
||||
|
||||
register_buffer = lambda name, val: self.register_buffer(name, val.to(torch.float32))
|
||||
@@ -495,34 +407,12 @@ class GaussianDiffusion(nn.Module):
|
||||
register_buffer('posterior_mean_coef1', betas * torch.sqrt(alphas_cumprod_prev) / (1. - alphas_cumprod))
|
||||
register_buffer('posterior_mean_coef2', (1. - alphas_cumprod_prev) * torch.sqrt(alphas) / (1. - alphas_cumprod))
|
||||
|
||||
# calculate p2 reweighting
|
||||
|
||||
register_buffer('p2_loss_weight', (p2_loss_weight_k + alphas_cumprod / (1 - alphas_cumprod)) ** -p2_loss_weight_gamma)
|
||||
|
||||
def predict_start_from_noise(self, x_t, t, noise):
|
||||
return (
|
||||
extract(self.sqrt_recip_alphas_cumprod, t, x_t.shape) * x_t -
|
||||
extract(self.sqrt_recipm1_alphas_cumprod, t, x_t.shape) * noise
|
||||
)
|
||||
|
||||
def predict_noise_from_start(self, x_t, t, x0):
|
||||
return (
|
||||
(extract(self.sqrt_recip_alphas_cumprod, t, x_t.shape) * x_t - x0) / \
|
||||
extract(self.sqrt_recipm1_alphas_cumprod, t, x_t.shape)
|
||||
)
|
||||
|
||||
def predict_v(self, x_start, t, noise):
|
||||
return (
|
||||
extract(self.sqrt_alphas_cumprod, t, x_start.shape) * noise -
|
||||
extract(self.sqrt_one_minus_alphas_cumprod, t, x_start.shape) * x_start
|
||||
)
|
||||
|
||||
def predict_start_from_v(self, x_t, t, v):
|
||||
return (
|
||||
extract(self.sqrt_alphas_cumprod, t, x_t.shape) * x_t -
|
||||
extract(self.sqrt_one_minus_alphas_cumprod, t, x_t.shape) * v
|
||||
)
|
||||
|
||||
def q_posterior(self, x_start, x_t, t):
|
||||
posterior_mean = (
|
||||
extract(self.posterior_mean_coef1, t, x_t.shape) * x_start +
|
||||
@@ -532,103 +422,49 @@ class GaussianDiffusion(nn.Module):
|
||||
posterior_log_variance_clipped = extract(self.posterior_log_variance_clipped, t, x_t.shape)
|
||||
return posterior_mean, posterior_variance, posterior_log_variance_clipped
|
||||
|
||||
def model_predictions(self, x, t, x_self_cond = None, clip_x_start = False):
|
||||
model_output = self.model(x, t, x_self_cond)
|
||||
maybe_clip = partial(torch.clamp, min = -1., max = 1.) if clip_x_start else identity
|
||||
def p_mean_variance(self, x, t, clip_denoised: bool):
|
||||
model_output = self.denoise_fn(x, t)
|
||||
|
||||
if self.objective == 'pred_noise':
|
||||
pred_noise = model_output
|
||||
x_start = self.predict_start_from_noise(x, t, pred_noise)
|
||||
x_start = maybe_clip(x_start)
|
||||
|
||||
x_start = self.predict_start_from_noise(x, t = t, noise = model_output)
|
||||
elif self.objective == 'pred_x0':
|
||||
x_start = model_output
|
||||
x_start = maybe_clip(x_start)
|
||||
pred_noise = self.predict_noise_from_start(x, t, x_start)
|
||||
|
||||
elif self.objective == 'pred_v':
|
||||
v = model_output
|
||||
x_start = self.predict_start_from_v(x, t, v)
|
||||
x_start = maybe_clip(x_start)
|
||||
pred_noise = self.predict_noise_from_start(x, t, x_start)
|
||||
|
||||
return ModelPrediction(pred_noise, x_start)
|
||||
|
||||
def p_mean_variance(self, x, t, x_self_cond = None, clip_denoised = True):
|
||||
preds = self.model_predictions(x, t, x_self_cond)
|
||||
x_start = preds.pred_x_start
|
||||
else:
|
||||
raise ValueError(f'unknown objective {self.objective}')
|
||||
|
||||
if clip_denoised:
|
||||
x_start.clamp_(-1., 1.)
|
||||
|
||||
model_mean, posterior_variance, posterior_log_variance = self.q_posterior(x_start = x_start, x_t = x, t = t)
|
||||
return model_mean, posterior_variance, posterior_log_variance, x_start
|
||||
return model_mean, posterior_variance, posterior_log_variance
|
||||
|
||||
@torch.no_grad()
|
||||
def p_sample(self, x, t: int, x_self_cond = None, clip_denoised = True):
|
||||
def p_sample(self, x, t, clip_denoised=True):
|
||||
b, *_, device = *x.shape, x.device
|
||||
batched_times = torch.full((x.shape[0],), t, device = x.device, dtype = torch.long)
|
||||
model_mean, _, model_log_variance, x_start = self.p_mean_variance(x = x, t = batched_times, x_self_cond = x_self_cond, clip_denoised = clip_denoised)
|
||||
noise = torch.randn_like(x) if t > 0 else 0. # no noise if t == 0
|
||||
pred_img = model_mean + (0.5 * model_log_variance).exp() * noise
|
||||
return pred_img, x_start
|
||||
model_mean, _, model_log_variance = self.p_mean_variance(x=x, t=t, clip_denoised=clip_denoised)
|
||||
noise = torch.randn_like(x)
|
||||
# no noise when t == 0
|
||||
nonzero_mask = (1 - (t == 0).float()).reshape(b, *((1,) * (len(x.shape) - 1)))
|
||||
return model_mean + nonzero_mask * (0.5 * model_log_variance).exp() * noise
|
||||
|
||||
@torch.no_grad()
|
||||
def p_sample_loop(self, shape):
|
||||
batch, device = shape[0], self.betas.device
|
||||
device = self.betas.device
|
||||
|
||||
b = shape[0]
|
||||
img = torch.randn(shape, device=device)
|
||||
|
||||
x_start = None
|
||||
|
||||
for t in tqdm(reversed(range(0, self.num_timesteps)), desc = 'sampling loop time step', total = self.num_timesteps):
|
||||
self_cond = x_start if self.self_condition else None
|
||||
img, x_start = self.p_sample(img, t, self_cond)
|
||||
|
||||
img = unnormalize_to_zero_to_one(img)
|
||||
return img
|
||||
|
||||
@torch.no_grad()
|
||||
def ddim_sample(self, shape, clip_denoised = True):
|
||||
batch, device, total_timesteps, sampling_timesteps, eta, objective = shape[0], self.betas.device, self.num_timesteps, self.sampling_timesteps, self.ddim_sampling_eta, self.objective
|
||||
|
||||
times = torch.linspace(-1, total_timesteps - 1, steps=sampling_timesteps + 1) # [-1, 0, 1, 2, ..., T-1] when sampling_timesteps == total_timesteps
|
||||
times = list(reversed(times.int().tolist()))
|
||||
time_pairs = list(zip(times[:-1], times[1:])) # [(T-1, T-2), (T-2, T-3), ..., (1, 0), (0, -1)]
|
||||
|
||||
img = torch.randn(shape, device = device)
|
||||
|
||||
x_start = None
|
||||
|
||||
for time, time_next in tqdm(time_pairs, desc = 'sampling loop time step'):
|
||||
time_cond = torch.full((batch,), time, device=device, dtype=torch.long)
|
||||
self_cond = x_start if self.self_condition else None
|
||||
pred_noise, x_start, *_ = self.model_predictions(img, time_cond, self_cond, clip_x_start = clip_denoised)
|
||||
|
||||
if time_next < 0:
|
||||
img = x_start
|
||||
continue
|
||||
|
||||
alpha = self.alphas_cumprod[time]
|
||||
alpha_next = self.alphas_cumprod[time_next]
|
||||
|
||||
sigma = eta * ((1 - alpha / alpha_next) * (1 - alpha_next) / (1 - alpha)).sqrt()
|
||||
c = (1 - alpha_next - sigma ** 2).sqrt()
|
||||
|
||||
noise = torch.randn_like(img)
|
||||
|
||||
img = x_start * alpha_next.sqrt() + \
|
||||
c * pred_noise + \
|
||||
sigma * noise
|
||||
for i in tqdm(reversed(range(0, self.num_timesteps)), desc='sampling loop time step', total=self.num_timesteps):
|
||||
img = self.p_sample(img, torch.full((b,), i, device=device, dtype=torch.long))
|
||||
|
||||
img = unnormalize_to_zero_to_one(img)
|
||||
return img
|
||||
|
||||
@torch.no_grad()
|
||||
def sample(self, batch_size = 16):
|
||||
image_size, channels = self.image_size, self.channels
|
||||
sample_fn = self.p_sample_loop if not self.is_ddim_sampling else self.ddim_sample
|
||||
return sample_fn((batch_size, channels, image_size, image_size))
|
||||
image_size = self.image_size
|
||||
channels = self.channels
|
||||
return self.p_sample_loop((batch_size, channels, image_size, image_size))
|
||||
|
||||
@torch.no_grad()
|
||||
def interpolate(self, x1, x2, t = None, lam = 0.5):
|
||||
@@ -637,11 +473,11 @@ class GaussianDiffusion(nn.Module):
|
||||
|
||||
assert x1.shape == x2.shape
|
||||
|
||||
t_batched = torch.stack([torch.tensor(t, device = device)] * b)
|
||||
xt1, xt2 = map(lambda x: self.q_sample(x, t = t_batched), (x1, x2))
|
||||
t_batched = torch.stack([torch.tensor(t, device=device)] * b)
|
||||
xt1, xt2 = map(lambda x: self.q_sample(x, t=t_batched), (x1, x2))
|
||||
|
||||
img = (1 - lam) * xt1 + lam * xt2
|
||||
for i in tqdm(reversed(range(0, t)), desc = 'interpolation sample time step', total = t):
|
||||
for i in tqdm(reversed(range(0, t)), desc='interpolation sample time step', total=t):
|
||||
img = self.p_sample(img, torch.full((b,), i, device=device, dtype=torch.long))
|
||||
|
||||
return img
|
||||
@@ -667,39 +503,18 @@ class GaussianDiffusion(nn.Module):
|
||||
b, c, h, w = x_start.shape
|
||||
noise = default(noise, lambda: torch.randn_like(x_start))
|
||||
|
||||
# noise sample
|
||||
|
||||
x = self.q_sample(x_start = x_start, t = t, noise = noise)
|
||||
|
||||
# if doing self-conditioning, 50% of the time, predict x_start from current set of times
|
||||
# and condition with unet with that
|
||||
# this technique will slow down training by 25%, but seems to lower FID significantly
|
||||
|
||||
x_self_cond = None
|
||||
if self.self_condition and random() < 0.5:
|
||||
with torch.no_grad():
|
||||
x_self_cond = self.model_predictions(x, t).pred_x_start
|
||||
x_self_cond.detach_()
|
||||
|
||||
# predict and take gradient step
|
||||
|
||||
model_out = self.model(x, t, x_self_cond)
|
||||
x = self.q_sample(x_start=x_start, t=t, noise=noise)
|
||||
model_out = self.denoise_fn(x, t)
|
||||
|
||||
if self.objective == 'pred_noise':
|
||||
target = noise
|
||||
elif self.objective == 'pred_x0':
|
||||
target = x_start
|
||||
elif self.objective == 'pred_v':
|
||||
v = self.predict_v(x_start, t, noise)
|
||||
target = v
|
||||
else:
|
||||
raise ValueError(f'unknown objective {self.objective}')
|
||||
|
||||
loss = self.loss_fn(model_out, target, reduction = 'none')
|
||||
loss = reduce(loss, 'b ... -> b (...)', 'mean')
|
||||
|
||||
loss = loss * extract(self.p2_loss_weight, t, loss.shape)
|
||||
return loss.mean()
|
||||
loss = self.loss_fn(model_out, target)
|
||||
return loss
|
||||
|
||||
def forward(self, img, *args, **kwargs):
|
||||
b, c, h, w, device, img_size, = *img.shape, img.device, self.image_size
|
||||
@@ -711,28 +526,18 @@ class GaussianDiffusion(nn.Module):
|
||||
|
||||
# dataset classes
|
||||
|
||||
class Dataset(Dataset):
|
||||
def __init__(
|
||||
self,
|
||||
folder,
|
||||
image_size,
|
||||
exts = ['jpg', 'jpeg', 'png', 'tiff'],
|
||||
augment_horizontal_flip = False,
|
||||
convert_image_to = None
|
||||
):
|
||||
class Dataset(data.Dataset):
|
||||
def __init__(self, folder, image_size, exts = ['jpg', 'jpeg', 'png']):
|
||||
super().__init__()
|
||||
self.folder = folder
|
||||
self.image_size = image_size
|
||||
self.paths = [p for ext in exts for p in Path(f'{folder}').glob(f'**/*.{ext}')]
|
||||
|
||||
maybe_convert_fn = partial(convert_image_to_fn, convert_image_to) if exists(convert_image_to) else nn.Identity()
|
||||
|
||||
self.transform = T.Compose([
|
||||
T.Lambda(maybe_convert_fn),
|
||||
T.Resize(image_size),
|
||||
T.RandomHorizontalFlip() if augment_horizontal_flip else nn.Identity(),
|
||||
T.CenterCrop(image_size),
|
||||
T.ToTensor()
|
||||
self.transform = transforms.Compose([
|
||||
transforms.Resize(image_size),
|
||||
transforms.RandomHorizontalFlip(),
|
||||
transforms.CenterCrop(image_size),
|
||||
transforms.ToTensor()
|
||||
])
|
||||
|
||||
def __len__(self):
|
||||
@@ -751,148 +556,103 @@ class Trainer(object):
|
||||
diffusion_model,
|
||||
folder,
|
||||
*,
|
||||
train_batch_size = 16,
|
||||
gradient_accumulate_every = 1,
|
||||
augment_horizontal_flip = True,
|
||||
ema_decay = 0.995,
|
||||
image_size = 128,
|
||||
train_batch_size = 32,
|
||||
train_lr = 1e-4,
|
||||
train_num_steps = 100000,
|
||||
ema_update_every = 10,
|
||||
ema_decay = 0.995,
|
||||
adam_betas = (0.9, 0.99),
|
||||
save_and_sample_every = 1000,
|
||||
num_samples = 25,
|
||||
results_folder = './results',
|
||||
gradient_accumulate_every = 2,
|
||||
amp = False,
|
||||
fp16 = False,
|
||||
split_batches = True,
|
||||
convert_image_to = None
|
||||
step_start_ema = 2000,
|
||||
update_ema_every = 10,
|
||||
save_and_sample_every = 1000,
|
||||
results_folder = './results'
|
||||
):
|
||||
super().__init__()
|
||||
|
||||
self.accelerator = Accelerator(
|
||||
split_batches = split_batches,
|
||||
mixed_precision = 'fp16' if fp16 else 'no'
|
||||
)
|
||||
|
||||
self.accelerator.native_amp = amp
|
||||
|
||||
self.model = diffusion_model
|
||||
self.ema = EMA(ema_decay)
|
||||
self.ema_model = copy.deepcopy(self.model)
|
||||
self.update_ema_every = update_ema_every
|
||||
|
||||
assert has_int_squareroot(num_samples), 'number of samples must have an integer square root'
|
||||
self.num_samples = num_samples
|
||||
self.step_start_ema = step_start_ema
|
||||
self.save_and_sample_every = save_and_sample_every
|
||||
|
||||
self.batch_size = train_batch_size
|
||||
self.gradient_accumulate_every = gradient_accumulate_every
|
||||
|
||||
self.train_num_steps = train_num_steps
|
||||
self.image_size = diffusion_model.image_size
|
||||
self.gradient_accumulate_every = gradient_accumulate_every
|
||||
self.train_num_steps = train_num_steps
|
||||
|
||||
# dataset and dataloader
|
||||
|
||||
self.ds = Dataset(folder, self.image_size, augment_horizontal_flip = augment_horizontal_flip, convert_image_to = convert_image_to)
|
||||
dl = DataLoader(self.ds, batch_size = train_batch_size, shuffle = True, pin_memory = True, num_workers = cpu_count())
|
||||
|
||||
dl = self.accelerator.prepare(dl)
|
||||
self.dl = cycle(dl)
|
||||
|
||||
# optimizer
|
||||
|
||||
self.opt = Adam(diffusion_model.parameters(), lr = train_lr, betas = adam_betas)
|
||||
|
||||
# for logging results in a folder periodically
|
||||
|
||||
if self.accelerator.is_main_process:
|
||||
self.ema = EMA(diffusion_model, beta = ema_decay, update_every = ema_update_every)
|
||||
|
||||
self.results_folder = Path(results_folder)
|
||||
self.results_folder.mkdir(exist_ok = True)
|
||||
|
||||
# step counter state
|
||||
self.ds = Dataset(folder, image_size)
|
||||
self.dl = cycle(data.DataLoader(self.ds, batch_size = train_batch_size, shuffle=True, pin_memory=True))
|
||||
self.opt = Adam(diffusion_model.parameters(), lr=train_lr)
|
||||
|
||||
self.step = 0
|
||||
|
||||
# prepare model, dataloader, optimizer with accelerator
|
||||
self.amp = amp
|
||||
self.scaler = GradScaler(enabled = amp)
|
||||
|
||||
self.model, self.opt = self.accelerator.prepare(self.model, self.opt)
|
||||
self.results_folder = Path(results_folder)
|
||||
self.results_folder.mkdir(exist_ok = True)
|
||||
|
||||
self.reset_parameters()
|
||||
|
||||
def reset_parameters(self):
|
||||
self.ema_model.load_state_dict(self.model.state_dict())
|
||||
|
||||
def step_ema(self):
|
||||
if self.step < self.step_start_ema:
|
||||
self.reset_parameters()
|
||||
return
|
||||
self.ema.update_model_average(self.ema_model, self.model)
|
||||
|
||||
def save(self, milestone):
|
||||
if not self.accelerator.is_local_main_process:
|
||||
return
|
||||
|
||||
data = {
|
||||
'step': self.step,
|
||||
'model': self.accelerator.get_state_dict(self.model),
|
||||
'opt': self.opt.state_dict(),
|
||||
'ema': self.ema.state_dict(),
|
||||
'scaler': self.accelerator.scaler.state_dict() if exists(self.accelerator.scaler) else None
|
||||
'model': self.model.state_dict(),
|
||||
'ema': self.ema_model.state_dict(),
|
||||
'scaler': self.scaler.state_dict()
|
||||
}
|
||||
|
||||
torch.save(data, str(self.results_folder / f'model-{milestone}.pt'))
|
||||
|
||||
def load(self, milestone):
|
||||
accelerator = self.accelerator
|
||||
device = accelerator.device
|
||||
|
||||
data = torch.load(str(self.results_folder / f'model-{milestone}.pt'), map_location=device)
|
||||
|
||||
model = self.accelerator.unwrap_model(self.model)
|
||||
model.load_state_dict(data['model'])
|
||||
data = torch.load(str(self.results_folder / f'model-{milestone}.pt'))
|
||||
|
||||
self.step = data['step']
|
||||
self.opt.load_state_dict(data['opt'])
|
||||
self.ema.load_state_dict(data['ema'])
|
||||
|
||||
if exists(self.accelerator.scaler) and exists(data['scaler']):
|
||||
self.accelerator.scaler.load_state_dict(data['scaler'])
|
||||
self.model.load_state_dict(data['model'])
|
||||
self.ema_model.load_state_dict(data['ema'])
|
||||
self.scaler.load_state_dict(data['scaler'])
|
||||
|
||||
def train(self):
|
||||
accelerator = self.accelerator
|
||||
device = accelerator.device
|
||||
|
||||
with tqdm(initial = self.step, total = self.train_num_steps, disable = not accelerator.is_main_process) as pbar:
|
||||
with tqdm(initial = self.step, total = self.train_num_steps) as pbar:
|
||||
|
||||
while self.step < self.train_num_steps:
|
||||
for i in range(self.gradient_accumulate_every):
|
||||
data = next(self.dl).cuda()
|
||||
|
||||
total_loss = 0.
|
||||
|
||||
for _ in range(self.gradient_accumulate_every):
|
||||
data = next(self.dl).to(device)
|
||||
|
||||
with self.accelerator.autocast():
|
||||
with autocast(enabled = self.amp):
|
||||
loss = self.model(data)
|
||||
loss = loss / self.gradient_accumulate_every
|
||||
total_loss += loss.item()
|
||||
self.scaler.scale(loss / self.gradient_accumulate_every).backward()
|
||||
|
||||
self.accelerator.backward(loss)
|
||||
pbar.set_description(f'loss: {loss.item():.4f}')
|
||||
|
||||
accelerator.clip_grad_norm_(self.model.parameters(), 1.0)
|
||||
pbar.set_description(f'loss: {total_loss:.4f}')
|
||||
|
||||
accelerator.wait_for_everyone()
|
||||
|
||||
self.opt.step()
|
||||
self.scaler.step(self.opt)
|
||||
self.scaler.update()
|
||||
self.opt.zero_grad()
|
||||
|
||||
accelerator.wait_for_everyone()
|
||||
if self.step % self.update_ema_every == 0:
|
||||
self.step_ema()
|
||||
|
||||
if self.step != 0 and self.step % self.save_and_sample_every == 0:
|
||||
self.ema_model.eval()
|
||||
|
||||
milestone = self.step // self.save_and_sample_every
|
||||
batches = num_to_groups(36, self.batch_size)
|
||||
all_images_list = list(map(lambda n: self.ema_model.sample(batch_size=n), batches))
|
||||
all_images = torch.cat(all_images_list, dim=0)
|
||||
utils.save_image(all_images, str(self.results_folder / f'sample-{milestone}.png'), nrow = 6)
|
||||
self.save(milestone)
|
||||
|
||||
self.step += 1
|
||||
if accelerator.is_main_process:
|
||||
self.ema.to(device)
|
||||
self.ema.update()
|
||||
|
||||
if self.step != 0 and self.step % self.save_and_sample_every == 0:
|
||||
self.ema.ema_model.eval()
|
||||
|
||||
with torch.no_grad():
|
||||
milestone = self.step // self.save_and_sample_every
|
||||
batches = num_to_groups(self.num_samples, self.batch_size)
|
||||
all_images_list = list(map(lambda n: self.ema.ema_model.sample(batch_size=n), batches))
|
||||
|
||||
all_images = torch.cat(all_images_list, dim = 0)
|
||||
utils.save_image(all_images, str(self.results_folder / f'sample-{milestone}.png'), nrow = int(math.sqrt(self.num_samples)))
|
||||
self.save(milestone)
|
||||
|
||||
pbar.update(1)
|
||||
|
||||
accelerator.print('training complete')
|
||||
print('training complete')
|
||||
|
||||
@@ -1,695 +0,0 @@
|
||||
import math
|
||||
from random import random
|
||||
from functools import partial
|
||||
from collections import namedtuple
|
||||
|
||||
import torch
|
||||
from torch import nn, einsum
|
||||
import torch.nn.functional as F
|
||||
|
||||
from einops import rearrange, reduce
|
||||
from einops.layers.torch import Rearrange
|
||||
|
||||
from tqdm.auto import tqdm
|
||||
|
||||
# constants
|
||||
|
||||
ModelPrediction = namedtuple('ModelPrediction', ['pred_noise', 'pred_x_start'])
|
||||
|
||||
# helpers functions
|
||||
|
||||
def exists(x):
|
||||
return x is not None
|
||||
|
||||
def default(val, d):
|
||||
if exists(val):
|
||||
return val
|
||||
return d() if callable(d) else d
|
||||
|
||||
def identity(t, *args, **kwargs):
|
||||
return t
|
||||
|
||||
def cycle(dl):
|
||||
while True:
|
||||
for data in dl:
|
||||
yield data
|
||||
|
||||
def has_int_squareroot(num):
|
||||
return (math.sqrt(num) ** 2) == num
|
||||
|
||||
def num_to_groups(num, divisor):
|
||||
groups = num // divisor
|
||||
remainder = num % divisor
|
||||
arr = [divisor] * groups
|
||||
if remainder > 0:
|
||||
arr.append(remainder)
|
||||
return arr
|
||||
|
||||
def convert_image_to_fn(img_type, image):
|
||||
if image.mode != img_type:
|
||||
return image.convert(img_type)
|
||||
return image
|
||||
|
||||
# normalization functions
|
||||
|
||||
def normalize_to_neg_one_to_one(img):
|
||||
return img * 2 - 1
|
||||
|
||||
def unnormalize_to_zero_to_one(t):
|
||||
return (t + 1) * 0.5
|
||||
|
||||
# small helper modules
|
||||
|
||||
class Residual(nn.Module):
|
||||
def __init__(self, fn):
|
||||
super().__init__()
|
||||
self.fn = fn
|
||||
|
||||
def forward(self, x, *args, **kwargs):
|
||||
return self.fn(x, *args, **kwargs) + x
|
||||
|
||||
def Upsample(dim, dim_out = None):
|
||||
return nn.Sequential(
|
||||
nn.Upsample(scale_factor = 2, mode = 'nearest'),
|
||||
nn.Conv1d(dim, default(dim_out, dim), 3, padding = 1)
|
||||
)
|
||||
|
||||
def Downsample(dim, dim_out = None):
|
||||
return nn.Conv1d(dim, default(dim_out, dim), 4, 2, 1)
|
||||
|
||||
class WeightStandardizedConv2d(nn.Conv1d):
|
||||
"""
|
||||
https://arxiv.org/abs/1903.10520
|
||||
weight standardization purportedly works synergistically with group normalization
|
||||
"""
|
||||
def forward(self, x):
|
||||
eps = 1e-5 if x.dtype == torch.float32 else 1e-3
|
||||
|
||||
weight = self.weight
|
||||
mean = reduce(weight, 'o ... -> o 1 1', 'mean')
|
||||
var = reduce(weight, 'o ... -> o 1 1', partial(torch.var, unbiased = False))
|
||||
normalized_weight = (weight - mean) * (var + eps).rsqrt()
|
||||
|
||||
return F.conv1d(x, normalized_weight, self.bias, self.stride, self.padding, self.dilation, self.groups)
|
||||
|
||||
class LayerNorm(nn.Module):
|
||||
def __init__(self, dim):
|
||||
super().__init__()
|
||||
self.g = nn.Parameter(torch.ones(1, dim, 1))
|
||||
|
||||
def forward(self, x):
|
||||
eps = 1e-5 if x.dtype == torch.float32 else 1e-3
|
||||
var = torch.var(x, dim = 1, unbiased = False, keepdim = True)
|
||||
mean = torch.mean(x, dim = 1, keepdim = True)
|
||||
return (x - mean) * (var + eps).rsqrt() * self.g
|
||||
|
||||
class PreNorm(nn.Module):
|
||||
def __init__(self, dim, fn):
|
||||
super().__init__()
|
||||
self.fn = fn
|
||||
self.norm = LayerNorm(dim)
|
||||
|
||||
def forward(self, x):
|
||||
x = self.norm(x)
|
||||
return self.fn(x)
|
||||
|
||||
# sinusoidal positional embeds
|
||||
|
||||
class SinusoidalPosEmb(nn.Module):
|
||||
def __init__(self, dim):
|
||||
super().__init__()
|
||||
self.dim = dim
|
||||
|
||||
def forward(self, x):
|
||||
device = x.device
|
||||
half_dim = self.dim // 2
|
||||
emb = math.log(10000) / (half_dim - 1)
|
||||
emb = torch.exp(torch.arange(half_dim, device=device) * -emb)
|
||||
emb = x[:, None] * emb[None, :]
|
||||
emb = torch.cat((emb.sin(), emb.cos()), dim=-1)
|
||||
return emb
|
||||
|
||||
class RandomOrLearnedSinusoidalPosEmb(nn.Module):
|
||||
""" following @crowsonkb 's lead with random (learned optional) sinusoidal pos emb """
|
||||
""" https://github.com/crowsonkb/v-diffusion-jax/blob/master/diffusion/models/danbooru_128.py#L8 """
|
||||
|
||||
def __init__(self, dim, is_random = False):
|
||||
super().__init__()
|
||||
assert (dim % 2) == 0
|
||||
half_dim = dim // 2
|
||||
self.weights = nn.Parameter(torch.randn(half_dim), requires_grad = not is_random)
|
||||
|
||||
def forward(self, x):
|
||||
x = rearrange(x, 'b -> b 1')
|
||||
freqs = x * rearrange(self.weights, 'd -> 1 d') * 2 * math.pi
|
||||
fouriered = torch.cat((freqs.sin(), freqs.cos()), dim = -1)
|
||||
fouriered = torch.cat((x, fouriered), dim = -1)
|
||||
return fouriered
|
||||
|
||||
# building block modules
|
||||
|
||||
class Block(nn.Module):
|
||||
def __init__(self, dim, dim_out, groups = 8):
|
||||
super().__init__()
|
||||
self.proj = WeightStandardizedConv2d(dim, dim_out, 3, padding = 1)
|
||||
self.norm = nn.GroupNorm(groups, dim_out)
|
||||
self.act = nn.SiLU()
|
||||
|
||||
def forward(self, x, scale_shift = None):
|
||||
x = self.proj(x)
|
||||
x = self.norm(x)
|
||||
|
||||
if exists(scale_shift):
|
||||
scale, shift = scale_shift
|
||||
x = x * (scale + 1) + shift
|
||||
|
||||
x = self.act(x)
|
||||
return x
|
||||
|
||||
class ResnetBlock(nn.Module):
|
||||
def __init__(self, dim, dim_out, *, time_emb_dim = None, groups = 8):
|
||||
super().__init__()
|
||||
self.mlp = nn.Sequential(
|
||||
nn.SiLU(),
|
||||
nn.Linear(time_emb_dim, dim_out * 2)
|
||||
) if exists(time_emb_dim) else None
|
||||
|
||||
self.block1 = Block(dim, dim_out, groups = groups)
|
||||
self.block2 = Block(dim_out, dim_out, groups = groups)
|
||||
self.res_conv = nn.Conv1d(dim, dim_out, 1) if dim != dim_out else nn.Identity()
|
||||
|
||||
def forward(self, x, time_emb = None):
|
||||
|
||||
scale_shift = None
|
||||
if exists(self.mlp) and exists(time_emb):
|
||||
time_emb = self.mlp(time_emb)
|
||||
time_emb = rearrange(time_emb, 'b c -> b c 1')
|
||||
scale_shift = time_emb.chunk(2, dim = 1)
|
||||
|
||||
h = self.block1(x, scale_shift = scale_shift)
|
||||
|
||||
h = self.block2(h)
|
||||
|
||||
return h + self.res_conv(x)
|
||||
|
||||
class LinearAttention(nn.Module):
|
||||
def __init__(self, dim, heads = 4, dim_head = 32):
|
||||
super().__init__()
|
||||
self.scale = dim_head ** -0.5
|
||||
self.heads = heads
|
||||
hidden_dim = dim_head * heads
|
||||
self.to_qkv = nn.Conv1d(dim, hidden_dim * 3, 1, bias = False)
|
||||
|
||||
self.to_out = nn.Sequential(
|
||||
nn.Conv1d(hidden_dim, dim, 1),
|
||||
LayerNorm(dim)
|
||||
)
|
||||
|
||||
def forward(self, x):
|
||||
b, c, n = x.shape
|
||||
qkv = self.to_qkv(x).chunk(3, dim = 1)
|
||||
q, k, v = map(lambda t: rearrange(t, 'b (h c) n -> b h c n', h = self.heads), qkv)
|
||||
|
||||
q = q.softmax(dim = -2)
|
||||
k = k.softmax(dim = -1)
|
||||
|
||||
q = q * self.scale
|
||||
|
||||
context = torch.einsum('b h d n, b h e n -> b h d e', k, v)
|
||||
|
||||
out = torch.einsum('b h d e, b h d n -> b h e n', context, q)
|
||||
out = rearrange(out, 'b h c n -> b (h c) n', h = self.heads)
|
||||
return self.to_out(out)
|
||||
|
||||
class Attention(nn.Module):
|
||||
def __init__(self, dim, heads = 4, dim_head = 32):
|
||||
super().__init__()
|
||||
self.scale = dim_head ** -0.5
|
||||
self.heads = heads
|
||||
hidden_dim = dim_head * heads
|
||||
|
||||
self.to_qkv = nn.Conv1d(dim, hidden_dim * 3, 1, bias = False)
|
||||
self.to_out = nn.Conv1d(hidden_dim, dim, 1)
|
||||
|
||||
def forward(self, x):
|
||||
b, c, n = x.shape
|
||||
qkv = self.to_qkv(x).chunk(3, dim = 1)
|
||||
q, k, v = map(lambda t: rearrange(t, 'b (h c) n -> b h c n', h = self.heads), qkv)
|
||||
|
||||
q = q * self.scale
|
||||
|
||||
sim = einsum('b h d i, b h d j -> b h i j', q, k)
|
||||
attn = sim.softmax(dim = -1)
|
||||
out = einsum('b h i j, b h d j -> b h i d', attn, v)
|
||||
|
||||
out = rearrange(out, 'b h n d -> b (h d) n')
|
||||
return self.to_out(out)
|
||||
|
||||
# model
|
||||
|
||||
class Unet1D(nn.Module):
|
||||
def __init__(
|
||||
self,
|
||||
dim,
|
||||
init_dim = None,
|
||||
out_dim = None,
|
||||
dim_mults=(1, 2, 4, 8),
|
||||
channels = 3,
|
||||
self_condition = False,
|
||||
resnet_block_groups = 8,
|
||||
learned_variance = False,
|
||||
learned_sinusoidal_cond = False,
|
||||
random_fourier_features = False,
|
||||
learned_sinusoidal_dim = 16
|
||||
):
|
||||
super().__init__()
|
||||
|
||||
# determine dimensions
|
||||
|
||||
self.channels = channels
|
||||
self.self_condition = self_condition
|
||||
input_channels = channels * (2 if self_condition else 1)
|
||||
|
||||
init_dim = default(init_dim, dim)
|
||||
self.init_conv = nn.Conv1d(input_channels, init_dim, 7, padding = 3)
|
||||
|
||||
dims = [init_dim, *map(lambda m: dim * m, dim_mults)]
|
||||
in_out = list(zip(dims[:-1], dims[1:]))
|
||||
|
||||
block_klass = partial(ResnetBlock, groups = resnet_block_groups)
|
||||
|
||||
# time embeddings
|
||||
|
||||
time_dim = dim * 4
|
||||
|
||||
self.random_or_learned_sinusoidal_cond = learned_sinusoidal_cond or random_fourier_features
|
||||
|
||||
if self.random_or_learned_sinusoidal_cond:
|
||||
sinu_pos_emb = RandomOrLearnedSinusoidalPosEmb(learned_sinusoidal_dim, random_fourier_features)
|
||||
fourier_dim = learned_sinusoidal_dim + 1
|
||||
else:
|
||||
sinu_pos_emb = SinusoidalPosEmb(dim)
|
||||
fourier_dim = dim
|
||||
|
||||
self.time_mlp = nn.Sequential(
|
||||
sinu_pos_emb,
|
||||
nn.Linear(fourier_dim, time_dim),
|
||||
nn.GELU(),
|
||||
nn.Linear(time_dim, time_dim)
|
||||
)
|
||||
|
||||
# layers
|
||||
|
||||
self.downs = nn.ModuleList([])
|
||||
self.ups = nn.ModuleList([])
|
||||
num_resolutions = len(in_out)
|
||||
|
||||
for ind, (dim_in, dim_out) in enumerate(in_out):
|
||||
is_last = ind >= (num_resolutions - 1)
|
||||
|
||||
self.downs.append(nn.ModuleList([
|
||||
block_klass(dim_in, dim_in, time_emb_dim = time_dim),
|
||||
block_klass(dim_in, dim_in, time_emb_dim = time_dim),
|
||||
Residual(PreNorm(dim_in, LinearAttention(dim_in))),
|
||||
Downsample(dim_in, dim_out) if not is_last else nn.Conv1d(dim_in, dim_out, 3, padding = 1)
|
||||
]))
|
||||
|
||||
mid_dim = dims[-1]
|
||||
self.mid_block1 = block_klass(mid_dim, mid_dim, time_emb_dim = time_dim)
|
||||
self.mid_attn = Residual(PreNorm(mid_dim, Attention(mid_dim)))
|
||||
self.mid_block2 = block_klass(mid_dim, mid_dim, time_emb_dim = time_dim)
|
||||
|
||||
for ind, (dim_in, dim_out) in enumerate(reversed(in_out)):
|
||||
is_last = ind == (len(in_out) - 1)
|
||||
|
||||
self.ups.append(nn.ModuleList([
|
||||
block_klass(dim_out + dim_in, dim_out, time_emb_dim = time_dim),
|
||||
block_klass(dim_out + dim_in, dim_out, time_emb_dim = time_dim),
|
||||
Residual(PreNorm(dim_out, LinearAttention(dim_out))),
|
||||
Upsample(dim_out, dim_in) if not is_last else nn.Conv1d(dim_out, dim_in, 3, padding = 1)
|
||||
]))
|
||||
|
||||
default_out_dim = channels * (1 if not learned_variance else 2)
|
||||
self.out_dim = default(out_dim, default_out_dim)
|
||||
|
||||
self.final_res_block = block_klass(dim * 2, dim, time_emb_dim = time_dim)
|
||||
self.final_conv = nn.Conv1d(dim, self.out_dim, 1)
|
||||
|
||||
def forward(self, x, time, x_self_cond = None):
|
||||
if self.self_condition:
|
||||
x_self_cond = default(x_self_cond, lambda: torch.zeros_like(x))
|
||||
x = torch.cat((x_self_cond, x), dim = 1)
|
||||
|
||||
x = self.init_conv(x)
|
||||
r = x.clone()
|
||||
|
||||
t = self.time_mlp(time)
|
||||
|
||||
h = []
|
||||
|
||||
for block1, block2, attn, downsample in self.downs:
|
||||
x = block1(x, t)
|
||||
h.append(x)
|
||||
|
||||
x = block2(x, t)
|
||||
x = attn(x)
|
||||
h.append(x)
|
||||
|
||||
x = downsample(x)
|
||||
|
||||
x = self.mid_block1(x, t)
|
||||
x = self.mid_attn(x)
|
||||
x = self.mid_block2(x, t)
|
||||
|
||||
for block1, block2, attn, upsample in self.ups:
|
||||
x = torch.cat((x, h.pop()), dim = 1)
|
||||
x = block1(x, t)
|
||||
|
||||
x = torch.cat((x, h.pop()), dim = 1)
|
||||
x = block2(x, t)
|
||||
x = attn(x)
|
||||
|
||||
x = upsample(x)
|
||||
|
||||
x = torch.cat((x, r), dim = 1)
|
||||
|
||||
x = self.final_res_block(x, t)
|
||||
return self.final_conv(x)
|
||||
|
||||
# gaussian diffusion trainer class
|
||||
|
||||
def extract(a, t, x_shape):
|
||||
b, *_ = t.shape
|
||||
out = a.gather(-1, t)
|
||||
return out.reshape(b, *((1,) * (len(x_shape) - 1)))
|
||||
|
||||
def linear_beta_schedule(timesteps):
|
||||
scale = 1000 / timesteps
|
||||
beta_start = scale * 0.0001
|
||||
beta_end = scale * 0.02
|
||||
return torch.linspace(beta_start, beta_end, timesteps, dtype = torch.float64)
|
||||
|
||||
def cosine_beta_schedule(timesteps, s = 0.008):
|
||||
"""
|
||||
cosine schedule
|
||||
as proposed in https://openreview.net/forum?id=-NEXDKk8gZ
|
||||
"""
|
||||
steps = timesteps + 1
|
||||
x = torch.linspace(0, timesteps, steps, dtype = torch.float64)
|
||||
alphas_cumprod = torch.cos(((x / timesteps) + s) / (1 + s) * math.pi * 0.5) ** 2
|
||||
alphas_cumprod = alphas_cumprod / alphas_cumprod[0]
|
||||
betas = 1 - (alphas_cumprod[1:] / alphas_cumprod[:-1])
|
||||
return torch.clip(betas, 0, 0.999)
|
||||
|
||||
class GaussianDiffusion1D(nn.Module):
|
||||
def __init__(
|
||||
self,
|
||||
model,
|
||||
*,
|
||||
seq_length,
|
||||
timesteps = 1000,
|
||||
sampling_timesteps = None,
|
||||
loss_type = 'l1',
|
||||
objective = 'pred_noise',
|
||||
beta_schedule = 'cosine',
|
||||
p2_loss_weight_gamma = 0.,
|
||||
p2_loss_weight_k = 1,
|
||||
ddim_sampling_eta = 1.
|
||||
):
|
||||
super().__init__()
|
||||
self.model = model
|
||||
self.channels = self.model.channels
|
||||
self.self_condition = self.model.self_condition
|
||||
|
||||
self.seq_length = seq_length
|
||||
|
||||
self.objective = objective
|
||||
|
||||
assert objective in {'pred_noise', 'pred_x0', 'pred_v'}, 'objective must be either pred_noise (predict noise) or pred_x0 (predict image start) or pred_v (predict v [v-parameterization as defined in appendix D of progressive distillation paper, used in imagen-video successfully])'
|
||||
|
||||
if beta_schedule == 'linear':
|
||||
betas = linear_beta_schedule(timesteps)
|
||||
elif beta_schedule == 'cosine':
|
||||
betas = cosine_beta_schedule(timesteps)
|
||||
else:
|
||||
raise ValueError(f'unknown beta schedule {beta_schedule}')
|
||||
|
||||
alphas = 1. - betas
|
||||
alphas_cumprod = torch.cumprod(alphas, dim=0)
|
||||
alphas_cumprod_prev = F.pad(alphas_cumprod[:-1], (1, 0), value = 1.)
|
||||
|
||||
timesteps, = betas.shape
|
||||
self.num_timesteps = int(timesteps)
|
||||
self.loss_type = loss_type
|
||||
|
||||
# sampling related parameters
|
||||
|
||||
self.sampling_timesteps = default(sampling_timesteps, timesteps) # default num sampling timesteps to number of timesteps at training
|
||||
|
||||
assert self.sampling_timesteps <= timesteps
|
||||
self.is_ddim_sampling = self.sampling_timesteps < timesteps
|
||||
self.ddim_sampling_eta = ddim_sampling_eta
|
||||
|
||||
# helper function to register buffer from float64 to float32
|
||||
|
||||
register_buffer = lambda name, val: self.register_buffer(name, val.to(torch.float32))
|
||||
|
||||
register_buffer('betas', betas)
|
||||
register_buffer('alphas_cumprod', alphas_cumprod)
|
||||
register_buffer('alphas_cumprod_prev', alphas_cumprod_prev)
|
||||
|
||||
# calculations for diffusion q(x_t | x_{t-1}) and others
|
||||
|
||||
register_buffer('sqrt_alphas_cumprod', torch.sqrt(alphas_cumprod))
|
||||
register_buffer('sqrt_one_minus_alphas_cumprod', torch.sqrt(1. - alphas_cumprod))
|
||||
register_buffer('log_one_minus_alphas_cumprod', torch.log(1. - alphas_cumprod))
|
||||
register_buffer('sqrt_recip_alphas_cumprod', torch.sqrt(1. / alphas_cumprod))
|
||||
register_buffer('sqrt_recipm1_alphas_cumprod', torch.sqrt(1. / alphas_cumprod - 1))
|
||||
|
||||
# calculations for posterior q(x_{t-1} | x_t, x_0)
|
||||
|
||||
posterior_variance = betas * (1. - alphas_cumprod_prev) / (1. - alphas_cumprod)
|
||||
|
||||
# above: equal to 1. / (1. / (1. - alpha_cumprod_tm1) + alpha_t / beta_t)
|
||||
|
||||
register_buffer('posterior_variance', posterior_variance)
|
||||
|
||||
# below: log calculation clipped because the posterior variance is 0 at the beginning of the diffusion chain
|
||||
|
||||
register_buffer('posterior_log_variance_clipped', torch.log(posterior_variance.clamp(min =1e-20)))
|
||||
register_buffer('posterior_mean_coef1', betas * torch.sqrt(alphas_cumprod_prev) / (1. - alphas_cumprod))
|
||||
register_buffer('posterior_mean_coef2', (1. - alphas_cumprod_prev) * torch.sqrt(alphas) / (1. - alphas_cumprod))
|
||||
|
||||
# calculate p2 reweighting
|
||||
|
||||
register_buffer('p2_loss_weight', (p2_loss_weight_k + alphas_cumprod / (1 - alphas_cumprod)) ** -p2_loss_weight_gamma)
|
||||
|
||||
def predict_start_from_noise(self, x_t, t, noise):
|
||||
return (
|
||||
extract(self.sqrt_recip_alphas_cumprod, t, x_t.shape) * x_t -
|
||||
extract(self.sqrt_recipm1_alphas_cumprod, t, x_t.shape) * noise
|
||||
)
|
||||
|
||||
def predict_noise_from_start(self, x_t, t, x0):
|
||||
return (
|
||||
(extract(self.sqrt_recip_alphas_cumprod, t, x_t.shape) * x_t - x0) / \
|
||||
extract(self.sqrt_recipm1_alphas_cumprod, t, x_t.shape)
|
||||
)
|
||||
|
||||
def predict_v(self, x_start, t, noise):
|
||||
return (
|
||||
extract(self.sqrt_alphas_cumprod, t, x_start.shape) * noise -
|
||||
extract(self.sqrt_one_minus_alphas_cumprod, t, x_start.shape) * x_start
|
||||
)
|
||||
|
||||
def predict_start_from_v(self, x_t, t, v):
|
||||
return (
|
||||
extract(self.sqrt_alphas_cumprod, t, x_t.shape) * x_t -
|
||||
extract(self.sqrt_one_minus_alphas_cumprod, t, x_t.shape) * v
|
||||
)
|
||||
|
||||
def q_posterior(self, x_start, x_t, t):
|
||||
posterior_mean = (
|
||||
extract(self.posterior_mean_coef1, t, x_t.shape) * x_start +
|
||||
extract(self.posterior_mean_coef2, t, x_t.shape) * x_t
|
||||
)
|
||||
posterior_variance = extract(self.posterior_variance, t, x_t.shape)
|
||||
posterior_log_variance_clipped = extract(self.posterior_log_variance_clipped, t, x_t.shape)
|
||||
return posterior_mean, posterior_variance, posterior_log_variance_clipped
|
||||
|
||||
def model_predictions(self, x, t, x_self_cond = None, clip_x_start = False):
|
||||
model_output = self.model(x, t, x_self_cond)
|
||||
maybe_clip = partial(torch.clamp, min = -1., max = 1.) if clip_x_start else identity
|
||||
|
||||
if self.objective == 'pred_noise':
|
||||
pred_noise = model_output
|
||||
x_start = self.predict_start_from_noise(x, t, pred_noise)
|
||||
x_start = maybe_clip(x_start)
|
||||
|
||||
elif self.objective == 'pred_x0':
|
||||
x_start = model_output
|
||||
x_start = maybe_clip(x_start)
|
||||
pred_noise = self.predict_noise_from_start(x, t, x_start)
|
||||
|
||||
elif self.objective == 'pred_v':
|
||||
v = model_output
|
||||
x_start = self.predict_start_from_v(x, t, v)
|
||||
x_start = maybe_clip(x_start)
|
||||
pred_noise = self.predict_noise_from_start(x, t, x_start)
|
||||
|
||||
return ModelPrediction(pred_noise, x_start)
|
||||
|
||||
def p_mean_variance(self, x, t, x_self_cond = None, clip_denoised = True):
|
||||
preds = self.model_predictions(x, t, x_self_cond)
|
||||
x_start = preds.pred_x_start
|
||||
|
||||
if clip_denoised:
|
||||
x_start.clamp_(-1., 1.)
|
||||
|
||||
model_mean, posterior_variance, posterior_log_variance = self.q_posterior(x_start = x_start, x_t = x, t = t)
|
||||
return model_mean, posterior_variance, posterior_log_variance, x_start
|
||||
|
||||
@torch.no_grad()
|
||||
def p_sample(self, x, t: int, x_self_cond = None, clip_denoised = True):
|
||||
b, *_, device = *x.shape, x.device
|
||||
batched_times = torch.full((x.shape[0],), t, device = x.device, dtype = torch.long)
|
||||
model_mean, _, model_log_variance, x_start = self.p_mean_variance(x = x, t = batched_times, x_self_cond = x_self_cond, clip_denoised = clip_denoised)
|
||||
noise = torch.randn_like(x) if t > 0 else 0. # no noise if t == 0
|
||||
pred_img = model_mean + (0.5 * model_log_variance).exp() * noise
|
||||
return pred_img, x_start
|
||||
|
||||
@torch.no_grad()
|
||||
def p_sample_loop(self, shape):
|
||||
batch, device = shape[0], self.betas.device
|
||||
|
||||
img = torch.randn(shape, device=device)
|
||||
|
||||
x_start = None
|
||||
|
||||
for t in tqdm(reversed(range(0, self.num_timesteps)), desc = 'sampling loop time step', total = self.num_timesteps):
|
||||
self_cond = x_start if self.self_condition else None
|
||||
img, x_start = self.p_sample(img, t, self_cond)
|
||||
|
||||
img = unnormalize_to_zero_to_one(img)
|
||||
return img
|
||||
|
||||
@torch.no_grad()
|
||||
def ddim_sample(self, shape, clip_denoised = True):
|
||||
batch, device, total_timesteps, sampling_timesteps, eta, objective = shape[0], self.betas.device, self.num_timesteps, self.sampling_timesteps, self.ddim_sampling_eta, self.objective
|
||||
|
||||
times = torch.linspace(-1, total_timesteps - 1, steps=sampling_timesteps + 1) # [-1, 0, 1, 2, ..., T-1] when sampling_timesteps == total_timesteps
|
||||
times = list(reversed(times.int().tolist()))
|
||||
time_pairs = list(zip(times[:-1], times[1:])) # [(T-1, T-2), (T-2, T-3), ..., (1, 0), (0, -1)]
|
||||
|
||||
img = torch.randn(shape, device = device)
|
||||
|
||||
x_start = None
|
||||
|
||||
for time, time_next in tqdm(time_pairs, desc = 'sampling loop time step'):
|
||||
time_cond = torch.full((batch,), time, device=device, dtype=torch.long)
|
||||
self_cond = x_start if self.self_condition else None
|
||||
pred_noise, x_start, *_ = self.model_predictions(img, time_cond, self_cond, clip_x_start = clip_denoised)
|
||||
|
||||
if time_next < 0:
|
||||
img = x_start
|
||||
continue
|
||||
|
||||
alpha = self.alphas_cumprod[time]
|
||||
alpha_next = self.alphas_cumprod[time_next]
|
||||
|
||||
sigma = eta * ((1 - alpha / alpha_next) * (1 - alpha_next) / (1 - alpha)).sqrt()
|
||||
c = (1 - alpha_next - sigma ** 2).sqrt()
|
||||
|
||||
noise = torch.randn_like(img)
|
||||
|
||||
img = x_start * alpha_next.sqrt() + \
|
||||
c * pred_noise + \
|
||||
sigma * noise
|
||||
|
||||
img = unnormalize_to_zero_to_one(img)
|
||||
return img
|
||||
|
||||
@torch.no_grad()
|
||||
def sample(self, batch_size = 16):
|
||||
seq_length, channels = self.seq_length, self.channels
|
||||
sample_fn = self.p_sample_loop if not self.is_ddim_sampling else self.ddim_sample
|
||||
return sample_fn((batch_size, channels, seq_length))
|
||||
|
||||
@torch.no_grad()
|
||||
def interpolate(self, x1, x2, t = None, lam = 0.5):
|
||||
b, *_, device = *x1.shape, x1.device
|
||||
t = default(t, self.num_timesteps - 1)
|
||||
|
||||
assert x1.shape == x2.shape
|
||||
|
||||
t_batched = torch.stack([torch.tensor(t, device = device)] * b)
|
||||
xt1, xt2 = map(lambda x: self.q_sample(x, t = t_batched), (x1, x2))
|
||||
|
||||
img = (1 - lam) * xt1 + lam * xt2
|
||||
for i in tqdm(reversed(range(0, t)), desc = 'interpolation sample time step', total = t):
|
||||
img = self.p_sample(img, torch.full((b,), i, device=device, dtype=torch.long))
|
||||
|
||||
return img
|
||||
|
||||
def q_sample(self, x_start, t, noise=None):
|
||||
noise = default(noise, lambda: torch.randn_like(x_start))
|
||||
|
||||
return (
|
||||
extract(self.sqrt_alphas_cumprod, t, x_start.shape) * x_start +
|
||||
extract(self.sqrt_one_minus_alphas_cumprod, t, x_start.shape) * noise
|
||||
)
|
||||
|
||||
@property
|
||||
def loss_fn(self):
|
||||
if self.loss_type == 'l1':
|
||||
return F.l1_loss
|
||||
elif self.loss_type == 'l2':
|
||||
return F.mse_loss
|
||||
else:
|
||||
raise ValueError(f'invalid loss type {self.loss_type}')
|
||||
|
||||
def p_losses(self, x_start, t, noise = None):
|
||||
b, c, n = x_start.shape
|
||||
noise = default(noise, lambda: torch.randn_like(x_start))
|
||||
|
||||
# noise sample
|
||||
|
||||
x = self.q_sample(x_start = x_start, t = t, noise = noise)
|
||||
|
||||
# if doing self-conditioning, 50% of the time, predict x_start from current set of times
|
||||
# and condition with unet with that
|
||||
# this technique will slow down training by 25%, but seems to lower FID significantly
|
||||
|
||||
x_self_cond = None
|
||||
if self.self_condition and random() < 0.5:
|
||||
with torch.no_grad():
|
||||
x_self_cond = self.model_predictions(x, t).pred_x_start
|
||||
x_self_cond.detach_()
|
||||
|
||||
# predict and take gradient step
|
||||
|
||||
model_out = self.model(x, t, x_self_cond)
|
||||
|
||||
if self.objective == 'pred_noise':
|
||||
target = noise
|
||||
elif self.objective == 'pred_x0':
|
||||
target = x_start
|
||||
elif self.objective == 'pred_v':
|
||||
v = self.predict_v(x_start, t, noise)
|
||||
target = v
|
||||
else:
|
||||
raise ValueError(f'unknown objective {self.objective}')
|
||||
|
||||
loss = self.loss_fn(model_out, target, reduction = 'none')
|
||||
loss = reduce(loss, 'b ... -> b (...)', 'mean')
|
||||
|
||||
loss = loss * extract(self.p2_loss_weight, t, loss.shape)
|
||||
return loss.mean()
|
||||
|
||||
def forward(self, img, *args, **kwargs):
|
||||
b, c, n, device, seq_length, = *img.shape, img.device, self.seq_length
|
||||
assert n == seq_length, f'seq length must be {seq_length}'
|
||||
t = torch.randint(0, self.num_timesteps, (b,), device=device).long()
|
||||
|
||||
img = normalize_to_neg_one_to_one(img)
|
||||
return self.p_losses(img, t, *args, **kwargs)
|
||||
@@ -1,240 +0,0 @@
|
||||
from math import sqrt
|
||||
from random import random
|
||||
import torch
|
||||
from torch import nn, einsum
|
||||
import torch.nn.functional as F
|
||||
|
||||
from tqdm import tqdm
|
||||
from einops import rearrange, repeat, reduce
|
||||
|
||||
# helpers
|
||||
|
||||
def exists(val):
|
||||
return val is not None
|
||||
|
||||
def default(val, d):
|
||||
if exists(val):
|
||||
return val
|
||||
return d() if callable(d) else d
|
||||
|
||||
# tensor helpers
|
||||
|
||||
def log(t, eps = 1e-20):
|
||||
return torch.log(t.clamp(min = eps))
|
||||
|
||||
# normalization functions
|
||||
|
||||
def normalize_to_neg_one_to_one(img):
|
||||
return img * 2 - 1
|
||||
|
||||
def unnormalize_to_zero_to_one(t):
|
||||
return (t + 1) * 0.5
|
||||
|
||||
# main class
|
||||
|
||||
class ElucidatedDiffusion(nn.Module):
|
||||
def __init__(
|
||||
self,
|
||||
net,
|
||||
*,
|
||||
image_size,
|
||||
channels = 3,
|
||||
num_sample_steps = 32, # number of sampling steps
|
||||
sigma_min = 0.002, # min noise level
|
||||
sigma_max = 80, # max noise level
|
||||
sigma_data = 0.5, # standard deviation of data distribution
|
||||
rho = 7, # controls the sampling schedule
|
||||
P_mean = -1.2, # mean of log-normal distribution from which noise is drawn for training
|
||||
P_std = 1.2, # standard deviation of log-normal distribution from which noise is drawn for training
|
||||
S_churn = 80, # parameters for stochastic sampling - depends on dataset, Table 5 in apper
|
||||
S_tmin = 0.05,
|
||||
S_tmax = 50,
|
||||
S_noise = 1.003,
|
||||
):
|
||||
super().__init__()
|
||||
assert net.random_or_learned_sinusoidal_cond
|
||||
self.self_condition = net.self_condition
|
||||
|
||||
self.net = net
|
||||
|
||||
# image dimensions
|
||||
|
||||
self.channels = channels
|
||||
self.image_size = image_size
|
||||
|
||||
# parameters
|
||||
|
||||
self.sigma_min = sigma_min
|
||||
self.sigma_max = sigma_max
|
||||
self.sigma_data = sigma_data
|
||||
|
||||
self.rho = rho
|
||||
|
||||
self.P_mean = P_mean
|
||||
self.P_std = P_std
|
||||
|
||||
self.num_sample_steps = num_sample_steps # otherwise known as N in the paper
|
||||
|
||||
self.S_churn = S_churn
|
||||
self.S_tmin = S_tmin
|
||||
self.S_tmax = S_tmax
|
||||
self.S_noise = S_noise
|
||||
|
||||
@property
|
||||
def device(self):
|
||||
return next(self.net.parameters()).device
|
||||
|
||||
# derived preconditioning params - Table 1
|
||||
|
||||
def c_skip(self, sigma):
|
||||
return (self.sigma_data ** 2) / (sigma ** 2 + self.sigma_data ** 2)
|
||||
|
||||
def c_out(self, sigma):
|
||||
return sigma * self.sigma_data * (self.sigma_data ** 2 + sigma ** 2) ** -0.5
|
||||
|
||||
def c_in(self, sigma):
|
||||
return 1 * (sigma ** 2 + self.sigma_data ** 2) ** -0.5
|
||||
|
||||
def c_noise(self, sigma):
|
||||
return log(sigma) * 0.25
|
||||
|
||||
# preconditioned network output
|
||||
# equation (7) in the paper
|
||||
|
||||
def preconditioned_network_forward(self, noised_images, sigma, self_cond = None, clamp = False):
|
||||
batch, device = noised_images.shape[0], noised_images.device
|
||||
|
||||
if isinstance(sigma, float):
|
||||
sigma = torch.full((batch,), sigma, device = device)
|
||||
|
||||
padded_sigma = rearrange(sigma, 'b -> b 1 1 1')
|
||||
|
||||
net_out = self.net(
|
||||
self.c_in(padded_sigma) * noised_images,
|
||||
self.c_noise(sigma),
|
||||
self_cond
|
||||
)
|
||||
|
||||
out = self.c_skip(padded_sigma) * noised_images + self.c_out(padded_sigma) * net_out
|
||||
|
||||
if clamp:
|
||||
out = out.clamp(-1., 1.)
|
||||
|
||||
return out
|
||||
|
||||
# sampling
|
||||
|
||||
# sample schedule
|
||||
# equation (5) in the paper
|
||||
|
||||
def sample_schedule(self, num_sample_steps = None):
|
||||
num_sample_steps = default(num_sample_steps, self.num_sample_steps)
|
||||
|
||||
N = num_sample_steps
|
||||
inv_rho = 1 / self.rho
|
||||
|
||||
steps = torch.arange(num_sample_steps, device = self.device, dtype = torch.float32)
|
||||
sigmas = (self.sigma_max ** inv_rho + steps / (N - 1) * (self.sigma_min ** inv_rho - self.sigma_max ** inv_rho)) ** self.rho
|
||||
|
||||
sigmas = F.pad(sigmas, (0, 1), value = 0.) # last step is sigma value of 0.
|
||||
return sigmas
|
||||
|
||||
@torch.no_grad()
|
||||
def sample(self, batch_size = 16, num_sample_steps = None, clamp = True):
|
||||
num_sample_steps = default(num_sample_steps, self.num_sample_steps)
|
||||
|
||||
shape = (batch_size, self.channels, self.image_size, self.image_size)
|
||||
|
||||
# get the schedule, which is returned as (sigma, gamma) tuple, and pair up with the next sigma and gamma
|
||||
|
||||
sigmas = self.sample_schedule(num_sample_steps)
|
||||
|
||||
gammas = torch.where(
|
||||
(sigmas >= self.S_tmin) & (sigmas <= self.S_tmax),
|
||||
min(self.S_churn / num_sample_steps, sqrt(2) - 1),
|
||||
0.
|
||||
)
|
||||
|
||||
sigmas_and_gammas = list(zip(sigmas[:-1], sigmas[1:], gammas[:-1]))
|
||||
|
||||
# images is noise at the beginning
|
||||
|
||||
init_sigma = sigmas[0]
|
||||
|
||||
images = init_sigma * torch.randn(shape, device = self.device)
|
||||
|
||||
# for self conditioning
|
||||
|
||||
x_start = None
|
||||
|
||||
# gradually denoise
|
||||
|
||||
for sigma, sigma_next, gamma in tqdm(sigmas_and_gammas, desc = 'sampling time step'):
|
||||
sigma, sigma_next, gamma = map(lambda t: t.item(), (sigma, sigma_next, gamma))
|
||||
|
||||
eps = self.S_noise * torch.randn(shape, device = self.device) # stochastic sampling
|
||||
|
||||
sigma_hat = sigma + gamma * sigma
|
||||
images_hat = images + sqrt(sigma_hat ** 2 - sigma ** 2) * eps
|
||||
|
||||
self_cond = x_start if self.self_condition else None
|
||||
|
||||
model_output = self.preconditioned_network_forward(images_hat, sigma_hat, self_cond, clamp = clamp)
|
||||
denoised_over_sigma = (images_hat - model_output) / sigma_hat
|
||||
|
||||
images_next = images_hat + (sigma_next - sigma_hat) * denoised_over_sigma
|
||||
|
||||
# second order correction, if not the last timestep
|
||||
|
||||
if sigma_next != 0:
|
||||
self_cond = model_output if self.self_condition else None
|
||||
|
||||
model_output_next = self.preconditioned_network_forward(images_next, sigma_next, self_cond, clamp = clamp)
|
||||
denoised_prime_over_sigma = (images_next - model_output_next) / sigma_next
|
||||
images_next = images_hat + 0.5 * (sigma_next - sigma_hat) * (denoised_over_sigma + denoised_prime_over_sigma)
|
||||
|
||||
images = images_next
|
||||
x_start = model_output
|
||||
|
||||
images = images.clamp(-1., 1.)
|
||||
return unnormalize_to_zero_to_one(images)
|
||||
|
||||
# training
|
||||
|
||||
def loss_weight(self, sigma):
|
||||
return (sigma ** 2 + self.sigma_data ** 2) * (sigma * self.sigma_data) ** -2
|
||||
|
||||
def noise_distribution(self, batch_size):
|
||||
return (self.P_mean + self.P_std * torch.randn((batch_size,), device = self.device)).exp()
|
||||
|
||||
def forward(self, images):
|
||||
batch_size, c, h, w, device, image_size, channels = *images.shape, images.device, self.image_size, self.channels
|
||||
|
||||
assert h == image_size and w == image_size, f'height and width of image must be {image_size}'
|
||||
assert c == channels, 'mismatch of image channels'
|
||||
|
||||
images = normalize_to_neg_one_to_one(images)
|
||||
|
||||
sigmas = self.noise_distribution(batch_size)
|
||||
padded_sigmas = rearrange(sigmas, 'b -> b 1 1 1')
|
||||
|
||||
noise = torch.randn_like(images)
|
||||
|
||||
noised_images = images + padded_sigmas * noise # alphas are 1. in the paper
|
||||
|
||||
self_cond = None
|
||||
|
||||
if self.self_condition and random() < 0.5:
|
||||
# from hinton's group's bit diffusion paper
|
||||
with torch.no_grad():
|
||||
self_cond = self.preconditioned_network_forward(noised_images, sigmas)
|
||||
self_cond.detach_()
|
||||
|
||||
denoised = self.preconditioned_network_forward(noised_images, sigmas, self_cond)
|
||||
|
||||
losses = F.mse_loss(denoised, images, reduction = 'none')
|
||||
losses = reduce(losses, 'b ... -> b', 'mean')
|
||||
|
||||
losses = losses * self.loss_weight(sigmas)
|
||||
|
||||
return losses.mean()
|
||||
@@ -1,5 +1,4 @@
|
||||
import torch
|
||||
from collections import namedtuple
|
||||
from math import pi, sqrt, log as ln
|
||||
from inspect import isfunction
|
||||
from torch import nn, einsum
|
||||
@@ -11,8 +10,6 @@ from denoising_diffusion_pytorch.denoising_diffusion_pytorch import GaussianDiff
|
||||
|
||||
NAT = 1. / ln(2)
|
||||
|
||||
ModelPrediction = namedtuple('ModelPrediction', ['pred_noise', 'pred_x_start', 'pred_variance'])
|
||||
|
||||
# helper functions
|
||||
|
||||
def exists(x):
|
||||
@@ -25,7 +22,7 @@ def default(val, d):
|
||||
|
||||
# tensor helpers
|
||||
|
||||
def log(t, eps = 1e-15):
|
||||
def log(t, eps = 1e-12):
|
||||
return torch.log(t.clamp(min = eps))
|
||||
|
||||
def meanflat(x):
|
||||
@@ -70,33 +67,17 @@ def discretized_gaussian_log_likelihood(x, *, means, log_scales, thres = 0.999):
|
||||
class LearnedGaussianDiffusion(GaussianDiffusion):
|
||||
def __init__(
|
||||
self,
|
||||
model,
|
||||
denoise_fn,
|
||||
vb_loss_weight = 0.001, # lambda was 0.001 in the paper
|
||||
*args,
|
||||
**kwargs
|
||||
):
|
||||
super().__init__(model, *args, **kwargs)
|
||||
assert model.out_dim == (model.channels * 2), 'dimension out of unet must be twice the number of channels for learned variance - you can also set the `learned_variance` keyword argument on the Unet to be `True`'
|
||||
assert not model.self_condition, 'not supported yet'
|
||||
|
||||
super().__init__(denoise_fn, *args, **kwargs)
|
||||
assert denoise_fn.out_dim == (denoise_fn.channels * 2), 'dimension out of unet must be twice the number of channels for learned variance - you can also set the `learned_variance` keyword argument on the Unet to be `True`'
|
||||
self.vb_loss_weight = vb_loss_weight
|
||||
|
||||
def model_predictions(self, x, t):
|
||||
model_output = self.model(x, t)
|
||||
model_output, pred_variance = model_output.chunk(2, dim = 1)
|
||||
|
||||
if self.objective == 'pred_noise':
|
||||
pred_noise = model_output
|
||||
x_start = self.predict_start_from_noise(x, t, model_output)
|
||||
|
||||
elif self.objective == 'pred_x0':
|
||||
pred_noise = self.predict_noise_from_start(x, t, model_output)
|
||||
x_start = model_output
|
||||
|
||||
return ModelPrediction(pred_noise, x_start, pred_variance)
|
||||
|
||||
def p_mean_variance(self, *, x, t, clip_denoised, model_output = None):
|
||||
model_output = default(model_output, lambda: self.model(x, t))
|
||||
model_output = default(model_output, lambda: self.denoise_fn(x, t))
|
||||
pred_noise, var_interp_frac_unnormalized = model_output.chunk(2, dim = 1)
|
||||
|
||||
min_log = extract(self.posterior_log_variance_clipped, t, x.shape)
|
||||
@@ -121,7 +102,7 @@ class LearnedGaussianDiffusion(GaussianDiffusion):
|
||||
|
||||
# model output
|
||||
|
||||
model_output = self.model(x_t, t)
|
||||
model_output = self.denoise_fn(x_t, t)
|
||||
|
||||
# calculating kl loss for learned variance (interpolation)
|
||||
|
||||
|
||||
@@ -1,184 +0,0 @@
|
||||
import math
|
||||
import torch
|
||||
from torch import sqrt
|
||||
from torch import nn, einsum
|
||||
import torch.nn.functional as F
|
||||
from torch.special import expm1
|
||||
|
||||
from tqdm import tqdm
|
||||
from einops import rearrange, repeat, reduce
|
||||
from einops.layers.torch import Rearrange
|
||||
|
||||
# helpers
|
||||
|
||||
def exists(val):
|
||||
return val is not None
|
||||
|
||||
def default(val, d):
|
||||
if exists(val):
|
||||
return val
|
||||
return d() if callable(d) else d
|
||||
|
||||
# normalization functions
|
||||
|
||||
def normalize_to_neg_one_to_one(img):
|
||||
return img * 2 - 1
|
||||
|
||||
def unnormalize_to_zero_to_one(t):
|
||||
return (t + 1) * 0.5
|
||||
|
||||
# diffusion helpers
|
||||
|
||||
def right_pad_dims_to(x, t):
|
||||
padding_dims = x.ndim - t.ndim
|
||||
if padding_dims <= 0:
|
||||
return t
|
||||
return t.view(*t.shape, *((1,) * padding_dims))
|
||||
|
||||
# continuous schedules
|
||||
# log(snr) that approximates the original linear schedule
|
||||
|
||||
def log(t, eps = 1e-20):
|
||||
return torch.log(t.clamp(min = eps))
|
||||
|
||||
def alpha_cosine_log_snr(t, s = 0.008):
|
||||
return -log((torch.cos((t + s) / (1 + s) * math.pi * 0.5) ** -2) - 1, eps = 1e-5)
|
||||
|
||||
class VParamContinuousTimeGaussianDiffusion(nn.Module):
|
||||
"""
|
||||
a new type of parameterization in v-space proposed in https://arxiv.org/abs/2202.00512 that
|
||||
(1) allows for improved distillation over noise prediction objective and
|
||||
(2) noted in imagen-video to improve upsampling unets by removing the color shifting artifacts
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
model,
|
||||
*,
|
||||
image_size,
|
||||
channels = 3,
|
||||
num_sample_steps = 500,
|
||||
clip_sample_denoised = True,
|
||||
):
|
||||
super().__init__()
|
||||
assert model.random_or_learned_sinusoidal_cond
|
||||
assert not model.self_condition, 'not supported yet'
|
||||
|
||||
self.model = model
|
||||
|
||||
# image dimensions
|
||||
|
||||
self.channels = channels
|
||||
self.image_size = image_size
|
||||
|
||||
# continuous noise schedule related stuff
|
||||
|
||||
self.log_snr = alpha_cosine_log_snr
|
||||
|
||||
# sampling
|
||||
|
||||
self.num_sample_steps = num_sample_steps
|
||||
self.clip_sample_denoised = clip_sample_denoised
|
||||
|
||||
@property
|
||||
def device(self):
|
||||
return next(self.model.parameters()).device
|
||||
|
||||
def p_mean_variance(self, x, time, time_next):
|
||||
# reviewer found an error in the equation in the paper (missing sigma)
|
||||
# following - https://openreview.net/forum?id=2LdBqxc1Yv¬eId=rIQgH0zKsRt
|
||||
|
||||
log_snr = self.log_snr(time)
|
||||
log_snr_next = self.log_snr(time_next)
|
||||
c = -expm1(log_snr - log_snr_next)
|
||||
|
||||
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_v = self.model(x, batch_log_snr)
|
||||
|
||||
# shown in Appendix D in the paper
|
||||
x_start = alpha * x - sigma * pred_v
|
||||
|
||||
if self.clip_sample_denoised:
|
||||
x_start.clamp_(-1., 1.)
|
||||
|
||||
model_mean = alpha_next * (x * (1 - c) / alpha + c * x_start)
|
||||
|
||||
posterior_variance = squared_sigma_next * c
|
||||
|
||||
return model_mean, posterior_variance
|
||||
|
||||
# sampling related functions
|
||||
|
||||
@torch.no_grad()
|
||||
def p_sample(self, x, time, time_next):
|
||||
batch, *_, device = *x.shape, x.device
|
||||
|
||||
model_mean, model_variance = self.p_mean_variance(x = x, time = time, time_next = time_next)
|
||||
|
||||
if time_next == 0:
|
||||
return model_mean
|
||||
|
||||
noise = torch.randn_like(x)
|
||||
return model_mean + sqrt(model_variance) * noise
|
||||
|
||||
@torch.no_grad()
|
||||
def p_sample_loop(self, shape):
|
||||
batch = shape[0]
|
||||
|
||||
img = torch.randn(shape, device = self.device)
|
||||
steps = torch.linspace(1., 0., self.num_sample_steps + 1, device = self.device)
|
||||
|
||||
for i in tqdm(range(self.num_sample_steps), desc = 'sampling loop time step', total = self.num_sample_steps):
|
||||
times = steps[i]
|
||||
times_next = steps[i + 1]
|
||||
img = self.p_sample(img, times, times_next)
|
||||
|
||||
img.clamp_(-1., 1.)
|
||||
img = unnormalize_to_zero_to_one(img)
|
||||
return img
|
||||
|
||||
@torch.no_grad()
|
||||
def sample(self, batch_size = 16):
|
||||
return self.p_sample_loop((batch_size, self.channels, self.image_size, self.image_size))
|
||||
|
||||
# training related functions - noise prediction
|
||||
|
||||
def q_sample(self, x_start, times, noise = None):
|
||||
noise = default(noise, lambda: torch.randn_like(x_start))
|
||||
|
||||
log_snr = self.log_snr(times)
|
||||
|
||||
log_snr_padded = right_pad_dims_to(x_start, log_snr)
|
||||
alpha, sigma = sqrt(log_snr_padded.sigmoid()), sqrt((-log_snr_padded).sigmoid())
|
||||
x_noised = x_start * alpha + noise * sigma
|
||||
|
||||
return x_noised, log_snr, alpha, sigma
|
||||
|
||||
def random_times(self, batch_size):
|
||||
return torch.zeros((batch_size,), device = self.device).float().uniform_(0, 1)
|
||||
|
||||
def p_losses(self, x_start, times, noise = None):
|
||||
noise = default(noise, lambda: torch.randn_like(x_start))
|
||||
|
||||
x, log_snr, alpha, sigma = self.q_sample(x_start = x_start, times = times, noise = noise)
|
||||
|
||||
# described in section 4 as the prediction objective, with derivation in Appendix D
|
||||
v = alpha * noise - sigma * x_start
|
||||
|
||||
model_out = self.model(x, log_snr)
|
||||
|
||||
return F.mse_loss(model_out, v)
|
||||
|
||||
def forward(self, img, *args, **kwargs):
|
||||
b, c, h, w, device, img_size, = *img.shape, img.device, self.image_size
|
||||
assert h == img_size and w == img_size, f'height and width of image must be {img_size}'
|
||||
|
||||
times = self.random_times(b)
|
||||
img = normalize_to_neg_one_to_one(img)
|
||||
return self.p_losses(img, times, *args, **kwargs)
|
||||
@@ -22,24 +22,22 @@ def default(val, d):
|
||||
class WeightedObjectiveGaussianDiffusion(GaussianDiffusion):
|
||||
def __init__(
|
||||
self,
|
||||
model,
|
||||
denoise_fn,
|
||||
*args,
|
||||
pred_noise_loss_weight = 0.1,
|
||||
pred_x_start_loss_weight = 0.1,
|
||||
**kwargs
|
||||
):
|
||||
super().__init__(model, *args, **kwargs)
|
||||
channels = model.channels
|
||||
assert model.out_dim == (channels * 2 + 2), 'dimension out (out_dim) of unet must be twice the number of channels + 2 (for the softmax weighted sum) - for channels of 3, this should be (3 * 2) + 2 = 8'
|
||||
assert not model.self_condition, 'not supported yet'
|
||||
assert not self.is_ddim_sampling, 'ddim sampling cannot be used'
|
||||
super().__init__(denoise_fn, *args, **kwargs)
|
||||
channels = denoise_fn.channels
|
||||
assert denoise_fn.out_dim == (channels * 2 + 2), 'dimension out (out_dim) of unet must be twice the number of channels + 2 (for the softmax weighted sum) - for channels of 3, this should be (3 * 2) + 2 = 8'
|
||||
|
||||
self.split_dims = (channels, channels, 2)
|
||||
self.pred_noise_loss_weight = pred_noise_loss_weight
|
||||
self.pred_x_start_loss_weight = pred_x_start_loss_weight
|
||||
|
||||
def p_mean_variance(self, *, x, t, clip_denoised, model_output = None):
|
||||
model_output = self.model(x, t)
|
||||
model_output = self.denoise_fn(x, t)
|
||||
|
||||
pred_noise, pred_x_start, weights = model_output.split(self.split_dims, dim = 1)
|
||||
normalized_weights = weights.softmax(dim = 1)
|
||||
@@ -60,7 +58,7 @@ class WeightedObjectiveGaussianDiffusion(GaussianDiffusion):
|
||||
noise = default(noise, lambda: torch.randn_like(x_start))
|
||||
x_t = self.q_sample(x_start = x_start, t = t, noise = noise)
|
||||
|
||||
model_output = self.model(x_t, t)
|
||||
model_output = self.denoise_fn(x_t, t)
|
||||
pred_noise, pred_x_start, weights = model_output.split(self.split_dims, dim = 1)
|
||||
|
||||
# get loss for predicted noise and x_start
|
||||
|
||||
|
Before Width: | Height: | Size: 842 KiB After Width: | Height: | Size: 842 KiB |
@@ -3,7 +3,7 @@ from setuptools import setup, find_packages
|
||||
setup(
|
||||
name = 'denoising-diffusion-pytorch',
|
||||
packages = find_packages(),
|
||||
version = '0.31.0',
|
||||
version = '0.16.2',
|
||||
license='MIT',
|
||||
description = 'Denoising Diffusion Probabilistic Models - Pytorch',
|
||||
author = 'Phil Wang',
|
||||
@@ -15,9 +15,7 @@ setup(
|
||||
'generative models'
|
||||
],
|
||||
install_requires=[
|
||||
'accelerate',
|
||||
'einops',
|
||||
'ema-pytorch',
|
||||
'pillow',
|
||||
'torch',
|
||||
'torchvision',
|
||||
@@ -30,4 +28,4 @@ setup(
|
||||
'License :: OSI Approved :: MIT License',
|
||||
'Programming Language :: Python :: 3.6',
|
||||
],
|
||||
)
|
||||
)
|
||||
Reference in New Issue
Block a user