normal.py 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124
  1. # mypy: allow-untyped-defs
  2. import math
  3. from typing import Optional, Union
  4. import torch
  5. from torch import Tensor
  6. from torch.distributions import constraints
  7. from torch.distributions.exp_family import ExponentialFamily
  8. from torch.distributions.utils import _standard_normal, broadcast_all
  9. from torch.types import _Number, _size
  10. __all__ = ["Normal"]
  11. class Normal(ExponentialFamily):
  12. r"""
  13. Creates a normal (also called Gaussian) distribution parameterized by
  14. :attr:`loc` and :attr:`scale`.
  15. Example::
  16. >>> # xdoctest: +IGNORE_WANT("non-deterministic")
  17. >>> m = Normal(torch.tensor([0.0]), torch.tensor([1.0]))
  18. >>> m.sample() # normally distributed with loc=0 and scale=1
  19. tensor([ 0.1046])
  20. Args:
  21. loc (float or Tensor): mean of the distribution (often referred to as mu)
  22. scale (float or Tensor): standard deviation of the distribution
  23. (often referred to as sigma)
  24. """
  25. # pyrefly: ignore [bad-override]
  26. arg_constraints = {"loc": constraints.real, "scale": constraints.positive}
  27. support = constraints.real
  28. has_rsample = True
  29. _mean_carrier_measure = 0
  30. @property
  31. def mean(self) -> Tensor:
  32. return self.loc
  33. @property
  34. def mode(self) -> Tensor:
  35. return self.loc
  36. @property
  37. def stddev(self) -> Tensor:
  38. return self.scale
  39. @property
  40. def variance(self) -> Tensor:
  41. return self.stddev.pow(2)
  42. def __init__(
  43. self,
  44. loc: Union[Tensor, float],
  45. scale: Union[Tensor, float],
  46. validate_args: Optional[bool] = None,
  47. ) -> None:
  48. self.loc, self.scale = broadcast_all(loc, scale)
  49. if isinstance(loc, _Number) and isinstance(scale, _Number):
  50. batch_shape = torch.Size()
  51. else:
  52. batch_shape = self.loc.size()
  53. super().__init__(batch_shape, validate_args=validate_args)
  54. def expand(self, batch_shape, _instance=None):
  55. new = self._get_checked_instance(Normal, _instance)
  56. batch_shape = torch.Size(batch_shape)
  57. new.loc = self.loc.expand(batch_shape)
  58. new.scale = self.scale.expand(batch_shape)
  59. super(Normal, new).__init__(batch_shape, validate_args=False)
  60. new._validate_args = self._validate_args
  61. return new
  62. def sample(self, sample_shape=torch.Size()):
  63. shape = self._extended_shape(sample_shape)
  64. with torch.no_grad():
  65. return torch.normal(self.loc.expand(shape), self.scale.expand(shape))
  66. def rsample(self, sample_shape: _size = torch.Size()) -> Tensor:
  67. shape = self._extended_shape(sample_shape)
  68. eps = _standard_normal(shape, dtype=self.loc.dtype, device=self.loc.device)
  69. return self.loc + eps * self.scale
  70. def log_prob(self, value):
  71. if self._validate_args:
  72. self._validate_sample(value)
  73. # compute the variance
  74. # pyrefly: ignore [unsupported-operation]
  75. var = self.scale**2
  76. log_scale = (
  77. math.log(self.scale)
  78. if isinstance(self.scale, _Number)
  79. else self.scale.log()
  80. )
  81. return (
  82. -((value - self.loc) ** 2) / (2 * var)
  83. - log_scale
  84. - math.log(math.sqrt(2 * math.pi))
  85. )
  86. def cdf(self, value):
  87. if self._validate_args:
  88. self._validate_sample(value)
  89. return 0.5 * (
  90. 1 + torch.erf((value - self.loc) * self.scale.reciprocal() / math.sqrt(2))
  91. )
  92. def icdf(self, value):
  93. return self.loc + self.scale * torch.erfinv(2 * value - 1) * math.sqrt(2)
  94. def entropy(self):
  95. return 0.5 + 0.5 * math.log(2 * math.pi) + torch.log(self.scale)
  96. @property
  97. def _natural_params(self) -> tuple[Tensor, Tensor]:
  98. return (self.loc / self.scale.pow(2), -0.5 * self.scale.pow(2).reciprocal())
  99. # pyrefly: ignore [bad-override]
  100. def _log_normalizer(self, x, y):
  101. return -0.25 * x.pow(2) / y + 0.5 * torch.log(-math.pi / y)