code: # Copyright 2018 The TensorFlow Probability Authors.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  # ============================================================================
  """Tests for initializers."""
  
  from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  
  import importlib
  # Dependency imports
  import numpy as np
  import tensorflow as tf
  import tensorflow_probability as tfp
  
  
  def try_import(name):  # pylint: disable=invalid-name
    module = None
    try:
      module = importlib.import_module(name)
    except ImportError as e:
      tf.logging.warning("Could not import %s: %s" % (name, str(e)))
    return module
  
  stats = try_import("scipy.stats")
  
  tfd = tfp.distributions
  
  
  class HalfNormalTest(tf.test.TestCase):
  
    def setUp(self):
      self._rng = np.random.RandomState(123)
  
    def assertAllFinite(self, tensor):
      is_finite = np.isfinite(self.evaluate(tensor))
      all_true = np.ones_like(is_finite, dtype=np.bool)
      self.assertAllEqual(all_true, is_finite)
  
    def _testParamShapes(self, sample_shape, expected):
      param_shapes = tfd.HalfNormal.param_shapes(sample_shape)
      scale_shape = param_shapes["scale"]
      self.assertAllEqual(expected, self.evaluate(scale_shape))
      scale = tf.ones(scale_shape)
      self.assertAllEqual(expected,
                          self.evaluate(tf.shape(tfd.HalfNormal(scale).sample())))
  
    def _testParamStaticShapes(self, sample_shape, expected):
      param_shapes = tfd.HalfNormal.param_static_shapes(sample_shape)
      scale_shape = param_shapes["scale"]
      self.assertEqual(expected, scale_shape)
  
    def _testBatchShapes(self, dist, tensor):
      self.assertAllEqual(self.evaluate(dist.batch_shape_tensor()), tensor.shape)
      self.assertAllEqual(
          self.evaluate(dist.batch_shape_tensor()), self.evaluate(tensor).shape)
      self.assertAllEqual(dist.batch_shape, tensor.shape)
      self.assertAllEqual(dist.batch_shape, self.evaluate(tensor).shape)
  
    def testParamShapes(self):
      sample_shape = [10, 3, 4]
      self._testParamShapes(sample_shape, sample_shape)
      self._testParamShapes(tf.constant(sample_shape), sample_shape)
  
    def testParamStaticShapes(self):
      sample_shape = [10, 3, 4]
      self._testParamStaticShapes(sample_shape, sample_shape)
      self._testParamStaticShapes(tf.TensorShape(sample_shape), sample_shape)
  
    def testHalfNormalLogPDF(self):
      batch_size = 6
      scale = tf.constant([3.0] * batch_size)
      x = np.array([-2.5, 2.5, 4.0, 0.0, -1.0, 2.0], dtype=np.float32)
      halfnorm = tfd.HalfNormal(scale=scale)
  
      log_pdf = halfnorm.log_prob(x)
      self._testBatchShapes(halfnorm, log_pdf)
  
      pdf = halfnorm.prob(x)
      self._testBatchShapes(halfnorm, pdf)
  
      if not stats:
        return
      expected_log_pdf = stats.halfnorm(scale=self.evaluate(scale)).logpdf(x)
      self.assertAllClose(expected_log_pdf, self.evaluate(log_pdf))
      self.assertAllClose(np.exp(expected_log_pdf), self.evaluate(pdf))
  
    def testHalfNormalLogPDFMultidimensional(self):
      batch_size = 6
      scale = tf.constant([[3.0, 1.0]] * batch_size)
      x = np.array([[-2.5, 2.5, 4.0, 0.0, -1.0, 2.0]], dtype=np.float32).T
      halfnorm = tfd.HalfNormal(scale=scale)
  
      log_pdf = halfnorm.log_prob(x)
      self._testBatchShapes(halfnorm, log_pdf)
  
      pdf = halfnorm.prob(x)
      self._testBatchShapes(halfnorm, pdf)
  
      if not stats:
        return
      expected_log_pdf = stats.halfnorm(scale=self.evaluate(scale)).logpdf(x)
      self.assertAllClose(expected_log_pdf, self.evaluate(log_pdf))
      self.assertAllClose(np.exp(expected_log_pdf), self.evaluate(pdf))
  
    def testHalfNormalCDF(self):
      batch_size = 50
      scale = self._rng.rand(batch_size) + 1.0
      x = np.linspace(-8.0, 8.0, batch_size).astype(np.float64)
      halfnorm = tfd.HalfNormal(scale=scale)
  
      cdf = halfnorm.cdf(x)
      self._testBatchShapes(halfnorm, cdf)
  
      log_cdf = halfnorm.log_cdf(x)
      self._testBatchShapes(halfnorm, log_cdf)
  
      if not stats:
        return
      expected_logcdf = stats.halfnorm(scale=scale).logcdf(x)
      self.assertAllClose(expected_logcdf, self.evaluate(log_cdf), atol=0)
      self.assertAllClose(np.exp(expected_logcdf), self.evaluate(cdf), atol=0)
  
    def testHalfNormalSurvivalFunction(self):
      batch_size = 50
      scale = self._rng.rand(batch_size) + 1.0
      x = np.linspace(-8.0, 8.0, batch_size).astype(np.float64)
      halfnorm = tfd.HalfNormal(scale=scale)
  
      sf = halfnorm.survival_function(x)
      self._testBatchShapes(halfnorm, sf)
  
      log_sf = halfnorm.log_survival_function(x)
      self._testBatchShapes(halfnorm, log_sf)
  
      if not stats:
        return
      expected_logsf = stats.halfnorm(scale=scale).logsf(x)
      self.assertAllClose(expected_logsf, self.evaluate(log_sf), atol=0)
      self.assertAllClose(np.exp(expected_logsf), self.evaluate(sf), atol=0)
  
    def testHalfNormalQuantile(self):
      batch_size = 50
      scale = self._rng.rand(batch_size) + 1.0
      p = np.linspace(0., 1.0, batch_size).astype(np.float64)
  
      halfnorm = tfd.HalfNormal(scale=scale)
      x = halfnorm.quantile(p)
      self._testBatchShapes(halfnorm, x)
  
      if not stats:
        return
      expected_x = stats.halfnorm(scale=scale).ppf(p)
      self.assertAllClose(expected_x, self.evaluate(x), atol=0)
  
    def testFiniteGradients(self):
      for dtype in [np.float32, np.float64]:
        g = tf.Graph()
        with g.as_default():
          scale = tf.Variable(dtype(3.0))
          dist = tfd.HalfNormal(scale=scale)
          x = np.array([0.01, 0.1, 1., 5., 10.]).astype(dtype)
          for func in [
              dist.cdf, dist.log_cdf, dist.survival_function,
              dist.log_prob, dist.prob, dist.log_survival_function,
          ]:
            print(func.__name__)
            value = func(x)
            with self.test_session(graph=g):
              tf.global_variables_initializer().run()
              grads = tf.gradients(value, [scale])
              self.assertAllFinite(value)
              self.assertAllFinite(grads[0])
  
    def testHalfNormalEntropy(self):
      scale = np.array([[1.0, 2.0, 3.0]])
      halfnorm = tfd.HalfNormal(scale=scale)
  
      # See https://en.wikipedia.org/wiki/Half-normal_distribution for the
      # entropy formula used here.
      expected_entropy = 0.5 * np.log(np.pi * scale**2.0 / 2.0) + 0.5
  
      entropy = halfnorm.entropy()
      self._testBatchShapes(halfnorm, entropy)
      self.assertAllClose(expected_entropy, self.evaluate(entropy))
  
    def testHalfNormalMeanAndMode(self):
      scale = np.array([11., 12., 13.])
  
      halfnorm = tfd.HalfNormal(scale=scale)
      expected_mean = scale * np.sqrt(2.0) / np.sqrt(np.pi)
  
      self.assertAllEqual((3,), self.evaluate(halfnorm.mean()).shape)
      self.assertAllEqual(expected_mean, self.evaluate(halfnorm.mean()))
  
      self.assertAllEqual((3,), self.evaluate(halfnorm.mode()).shape)
      self.assertAllEqual([0., 0., 0.], self.evaluate(halfnorm.mode()))
  
    def testHalfNormalVariance(self):
      scale = np.array([7., 7., 7.])
      halfnorm = tfd.HalfNormal(scale=scale)
      expected_variance = scale**2.0 * (1.0 - 2.0 / np.pi)
  
      self.assertAllEqual((3,), self.evaluate(halfnorm.variance()).shape)
      self.assertAllEqual(expected_variance, self.evaluate(halfnorm.variance()))
  
    def testHalfNormalStandardDeviation(self):
      scale = np.array([7., 7., 7.])
      halfnorm = tfd.HalfNormal(scale=scale)
      expected_variance = scale**2.0 * (1.0 - 2.0 / np.pi)
  
      self.assertAllEqual((3,), halfnorm.stddev().shape)
      self.assertAllEqual(
          np.sqrt(expected_variance), self.evaluate(halfnorm.stddev()))
  
    def testHalfNormalSample(self):
      scale = tf.constant(3.0)
      n = tf.constant(100000)
      halfnorm = tfd.HalfNormal(scale=scale)
  
      sample = halfnorm.sample(n)
  
      self.assertEqual(self.evaluate(sample).shape, (100000,))
      self.assertAllClose(
          self.evaluate(sample).mean(),
          3.0 * np.sqrt(2.0) / np.sqrt(np.pi),
          atol=1e-1)
  
      expected_shape = tf.TensorShape([self.evaluate(n)]).concatenate(
          tf.TensorShape(self.evaluate(halfnorm.batch_shape_tensor())))
      self.assertAllEqual(expected_shape, sample.shape)
      self.assertAllEqual(expected_shape, self.evaluate(sample).shape)
  
      expected_shape_static = (
          tf.TensorShape([self.evaluate(n)]).concatenate(halfnorm.batch_shape))
      self.assertAllEqual(expected_shape_static, sample.shape)
      self.assertAllEqual(expected_shape_static, self.evaluate(sample).shape)
  
    def testHalfNormalSampleMultiDimensional(self):
      batch_size = 2
      scale = tf.constant([[2.0, 3.0]] * batch_size)
      n = tf.constant(100000)
      halfnorm = tfd.HalfNormal(scale=scale)
  
      sample = halfnorm.sample(n)
      self.assertEqual(sample.shape, (100000, batch_size, 2))
      self.assertAllClose(
          self.evaluate(sample)[:, 0, 0].mean(),
          2.0 * np.sqrt(2.0) / np.sqrt(np.pi),
          atol=1e-1)
      self.assertAllClose(
          self.evaluate(sample)[:, 0, 1].mean(),
          3.0 * np.sqrt(2.0) / np.sqrt(np.pi),
          atol=1e-1)
  
      expected_shape = tf.TensorShape([self.evaluate(n)]).concatenate(
          tf.TensorShape(self.evaluate(halfnorm.batch_shape_tensor())))
      self.assertAllEqual(expected_shape, sample.shape)
      self.assertAllEqual(expected_shape, self.evaluate(sample).shape)
  
      expected_shape_static = (
          tf.TensorShape([self.evaluate(n)]).concatenate(halfnorm.batch_shape))
      self.assertAllEqual(expected_shape_static, sample.shape)
      self.assertAllEqual(expected_shape_static, self.evaluate(sample).shape)
  
    def testNegativeSigmaFails(self):
      halfnorm = tfd.HalfNormal(scale=[-5.], validate_args=True, name="G")
      with self.assertRaisesOpError("Condition x > 0 did not hold"):
        self.evaluate(halfnorm.mean())
  
    def testHalfNormalShape(self):
      scale = tf.constant([6.0] * 5)
      halfnorm = tfd.HalfNormal(scale=scale)
  
      self.assertEqual(self.evaluate(halfnorm.batch_shape_tensor()), [5])
      self.assertEqual(halfnorm.batch_shape, tf.TensorShape([5]))
      self.assertAllEqual(self.evaluate(halfnorm.event_shape_tensor()), [])
      self.assertEqual(halfnorm.event_shape, tf.TensorShape([]))
  
    def testHalfNormalShapeWithPlaceholders(self):
      scale = tf.placeholder_with_default(input=[1., 2], shape=None)
      halfnorm = tfd.HalfNormal(scale=scale)
  
      # get_batch_shape should return an "<unknown>" tensor.
      self.assertEqual(halfnorm.batch_shape, tf.TensorShape(None))
      self.assertEqual(halfnorm.event_shape, ())
      self.assertAllEqual(self.evaluate(halfnorm.event_shape_tensor()), [])
      self.assertAllEqual(self.evaluate(halfnorm.batch_shape_tensor()), [2])
  
  
  if __name__ == "__main__":
    tf.test.main()
  
