bernoulli.py 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141
  1. # mypy: allow-untyped-defs
  2. from typing import Optional, Union
  3. import torch
  4. from torch import nan, Tensor
  5. from torch.distributions import constraints
  6. from torch.distributions.exp_family import ExponentialFamily
  7. from torch.distributions.utils import (
  8. broadcast_all,
  9. lazy_property,
  10. logits_to_probs,
  11. probs_to_logits,
  12. )
  13. from torch.nn.functional import binary_cross_entropy_with_logits
  14. from torch.types import _Number, Number
  15. __all__ = ["Bernoulli"]
  16. class Bernoulli(ExponentialFamily):
  17. r"""
  18. Creates a Bernoulli distribution parameterized by :attr:`probs`
  19. or :attr:`logits` (but not both).
  20. Samples are binary (0 or 1). They take the value `1` with probability `p`
  21. and `0` with probability `1 - p`.
  22. Example::
  23. >>> # xdoctest: +IGNORE_WANT("non-deterministic")
  24. >>> m = Bernoulli(torch.tensor([0.3]))
  25. >>> m.sample() # 30% chance 1; 70% chance 0
  26. tensor([ 0.])
  27. Args:
  28. probs (Number, Tensor): the probability of sampling `1`
  29. logits (Number, Tensor): the log-odds of sampling `1`
  30. validate_args (bool, optional): whether to validate arguments, None by default
  31. """
  32. arg_constraints = {"probs": constraints.unit_interval, "logits": constraints.real}
  33. support = constraints.boolean
  34. has_enumerate_support = True
  35. _mean_carrier_measure = 0
  36. def __init__(
  37. self,
  38. probs: Optional[Union[Tensor, Number]] = None,
  39. logits: Optional[Union[Tensor, Number]] = None,
  40. validate_args: Optional[bool] = None,
  41. ) -> None:
  42. if (probs is None) == (logits is None):
  43. raise ValueError(
  44. "Either `probs` or `logits` must be specified, but not both."
  45. )
  46. if probs is not None:
  47. is_scalar = isinstance(probs, _Number)
  48. (self.probs,) = broadcast_all(probs)
  49. else:
  50. assert logits is not None # helps mypy
  51. is_scalar = isinstance(logits, _Number)
  52. (self.logits,) = broadcast_all(logits)
  53. self._param = self.probs if probs is not None else self.logits
  54. if is_scalar:
  55. batch_shape = torch.Size()
  56. else:
  57. batch_shape = self._param.size()
  58. super().__init__(batch_shape, validate_args=validate_args)
  59. def expand(self, batch_shape, _instance=None):
  60. new = self._get_checked_instance(Bernoulli, _instance)
  61. batch_shape = torch.Size(batch_shape)
  62. if "probs" in self.__dict__:
  63. new.probs = self.probs.expand(batch_shape)
  64. new._param = new.probs
  65. if "logits" in self.__dict__:
  66. new.logits = self.logits.expand(batch_shape)
  67. new._param = new.logits
  68. super(Bernoulli, new).__init__(batch_shape, validate_args=False)
  69. new._validate_args = self._validate_args
  70. return new
  71. def _new(self, *args, **kwargs):
  72. return self._param.new(*args, **kwargs)
  73. @property
  74. def mean(self) -> Tensor:
  75. return self.probs
  76. @property
  77. def mode(self) -> Tensor:
  78. mode = (self.probs >= 0.5).to(self.probs)
  79. mode[self.probs == 0.5] = nan
  80. return mode
  81. @property
  82. def variance(self) -> Tensor:
  83. return self.probs * (1 - self.probs)
  84. @lazy_property
  85. def logits(self) -> Tensor:
  86. return probs_to_logits(self.probs, is_binary=True)
  87. @lazy_property
  88. def probs(self) -> Tensor:
  89. return logits_to_probs(self.logits, is_binary=True)
  90. @property
  91. def param_shape(self) -> torch.Size:
  92. return self._param.size()
  93. def sample(self, sample_shape=torch.Size()):
  94. shape = self._extended_shape(sample_shape)
  95. with torch.no_grad():
  96. return torch.bernoulli(self.probs.expand(shape))
  97. def log_prob(self, value):
  98. if self._validate_args:
  99. self._validate_sample(value)
  100. logits, value = broadcast_all(self.logits, value)
  101. return -binary_cross_entropy_with_logits(logits, value, reduction="none")
  102. def entropy(self):
  103. return binary_cross_entropy_with_logits(
  104. self.logits, self.probs, reduction="none"
  105. )
  106. def enumerate_support(self, expand=True):
  107. values = torch.arange(2, dtype=self._param.dtype, device=self._param.device)
  108. values = values.view((-1,) + (1,) * len(self._batch_shape))
  109. if expand:
  110. values = values.expand((-1,) + self._batch_shape)
  111. return values
  112. @property
  113. def _natural_params(self) -> tuple[Tensor]:
  114. return (torch.logit(self.probs),)
  115. def _log_normalizer(self, x):
  116. return torch.log1p(torch.exp(x))