from typing import List import torch import numpy as np class DiagonalGaussianDistribution(object): def __init__(self, parameters, deterministic=False): self.parameters = parameters self.mean, self.logvar = torch.chunk(parameters, 2, dim=1) self.logvar = torch.clamp(self.logvar, -30.0, 20.0) self.deterministic = deterministic self.std = torch.exp(0.5 * self.logvar) self.var = torch.exp(self.logvar) if self.deterministic: self.var = self.std = torch.zeros_like(self.mean).to( device=self.parameters.device ) def sample(self): x = self.mean + self.std * torch.randn(self.mean.shape).to( device=self.parameters.device ) return x def kl(self, other=None): if self.deterministic: return torch.Tensor([0.0]) else: dim = [1, 2, 3] if self.mean.dim() == 4 else [1, 3, 4] # BCHW or BCTHW if other is None: return 0.5 * torch.sum( torch.pow(self.mean, 2) + self.var - 1.0 - self.logvar, dim=dim, ) else: return 0.5 * torch.sum( torch.pow(self.mean - other.mean, 2) / other.var + self.var / other.var - 1.0 - self.logvar + other.logvar, dim=dim, ) def nll(self, sample, dims=[1, 2, 3]): if self.deterministic: return torch.Tensor([0.0]) logtwopi = np.log(2.0 * np.pi) return 0.5 * torch.sum( logtwopi + self.logvar + torch.pow(sample - self.mean, 2) / self.var, dim=dims, ) def mode(self): return self.mean @classmethod def cat(cls, distributions: List["DiagonalGaussianDistribution"], dim: int = 0): """ Concatenates a list of DiagonalGaussianDistributions along the batch dimension. """ parameters = torch.cat([dist.parameters for dist in distributions], dim=dim) all_deterministic = all([dist.deterministic for dist in distributions]) any_deterministic = any([dist.deterministic for dist in distributions]) assert ( all_deterministic == any_deterministic ), "`deterministic` must be the same for all distributions when concatenating." return cls(parameters, deterministic=all_deterministic)