apis: tensorflow.TensorShape tensorflow.Graph tensorflow.constant numpy.ones_like numpy.sqrt numpy.linspace numpy.log tensorflow.placeholder_with_default tensorflow.gradients tensorflow.test.main tensorflow.ones tensorflow.global_variables_initializer numpy.exp numpy.array numpy.random.RandomState
api_extract: [(303, 'tensorflow.test.main', 'tf.test.main', ([], {}), True, 'import tensorflow as tf\n'), (31, 'importlib.import_module', 'importlib.import_module', (['name'], {}), False, 'import importlib\n'), (44, 'numpy.random.RandomState', 'np.random.RandomState', (['(123)'], {}), True, 'import numpy as np\n'), (48, 'numpy.ones_like', 'np.ones_like', (['is_finite'], {'dtype': 'np.bool'}), True, 'import numpy as np\n'), (55, 'tensorflow.ones', 'tf.ones', (['scale_shape'], {}), True, 'import tensorflow as tf\n'), (83, 'tensorflow.constant', 'tf.constant', (['([3.0] * batch_size)'], {}), True, 'import tensorflow as tf\n'), (84, 'numpy.array', 'np.array', (['[-2.5, 2.5, 4.0, 0.0, -1.0, 2.0]'], {'dtype': 'np.float32'}), True, 'import numpy as np\n'), (101, 'tensorflow.constant', 'tf.constant', (['([[3.0, 1.0]] * batch_size)'], {}), True, 'import tensorflow as tf\n'), (187, 'numpy.array', 'np.array', (['[[1.0, 2.0, 3.0]]'], {}), True, 'import numpy as np\n'), (199, 'numpy.array', 'np.array', (['[11.0, 12.0, 13.0]'], {}), True, 'import numpy as np\n'), (211, 'numpy.array', 'np.array', (['[7.0, 7.0, 7.0]'], {}), True, 'import numpy as np\n'), (219, 'numpy.array', 'np.array', (['[7.0, 7.0, 7.0]'], {}), True, 'import numpy as np\n'), (228, 'tensorflow.constant', 'tf.constant', (['(3.0)'], {}), True, 'import tensorflow as tf\n'), (229, 'tensorflow.constant', 'tf.constant', (['(100000)'], {}), True, 'import tensorflow as tf\n'), (252, 'tensorflow.constant', 'tf.constant', (['([[2.0, 3.0]] * batch_size)'], {}), True, 'import tensorflow as tf\n'), (253, 'tensorflow.constant', 'tf.constant', (['(100000)'], {}), True, 'import tensorflow as tf\n'), (283, 'tensorflow.constant', 'tf.constant', (['([6.0] * 5)'], {}), True, 'import tensorflow as tf\n'), (292, 'tensorflow.placeholder_with_default', 'tf.placeholder_with_default', ([], {'input': '[1.0, 2]', 'shape': 'None'}), True, 'import tensorflow as tf\n'), (74, 'tensorflow.constant', 'tf.constant', (['sample_shape'], {}), True, 'import tensorflow as tf\n'), (79, 'tensorflow.TensorShape', 'tf.TensorShape', (['sample_shape'], {}), True, 'import tensorflow as tf\n'), (97, 'numpy.exp', 'np.exp', (['expected_log_pdf'], {}), True, 'import numpy as np\n'), (102, 'numpy.array', 'np.array', (['[[-2.5, 2.5, 4.0, 0.0, -1.0, 2.0]]'], {'dtype': 'np.float32'}), True, 'import numpy as np\n'), (115, 'numpy.exp', 'np.exp', (['expected_log_pdf'], {}), True, 'import numpy as np\n'), (133, 'numpy.exp', 'np.exp', (['expected_logcdf'], {}), True, 'import numpy as np\n'), (151, 'numpy.exp', 'np.exp', (['expected_logsf'], {}), True, 'import numpy as np\n'), (169, 'tensorflow.Graph', 'tf.Graph', ([], {}), True, 'import tensorflow as tf\n'), (202, 'numpy.sqrt', 'np.sqrt', (['np.pi'], {}), True, 'import numpy as np\n'), (225, 'numpy.sqrt', 'np.sqrt', (['expected_variance'], {}), True, 'import numpy as np\n'), (287, 'tensorflow.TensorShape', 'tf.TensorShape', (['[5]'], {}), True, 'import tensorflow as tf\n'), (289, 'tensorflow.TensorShape', 'tf.TensorShape', (['[]'], {}), True, 'import tensorflow as tf\n'), (296, 'tensorflow.TensorShape', 'tf.TensorShape', (['None'], {}), True, 'import tensorflow as tf\n'), (120, 'numpy.linspace', 'np.linspace', (['(-8.0)', '(8.0)', 'batch_size'], {}), True, 'import numpy as np\n'), (138, 'numpy.linspace', 'np.linspace', (['(-8.0)', '(8.0)', 'batch_size'], {}), True, 'import numpy as np\n'), (156, 'numpy.linspace', 'np.linspace', (['(0.0)', '(1.0)', 'batch_size'], {}), True, 'import numpy as np\n'), (192, 'numpy.log', 'np.log', (['(np.pi * scale ** 2.0 / 2.0)'], {}), True, 'import numpy as np\n'), (202, 'numpy.sqrt', 'np.sqrt', (['(2.0)'], {}), True, 'import numpy as np\n'), (237, 'numpy.sqrt', 'np.sqrt', (['np.pi'], {}), True, 'import numpy as np\n'), (260, 'numpy.sqrt', 'np.sqrt', (['np.pi'], {}), True, 'import numpy as np\n'), (264, 'numpy.sqrt', 'np.sqrt', (['np.pi'], {}), True, 'import numpy as np\n'), (237, 'numpy.sqrt', 'np.sqrt', (['(2.0)'], {}), True, 'import numpy as np\n'), (260, 'numpy.sqrt', 'np.sqrt', (['(2.0)'], {}), True, 'import numpy as np\n'), (264, 'numpy.sqrt', 'np.sqrt', (['(2.0)'], {}), True, 'import numpy as np\n'), (173, 'numpy.array', 'np.array', (['[0.01, 0.1, 1.0, 5.0, 10.0]'], {}), True, 'import numpy as np\n'), (182, 'tensorflow.gradients', 'tf.gradients', (['value', '[scale]'], {}), True, 'import tensorflow as tf\n'), (181, 'tensorflow.global_variables_initializer', 'tf.global_variables_initializer', ([], {}), True, 'import tensorflow as tf\n')]