"""Tensorflow Tensor Library Module."""
import logging
import tensorflow as tf
import tensorflow_probability as tfp
log = logging.getLogger(__name__)
[docs]
class tensorflow_backend:
"""TensorFlow backend for pyhf"""
__slots__ = ['name', 'precision', 'dtypemap', 'default_do_grad']
#: The array type for tensorflow
array_type = tf.Tensor
#: The array content type for tensorflow
array_subtype = tf.Tensor
[docs]
def __init__(self, **kwargs):
self.name = 'tensorflow'
self.precision = kwargs.get('precision', '64b')
self.dtypemap = {
'float': tf.float64 if self.precision == '64b' else tf.float32,
'int': tf.int64 if self.precision == '64b' else tf.int32,
'bool': tf.bool,
}
self.default_do_grad = True
[docs]
def _setup(self):
"""
Run any global setups for the tensorflow lib.
"""
[docs]
def clip(self, tensor_in, min_value, max_value):
"""
Clips (limits) the tensor values to be within a specified min and max.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> a = pyhf.tensorlib.astensor([-2, -1, 0, 1, 2])
>>> t = pyhf.tensorlib.clip(a, -1, 1)
>>> print(t)
tf.Tensor([-1. -1. 0. 1. 1.], shape=(5,), dtype=float64)
Args:
tensor_in (:obj:`tensor`): The input tensor object
min_value (:obj:`scalar` or :obj:`tensor` or :obj:`None`): The minimum value to be clipped to
max_value (:obj:`scalar` or :obj:`tensor` or :obj:`None`): The maximum value to be clipped to
Returns:
TensorFlow Tensor: A clipped `tensor`
"""
if min_value is None:
min_value = tf.reduce_min(tensor_in)
if max_value is None:
max_value = tf.reduce_max(tensor_in)
return tf.clip_by_value(tensor_in, min_value, max_value)
[docs]
def erf(self, tensor_in):
"""
The error function of complex argument.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> a = pyhf.tensorlib.astensor([-2., -1., 0., 1., 2.])
>>> t = pyhf.tensorlib.erf(a)
>>> print(t)
tf.Tensor([-0.99532227 -0.84270079 0. 0.84270079 0.99532227], shape=(5,), dtype=float64)
Args:
tensor_in (:obj:`tensor`): The input tensor object
Returns:
TensorFlow Tensor: The values of the error function at the given points.
"""
return tf.math.erf(tensor_in)
[docs]
def erfinv(self, tensor_in):
"""
The inverse of the error function of complex argument.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> a = pyhf.tensorlib.astensor([-2., -1., 0., 1., 2.])
>>> t = pyhf.tensorlib.erfinv(pyhf.tensorlib.erf(a))
>>> print(t)
tf.Tensor([-2. -1. 0. 1. 2.], shape=(5,), dtype=float64)
Args:
tensor_in (:obj:`tensor`): The input tensor object
Returns:
TensorFlow Tensor: The values of the inverse of the error function at the given points.
"""
return tf.math.erfinv(tensor_in)
[docs]
def tile(self, tensor_in, repeats):
"""
Repeat tensor data along a specific dimension
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> a = pyhf.tensorlib.astensor([[1.0], [2.0]])
>>> t = pyhf.tensorlib.tile(a, (1, 2))
>>> print(t)
tf.Tensor(
[[1. 1.]
[2. 2.]], shape=(2, 2), dtype=float64)
Args:
tensor_in (:obj:`tensor`): The tensor to be repeated
repeats (:obj:`tensor`): The tuple of multipliers for each dimension
Returns:
TensorFlow Tensor: The tensor with repeated axes
"""
try:
return tf.tile(tensor_in, repeats)
except tf.errors.InvalidArgumentError:
shape = tf.shape(tensor_in).numpy().tolist()
diff = len(repeats) - len(shape)
if diff < 0:
raise
return tf.tile(tf.reshape(tensor_in, [1] * diff + shape), repeats)
[docs]
def conditional(self, predicate, true_callable, false_callable):
"""
Runs a callable conditional on the boolean value of the evaluation of a predicate
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> tensorlib = pyhf.tensorlib
>>> a = tensorlib.astensor([4])
>>> b = tensorlib.astensor([5])
>>> t = tensorlib.conditional((a < b)[0], lambda: a + b, lambda: a - b)
>>> print(t)
tf.Tensor([9.], shape=(1,), dtype=float64)
Args:
predicate (:obj:`scalar`): The logical condition that determines which callable to evaluate
true_callable (:obj:`callable`): The callable that is evaluated when the :code:`predicate` evaluates to :code:`true`
false_callable (:obj:`callable`): The callable that is evaluated when the :code:`predicate` evaluates to :code:`false`
Returns:
TensorFlow Tensor: The output of the callable that was evaluated
"""
return tf.cond(predicate, true_callable, false_callable)
[docs]
def tolist(self, tensor_in):
try:
return tensor_in.numpy().tolist()
except AttributeError:
if isinstance(tensor_in, list):
return tensor_in
raise
[docs]
def outer(self, tensor_in_1, tensor_in_2):
dtype = self.dtypemap["float"]
tensor_in_1 = (
tensor_in_1 if tensor_in_1.dtype != tf.bool else tf.cast(tensor_in_1, dtype)
)
tensor_in_1 = (
tensor_in_1 if tensor_in_2.dtype != tf.bool else tf.cast(tensor_in_2, dtype)
)
return tf.einsum('i,j->ij', tensor_in_1, tensor_in_2)
[docs]
def gather(self, tensor, indices):
return tf.compat.v2.gather(tensor, indices)
[docs]
def boolean_mask(self, tensor, mask):
return tf.boolean_mask(tensor, mask)
[docs]
def isfinite(self, tensor):
return tf.math.is_finite(tensor)
[docs]
def astensor(self, tensor_in, dtype='float'):
"""
Convert to a TensorFlow Tensor.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> tensor = pyhf.tensorlib.astensor([[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]])
>>> tensor
<tf.Tensor: shape=(2, 3), dtype=float64, numpy=
array([[1., 2., 3.],
[4., 5., 6.]])>
>>> type(tensor)
<class 'tensorflow.python.framework.ops.EagerTensor'>
Args:
tensor_in (Number or Tensor): Tensor object
Returns:
`tf.Tensor`: A symbolic handle to one of the outputs of a `tf.Operation`.
"""
try:
dtype = self.dtypemap[dtype]
except KeyError:
log.error(
'Invalid dtype: dtype must be float, int, or bool.', exc_info=True
)
raise
tensor = tensor_in
# If already a tensor then done
try:
# Use a tensor attribute that isn't meaningless when eager execution is enabled
tensor.device
except AttributeError:
tensor = tf.convert_to_tensor(tensor_in)
if tensor.dtype is not dtype:
tensor = tf.cast(tensor, dtype)
return tensor
[docs]
def sum(self, tensor_in, axis=None):
return (
tf.reduce_sum(tensor_in)
if (axis is None or tensor_in.shape == tf.TensorShape([]))
else tf.reduce_sum(tensor_in, axis)
)
[docs]
def product(self, tensor_in, axis=None):
return (
tf.reduce_prod(tensor_in)
if axis is None
else tf.reduce_prod(tensor_in, axis)
)
[docs]
def abs(self, tensor):
return tf.abs(tensor)
[docs]
def ones(self, shape, dtype="float"):
try:
dtype = self.dtypemap[dtype]
except KeyError:
log.error(
f"Invalid dtype: dtype must be one of {list(self.dtypemap)}.",
exc_info=True,
)
raise
return tf.ones(shape, dtype=dtype)
[docs]
def zeros(self, shape, dtype="float"):
try:
dtype = self.dtypemap[dtype]
except KeyError:
log.error(
f"Invalid dtype: dtype must be one of {list(self.dtypemap)}.",
exc_info=True,
)
raise
return tf.zeros(shape, dtype=dtype)
[docs]
def power(self, tensor_in_1, tensor_in_2):
return tf.pow(tensor_in_1, tensor_in_2)
[docs]
def sqrt(self, tensor_in):
return tf.sqrt(tensor_in)
[docs]
def shape(self, tensor):
return tuple(map(int, tensor.shape))
[docs]
def reshape(self, tensor, newshape):
return tf.reshape(tensor, newshape)
[docs]
def ravel(self, tensor):
"""
Return a flattened view of the tensor, not a copy.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> tensor = pyhf.tensorlib.astensor([[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]])
>>> t_ravel = pyhf.tensorlib.ravel(tensor)
>>> print(t_ravel)
tf.Tensor([1. 2. 3. 4. 5. 6.], shape=(6,), dtype=float64)
Args:
tensor (Tensor): Tensor object
Returns:
`tf.Tensor`: A flattened array.
"""
return self.reshape(tensor, -1)
[docs]
def divide(self, tensor_in_1, tensor_in_2):
return tf.divide(tensor_in_1, tensor_in_2)
[docs]
def log(self, tensor_in):
return tf.math.log(tensor_in)
[docs]
def exp(self, tensor_in):
return tf.exp(tensor_in)
[docs]
def percentile(self, tensor_in, q, axis=None, interpolation="linear"):
r"""
Compute the :math:`q`-th percentile of the tensor along the specified axis.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> a = pyhf.tensorlib.astensor([[10, 7, 4], [3, 2, 1]])
>>> t = pyhf.tensorlib.percentile(a, 50)
>>> print(t)
tf.Tensor(3.5, shape=(), dtype=float64)
>>> t = pyhf.tensorlib.percentile(a, 50, axis=1)
>>> print(t)
tf.Tensor([7. 2.], shape=(2,), dtype=float64)
Args:
tensor_in (`tensor`): The tensor containing the data
q (:obj:`float` or `tensor`): The :math:`q`-th percentile to compute
axis (`number` or `tensor`): The dimensions along which to compute
interpolation (:obj:`str`): The interpolation method to use when the
desired percentile lies between two data points ``i < j``:
- ``'linear'``: ``i + (j - i) * fraction``, where ``fraction`` is the
fractional part of the index surrounded by ``i`` and ``j``.
- ``'lower'``: ``i``.
- ``'higher'``: ``j``.
- ``'midpoint'``: ``(i + j) / 2``.
- ``'nearest'``: ``i`` or ``j``, whichever is nearest.
Returns:
TensorFlow Tensor: The value of the :math:`q`-th percentile of the tensor along the specified axis.
.. versionadded:: 0.7.0
"""
return tfp.stats.percentile(
tensor_in, q, axis=axis, interpolation=interpolation
)
[docs]
def stack(self, sequence, axis=0):
return tf.stack(sequence, axis=axis)
[docs]
def where(self, mask, tensor_in_1, tensor_in_2):
"""
Apply a boolean selection mask to the elements of the input tensors.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> t = pyhf.tensorlib.where(
... pyhf.tensorlib.astensor([1, 0, 1], dtype='bool'),
... pyhf.tensorlib.astensor([1, 1, 1]),
... pyhf.tensorlib.astensor([2, 2, 2]),
... )
>>> print(t)
tf.Tensor([1. 2. 1.], shape=(3,), dtype=float64)
Args:
mask (bool): Boolean mask (boolean or tensor object of booleans)
tensor_in_1 (Tensor): Tensor object
tensor_in_2 (Tensor): Tensor object
Returns:
TensorFlow Tensor: The result of the mask being applied to the tensors.
"""
return tf.where(mask, tensor_in_1, tensor_in_2)
[docs]
def concatenate(self, sequence, axis=0):
"""
Join a sequence of arrays along an existing axis.
Args:
sequence: sequence of tensors
axis: dimension along which to concatenate
Returns:
output: the concatenated tensor
"""
return tf.concat(sequence, axis=axis)
[docs]
def simple_broadcast(self, *args):
"""
Broadcast a sequence of 1 dimensional arrays.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> b = pyhf.tensorlib.simple_broadcast(
... pyhf.tensorlib.astensor([1]),
... pyhf.tensorlib.astensor([2, 3, 4]),
... pyhf.tensorlib.astensor([5, 6, 7]))
>>> print([str(t) for t in b]) # doctest: +NORMALIZE_WHITESPACE
['tf.Tensor([1. 1. 1.], shape=(3,), dtype=float64)',
'tf.Tensor([2. 3. 4.], shape=(3,), dtype=float64)',
'tf.Tensor([5. 6. 7.], shape=(3,), dtype=float64)']
Args:
args (Array of Tensors): Sequence of arrays
Returns:
list of Tensors: The sequence broadcast together.
"""
max_dim = max(map(tf.size, args))
try:
assert not [arg for arg in args if 1 < tf.size(arg) < max_dim]
except AssertionError:
log.error(
'ERROR: The arguments must be of compatible size: 1 or %i', max_dim
)
raise
return [tf.broadcast_to(arg, (max_dim,)) for arg in args]
[docs]
def einsum(self, subscripts, *operands):
"""
A generalized contraction between tensors of arbitrary dimension.
This function returns a tensor whose elements are defined by equation,
which is written in a shorthand form inspired by the Einstein summation
convention.
Args:
subscripts: str, specifies the subscripts for summation
operands: list of array_like, these are the tensors for the operation
Returns:
TensorFlow Tensor: the calculation based on the Einstein summation convention
"""
return tf.einsum(subscripts, *operands)
[docs]
def poisson_logpdf(self, n, lam):
r"""
The log of the continuous approximation, using :math:`n! = \Gamma\left(n+1\right)`,
to the probability mass function of the Poisson distribution evaluated
at :code:`n` given the parameter :code:`lam`.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> t = pyhf.tensorlib.poisson_logpdf(5., 6.)
>>> print(t) # doctest:+ELLIPSIS
tf.Tensor(-1.82869439..., shape=(), dtype=float64)
>>> values = pyhf.tensorlib.astensor([5., 9.])
>>> rates = pyhf.tensorlib.astensor([6., 8.])
>>> t = pyhf.tensorlib.poisson_logpdf(values, rates)
>>> print(t)
tf.Tensor([-1.8286944 -2.0868536], shape=(2,), dtype=float64)
Args:
n (:obj:`tensor` or :obj:`float`): The value at which to evaluate the approximation to the Poisson distribution p.m.f.
(the observed number of events)
lam (:obj:`tensor` or :obj:`float`): The mean of the Poisson distribution p.m.f.
(the expected number of events)
Returns:
TensorFlow Tensor: Value of the continuous approximation to log(Poisson(n|lam))
"""
lam = self.astensor(lam)
return tfp.distributions.Poisson(lam).log_prob(n)
[docs]
def poisson(self, n, lam):
r"""
The continuous approximation, using :math:`n! = \Gamma\left(n+1\right)`,
to the probability mass function of the Poisson distribution evaluated
at :code:`n` given the parameter :code:`lam`.
.. note::
Though the p.m.f of the Poisson distribution is not defined for
:math:`\lambda = 0`, the limit as :math:`\lambda \to 0` is still
defined, which gives a degenerate p.m.f. of
.. math::
\lim_{\lambda \to 0} \,\mathrm{Pois}(n | \lambda) =
\left\{\begin{array}{ll}
1, & n = 0,\\
0, & n > 0
\end{array}\right.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> t = pyhf.tensorlib.poisson(5., 6.)
>>> print(t) # doctest:+ELLIPSIS
tf.Tensor(0.16062314..., shape=(), dtype=float64)
>>> values = pyhf.tensorlib.astensor([5., 9.])
>>> rates = pyhf.tensorlib.astensor([6., 8.])
>>> t = pyhf.tensorlib.poisson(values, rates)
>>> print(t)
tf.Tensor([0.16062314 0.12407692], shape=(2,), dtype=float64)
Args:
n (:obj:`tensor` or :obj:`float`): The value at which to evaluate the approximation to the Poisson distribution p.m.f.
(the observed number of events)
lam (:obj:`tensor` or :obj:`float`): The mean of the Poisson distribution p.m.f.
(the expected number of events)
Returns:
TensorFlow Tensor: Value of the continuous approximation to Poisson(n|lam)
"""
lam = self.astensor(lam)
return tf.exp(tfp.distributions.Poisson(lam).log_prob(n))
[docs]
def normal_logpdf(self, x, mu, sigma):
r"""
The log of the probability density function of the Normal distribution evaluated
at :code:`x` given parameters of mean of :code:`mu` and standard deviation
of :code:`sigma`.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> t = pyhf.tensorlib.normal_logpdf(0.5, 0., 1.)
>>> print(t) # doctest:+ELLIPSIS
tf.Tensor(-1.04393853..., shape=(), dtype=float64)
>>> values = pyhf.tensorlib.astensor([0.5, 2.0])
>>> means = pyhf.tensorlib.astensor([0., 2.3])
>>> sigmas = pyhf.tensorlib.astensor([1., 0.8])
>>> t = pyhf.tensorlib.normal_logpdf(values, means, sigmas)
>>> print(t)
tf.Tensor([-1.04393853 -0.76610747], shape=(2,), dtype=float64)
Args:
x (:obj:`tensor` or :obj:`float`): The value at which to evaluate the Normal distribution p.d.f.
mu (:obj:`tensor` or :obj:`float`): The mean of the Normal distribution
sigma (:obj:`tensor` or :obj:`float`): The standard deviation of the Normal distribution
Returns:
TensorFlow Tensor: Value of log(Normal(x|mu, sigma))
"""
mu = self.astensor(mu)
sigma = self.astensor(sigma)
return tfp.distributions.Normal(mu, sigma).log_prob(x)
[docs]
def normal(self, x, mu, sigma):
r"""
The probability density function of the Normal distribution evaluated
at :code:`x` given parameters of mean of :code:`mu` and standard deviation
of :code:`sigma`.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> t = pyhf.tensorlib.normal(0.5, 0., 1.)
>>> print(t) # doctest:+ELLIPSIS
tf.Tensor(0.35206532..., shape=(), dtype=float64)
>>> values = pyhf.tensorlib.astensor([0.5, 2.0])
>>> means = pyhf.tensorlib.astensor([0., 2.3])
>>> sigmas = pyhf.tensorlib.astensor([1., 0.8])
>>> t = pyhf.tensorlib.normal(values, means, sigmas)
>>> print(t)
tf.Tensor([0.35206533 0.46481887], shape=(2,), dtype=float64)
Args:
x (:obj:`tensor` or :obj:`float`): The value at which to evaluate the Normal distribution p.d.f.
mu (:obj:`tensor` or :obj:`float`): The mean of the Normal distribution
sigma (:obj:`tensor` or :obj:`float`): The standard deviation of the Normal distribution
Returns:
TensorFlow Tensor: Value of Normal(x|mu, sigma)
"""
mu = self.astensor(mu)
sigma = self.astensor(sigma)
return tfp.distributions.Normal(mu, sigma).prob(x)
[docs]
def normal_cdf(self, x, mu=0.0, sigma=1):
"""
Compute the value of cumulative distribution function for the Normal distribution at x.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> t = pyhf.tensorlib.normal_cdf(0.8)
>>> print(t) # doctest:+ELLIPSIS
tf.Tensor(0.78814460..., shape=(), dtype=float64)
>>> values = pyhf.tensorlib.astensor([0.8, 2.0])
>>> t = pyhf.tensorlib.normal_cdf(values)
>>> print(t)
tf.Tensor([0.7881446 0.97724987], shape=(2,), dtype=float64)
Args:
x (:obj:`tensor` or :obj:`float`): The observed value of the random variable to evaluate the CDF for
mu (:obj:`tensor` or :obj:`float`): The mean of the Normal distribution
sigma (:obj:`tensor` or :obj:`float`): The standard deviation of the Normal distribution
Returns:
TensorFlow Tensor: The CDF
"""
mu = self.astensor(mu)
sigma = self.astensor(sigma)
return tfp.distributions.Normal(mu, sigma).cdf(x)
[docs]
def poisson_dist(self, rate):
r"""
Construct a Poisson distribution with rate parameter :code:`rate`.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> rates = pyhf.tensorlib.astensor([5, 8])
>>> values = pyhf.tensorlib.astensor([4, 9])
>>> poissons = pyhf.tensorlib.poisson_dist(rates)
>>> t = poissons.log_prob(values)
>>> print(t)
tf.Tensor([-1.74030218 -2.0868536 ], shape=(2,), dtype=float64)
Args:
rate (:obj:`tensor` or :obj:`float`): The mean of the Poisson distribution (the expected number of events)
Returns:
TensorFlow Probability Poisson distribution: The Poisson distribution class
"""
rate = self.astensor(rate)
return tfp.distributions.Poisson(rate)
[docs]
def normal_dist(self, mu, sigma):
r"""
Construct a Normal distribution with mean :code:`mu` and standard deviation :code:`sigma`.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> means = pyhf.tensorlib.astensor([5, 8])
>>> stds = pyhf.tensorlib.astensor([1, 0.5])
>>> values = pyhf.tensorlib.astensor([4, 9])
>>> normals = pyhf.tensorlib.normal_dist(means, stds)
>>> t = normals.log_prob(values)
>>> print(t)
tf.Tensor([-1.41893853 -2.22579135], shape=(2,), dtype=float64)
Args:
mu (:obj:`tensor` or :obj:`float`): The mean of the Normal distribution
sigma (:obj:`tensor` or :obj:`float`): The standard deviation of the Normal distribution
Returns:
TensorFlow Probability Normal distribution: The Normal distribution class
"""
mu = self.astensor(mu)
sigma = self.astensor(sigma)
return tfp.distributions.Normal(mu, sigma)
[docs]
def to_numpy(self, tensor_in):
"""
Convert the TensorFlow tensor to a :class:`numpy.ndarray`.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> tensor = pyhf.tensorlib.astensor([[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]])
>>> print(tensor)
tf.Tensor(
[[1. 2. 3.]
[4. 5. 6.]], shape=(2, 3), dtype=float64)
>>> numpy_ndarray = pyhf.tensorlib.to_numpy(tensor)
>>> numpy_ndarray
array([[1., 2., 3.],
[4., 5., 6.]])
>>> type(numpy_ndarray)
<class 'numpy.ndarray'>
Args:
tensor_in (:obj:`tensor`): The input tensor object.
Returns:
:class:`numpy.ndarray`: The tensor converted to a NumPy ``ndarray``.
"""
return tensor_in.numpy()
[docs]
def transpose(self, tensor_in):
"""
Transpose the tensor.
Example:
>>> import pyhf
>>> pyhf.set_backend("tensorflow")
>>> tensor = pyhf.tensorlib.astensor([[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]])
>>> print(tensor)
tf.Tensor(
[[1. 2. 3.]
[4. 5. 6.]], shape=(2, 3), dtype=float64)
>>> tensor_T = pyhf.tensorlib.transpose(tensor)
>>> print(tensor_T)
tf.Tensor(
[[1. 4.]
[2. 5.]
[3. 6.]], shape=(3, 2), dtype=float64)
Args:
tensor_in (:obj:`tensor`): The input tensor object.
Returns:
TensorFlow Tensor: The transpose of the input tensor.
.. versionadded:: 0.7.0
"""
return tf.transpose(tensor_in)