| 97 | return log_probs |
| 98 | |
| 99 | class GaussianDiffusion: |
| 100 | def __init__(self,betas, loss_type, model_mean_type, model_var_type): |
| 101 | self.loss_type = loss_type |
| 102 | self.model_mean_type = model_mean_type |
| 103 | self.model_var_type = model_var_type |
| 104 | assert isinstance(betas, np.ndarray) |
| 105 | self.np_betas = betas = betas.astype(np.float64) # computations here in float64 for accuracy |
| 106 | assert (betas > 0).all() and (betas <= 1).all() |
| 107 | timesteps, = betas.shape |
| 108 | self.num_timesteps = int(timesteps) |
| 109 | |
| 110 | # initialize twice the actual length so we can keep running for eval |
| 111 | # betas = np.concatenate([betas, np.full_like(betas[:int(0.2*len(betas))], betas[-1])]) |
| 112 | |
| 113 | alphas = 1. - betas |
| 114 | alphas_cumprod = torch.from_numpy(np.cumprod(alphas, axis=0)).float() |
| 115 | alphas_cumprod_prev = torch.from_numpy(np.append(1., alphas_cumprod[:-1])).float() |
| 116 | |
| 117 | self.betas = torch.from_numpy(betas).float() |
| 118 | self.alphas_cumprod = alphas_cumprod.float() |
| 119 | self.alphas_cumprod_prev = alphas_cumprod_prev.float() |
| 120 | |
| 121 | # calculations for diffusion q(x_t | x_{t-1}) and others |
| 122 | self.sqrt_alphas_cumprod = torch.sqrt(alphas_cumprod).float() |
| 123 | self.sqrt_one_minus_alphas_cumprod = torch.sqrt(1. - alphas_cumprod).float() |
| 124 | self.log_one_minus_alphas_cumprod = torch.log(1. - alphas_cumprod).float() |
| 125 | self.sqrt_recip_alphas_cumprod = torch.sqrt(1. / alphas_cumprod).float() |
| 126 | self.sqrt_recipm1_alphas_cumprod = torch.sqrt(1. / alphas_cumprod - 1).float() |
| 127 | |
| 128 | betas = torch.from_numpy(betas).float() |
| 129 | alphas = torch.from_numpy(alphas).float() |
| 130 | # calculations for posterior q(x_{t-1} | x_t, x_0) |
| 131 | posterior_variance = betas * (1. - alphas_cumprod_prev) / (1. - alphas_cumprod) |
| 132 | # above: equal to 1. / (1. / (1. - alpha_cumprod_tm1) + alpha_t / beta_t) |
| 133 | self.posterior_variance = posterior_variance |
| 134 | # below: log calculation clipped because the posterior variance is 0 at the beginning of the diffusion chain |
| 135 | self.posterior_log_variance_clipped = torch.log(torch.max(posterior_variance, 1e-20 * torch.ones_like(posterior_variance))) |
| 136 | self.posterior_mean_coef1 = betas * torch.sqrt(alphas_cumprod_prev) / (1. - alphas_cumprod) |
| 137 | self.posterior_mean_coef2 = (1. - alphas_cumprod_prev) * torch.sqrt(alphas) / (1. - alphas_cumprod) |
| 138 | |
| 139 | @staticmethod |
| 140 | def _extract(a, t, x_shape): |
| 141 | """ |
| 142 | Extract some coefficients at specified timesteps, |
| 143 | then reshape to [batch_size, 1, 1, 1, 1, ...] for broadcasting purposes. |
| 144 | """ |
| 145 | bs, = t.shape |
| 146 | assert x_shape[0] == bs |
| 147 | out = torch.gather(a, 0, t) |
| 148 | assert out.shape == torch.Size([bs]) |
| 149 | return torch.reshape(out, [bs] + ((len(x_shape) - 1) * [1])) |
| 150 | |
| 151 | |
| 152 | |
| 153 | def q_mean_variance(self, x_start, t): |
| 154 | mean = self._extract(self.sqrt_alphas_cumprod.to(x_start.device), t, x_start.shape) * x_start |
| 155 | variance = self._extract(1. - self.alphas_cumprod.to(x_start.device), t, x_start.shape) |
| 156 | log_variance = self._extract(self.log_one_minus_alphas_cumprod.to(x_start.device), t, x_start.shape) |