code: # Copyright 2016 The TensorFlow Authors. All Rights Reserved.
  #
  # 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 Keras regularizers."""
  
  from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  
  from absl.testing import parameterized
  import numpy as np
  
  from tensorflow.python import keras
  from tensorflow.python.eager import context
  from tensorflow.python.keras import keras_parameterized
  from tensorflow.python.keras import regularizers
  from tensorflow.python.keras import testing_utils
  from tensorflow.python.keras.utils import np_utils
  from tensorflow.python.ops import math_ops
  from tensorflow.python.platform import test
  
  
  DATA_DIM = 5
  NUM_CLASSES = 2
  
  
  class KerasRegularizersTest(keras_parameterized.TestCase,
                              parameterized.TestCase):
  
    def create_model(self, kernel_regularizer=None, activity_regularizer=None):
      model = keras.models.Sequential()
      model.add(keras.layers.Dense(NUM_CLASSES,
                                   kernel_regularizer=kernel_regularizer,
                                   activity_regularizer=activity_regularizer,
                                   input_shape=(DATA_DIM,)))
      return model
  
    def get_data(self):
      (x_train, y_train), (x_test, y_test) = testing_utils.get_test_data(
          train_samples=10,
          test_samples=10,
          input_shape=(DATA_DIM,),
          num_classes=NUM_CLASSES)
      y_train = np_utils.to_categorical(y_train, NUM_CLASSES)
      y_test = np_utils.to_categorical(y_test, NUM_CLASSES)
      return (x_train, y_train), (x_test, y_test)
  
    def create_multi_input_model_from(self, layer1, layer2):
      input_1 = keras.layers.Input(shape=(DATA_DIM,))
      input_2 = keras.layers.Input(shape=(DATA_DIM,))
      out1 = layer1(input_1)
      out2 = layer2(input_2)
      out = keras.layers.Average()([out1, out2])
      model = keras.models.Model([input_1, input_2], out)
      model.add_loss(keras.backend.mean(out2))
      model.add_loss(math_ops.reduce_sum(input_1))
      return model
  
    @keras_parameterized.run_all_keras_modes
    @parameterized.named_parameters([
        ('l1', regularizers.l1()),
        ('l2', regularizers.l2()),
        ('l1_l2', regularizers.l1_l2()),
    ])
    def test_kernel_regularization(self, regularizer):
      (x_train, y_train), _ = self.get_data()
      model = self.create_model(kernel_regularizer=regularizer)
      model.compile(
          loss='categorical_crossentropy',
          optimizer='sgd',
          run_eagerly=testing_utils.should_run_eagerly(),
          experimental_run_tf_function=testing_utils.should_run_tf_function())
      self.assertEqual(len(model.losses), 1)
      model.fit(x_train, y_train, batch_size=10, epochs=1, verbose=0)
  
    @keras_parameterized.run_all_keras_modes
    @parameterized.named_parameters([
        ('l1', regularizers.l1()),
        ('l2', regularizers.l2()),
        ('l1_l2', regularizers.l1_l2()),
        ('l2_zero', keras.regularizers.l2(0.)),
    ])
    def test_activity_regularization(self, regularizer):
      (x_train, y_train), _ = self.get_data()
      model = self.create_model(activity_regularizer=regularizer)
      model.compile(
          loss='categorical_crossentropy',
          optimizer='sgd',
          run_eagerly=testing_utils.should_run_eagerly(),
          experimental_run_tf_function=testing_utils.should_run_tf_function())
      self.assertEqual(len(model.losses), 1 if context.executing_eagerly() else 1)
      model.fit(x_train, y_train, batch_size=10, epochs=1, verbose=0)
  
    @keras_parameterized.run_all_keras_modes
    @keras_parameterized.run_with_all_model_types
    def test_zero_regularization(self):
      # Verifies that training with zero regularization works.
      x, y = np.ones((10, 10)), np.ones((10, 3))
      model = testing_utils.get_model_from_layers(
          [keras.layers.Dense(3, kernel_regularizer=keras.regularizers.l2(0))],
          input_shape=(10,))
      model.compile(
          'sgd',
          'mse',
          run_eagerly=testing_utils.should_run_eagerly(),
          experimental_run_tf_function=testing_utils.should_run_tf_function())
      model.fit(x, y, batch_size=5, epochs=1)
  
    def test_custom_regularizer_saving(self):
  
      def my_regularizer(weights):
        return math_ops.reduce_sum(math_ops.abs(weights))
  
      inputs = keras.Input((10,))
      outputs = keras.layers.Dense(1, kernel_regularizer=my_regularizer)(inputs)
      model = keras.Model(inputs, outputs)
      model2 = model.from_config(
          model.get_config(), custom_objects={'my_regularizer': my_regularizer})
      self.assertEqual(model2.layers[1].kernel_regularizer, my_regularizer)
  
    @keras_parameterized.run_all_keras_modes
    @parameterized.named_parameters([
        ('l1', regularizers.l1()),
        ('l2', regularizers.l2()),
        ('l1_l2', regularizers.l1_l2()),
    ])
    def test_regularization_shared_layer(self, regularizer):
      dense_layer = keras.layers.Dense(
          NUM_CLASSES,
          kernel_regularizer=regularizer,
          activity_regularizer=regularizer)
      model = self.create_multi_input_model_from(dense_layer, dense_layer)
      model.compile(
          loss='categorical_crossentropy',
          optimizer='sgd',
          run_eagerly=testing_utils.should_run_eagerly(),
          experimental_run_tf_function=testing_utils.should_run_tf_function())
      self.assertLen(model.losses, 5)
  
    @keras_parameterized.run_all_keras_modes
    @parameterized.named_parameters([
        ('l1', regularizers.l1()),
        ('l2', regularizers.l2()),
        ('l1_l2', regularizers.l1_l2()),
    ])
    def test_regularization_shared_model(self, regularizer):
      dense_layer = keras.layers.Dense(
          NUM_CLASSES,
          kernel_regularizer=regularizer,
          activity_regularizer=regularizer)
  
      input_tensor = keras.layers.Input(shape=(DATA_DIM,))
      dummy_model = keras.models.Model(input_tensor, dense_layer(input_tensor))
  
      model = self.create_multi_input_model_from(dummy_model, dummy_model)
      model.compile(
          loss='categorical_crossentropy',
          optimizer='sgd',
          run_eagerly=testing_utils.should_run_eagerly(),
          experimental_run_tf_function=testing_utils.should_run_tf_function())
      self.assertLen(model.losses, 6)
  
    @keras_parameterized.run_all_keras_modes
    @parameterized.named_parameters([
        ('l1', regularizers.l1()),
        ('l2', regularizers.l2()),
        ('l1_l2', regularizers.l1_l2()),
    ])
    def test_regularization_shared_layer_in_different_models(self, regularizer):
      shared_dense = keras.layers.Dense(
          NUM_CLASSES,
          kernel_regularizer=regularizer,
          activity_regularizer=regularizer)
      models = []
      for _ in range(2):
        input_tensor = keras.layers.Input(shape=(DATA_DIM,))
        unshared_dense = keras.layers.Dense(
            NUM_CLASSES, kernel_regularizer=regularizer)
        out = unshared_dense(shared_dense(input_tensor))
        models.append(keras.models.Model(input_tensor, out))
  
      model = self.create_multi_input_model_from(
          layer1=models[0], layer2=models[1])
      model.compile(
          loss='categorical_crossentropy',
          optimizer='sgd',
          run_eagerly=testing_utils.should_run_eagerly(),
          experimental_run_tf_function=testing_utils.should_run_tf_function())
  
      # We expect to see 9 losses on the model:
      # - 2 from the 2 add_loss calls on the outer model.
      # - 3 from the weight regularizers on the shared_dense layer, unshared_dense
      # in inner model 1, unshared_dense in inner model 2.
      # - 4 from activity regularizers on the shared_dense layer.
      self.assertLen(model.losses, 9)
  
  
  if __name__ == '__main__':
    test.main()
   # Copyright 2015 The TensorFlow Authors. All Rights Reserved.
  #
  # 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.
  # ==============================================================================
  """Functional tests for coefficient-wise operations."""
  
  from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  
  import numpy as np
  
  from tensorflow.python.compat import compat
  from tensorflow.python.framework import constant_op
  from tensorflow.python.framework import dtypes as dtypes_lib
  from tensorflow.python.framework import errors
  from tensorflow.python.framework import ops
  from tensorflow.python.framework import sparse_tensor
  from tensorflow.python.framework import test_util
  from tensorflow.python.ops import array_ops
  from tensorflow.python.ops import gen_math_ops
  from tensorflow.python.ops import gradient_checker
  from tensorflow.python.ops import gradients_impl
  from tensorflow.python.ops import math_ops
  from tensorflow.python.ops import nn_grad  # pylint: disable=unused-import
  from tensorflow.python.platform import test
  
  _ADD = lambda x, y: x + y
  _SUB = lambda x, y: x - y
  _MUL = lambda x, y: x * y
  _POW = lambda x, y: x**y
  _TRUEDIV = lambda x, y: x / y
  _FLOORDIV = lambda x, y: x // y
  _MOD = lambda x, y: x % y
  
  _LT = lambda x, y: x < y
  _LE = lambda x, y: x <= y
  _GT = lambda x, y: x > y
  _GE = lambda x, y: x >= y
  
  _AND = lambda x, y: x & y
  _OR = lambda x, y: x | y
  _XOR = lambda x, y: x ^ y
  _INV = lambda x: ~x
  
  
  # TODO(zongheng): it'd be great to factor out this function and various random
  # SparseTensor gen funcs.
  def _sparsify(x, thresh=0.5, index_dtype=np.int64):
    x[x < thresh] = 0
  
    non_zero = np.where(x)
    x_indices = np.vstack(non_zero).astype(index_dtype).T
    x_values = x[non_zero]
    x_shape = x.shape
  
    return sparse_tensor.SparseTensor(
        indices=x_indices, values=x_values, dense_shape=x_shape), x_values
  
  
  def _default_tolerance(dtype):
    """Returns a sensible default tolerance for comparing results of a given type.
  
    Args:
      dtype: A datatype.
    """
    if dtype == np.float16:
      return 5e-3
    elif dtype in (np.float32, np.complex64):
      return 1e-3
    elif dtype in (np.float64, np.complex128):
      return 1e-5
    else:
      return None  # Fail fast for unexpected types
  
  
  class ComparisonOpTest(test.TestCase):
  
    def _compareScalar(self, func, x, y, dtype):
      with test_util.use_gpu():
        out = func(
            ops.convert_to_tensor(np.array([x]).astype(dtype)),
            ops.convert_to_tensor(np.array([y]).astype(dtype)))
        ret = self.evaluate(out)
      return ret[0]
  
    def testScalarCompareScalar(self):
      dtypes = [np.float16, np.float32, np.float64, np.int32, np.int64]
      data = [-1, 0, 1]
      for t in dtypes:
        for x in data:
          for y in data:
            self.assertEqual(self._compareScalar(math_ops.less, x, y, t), x < y)
            self.assertEqual(
                self._compareScalar(math_ops.less_equal, x, y, t), x <= y)
            self.assertEqual(
                self._compareScalar(math_ops.greater, x, y, t), x > y)
            self.assertEqual(
                self._compareScalar(math_ops.greater_equal, x, y, t), x >= y)
            self.assertEqual(self._compareScalar(math_ops.equal, x, y, t), x == y)
            self.assertEqual(
                self._compareScalar(math_ops.not_equal, x, y, t), x != y)
      data = [-1, 0, 1, -1j, 1j, 1 + 1j, 1 - 1j]
      for t in [np.complex64, np.complex128]:
        for x in data:
          for y in data:
            self.assertEqual(self._compareScalar(math_ops.equal, x, y, t), x == y)
            self.assertEqual(
                self._compareScalar(math_ops.not_equal, x, y, t), x != y)
  
    def _compare(self, x, y, np_func, tf_func):
      np_ans = np_func(x, y)
      with test_util.use_gpu():
        out = tf_func(ops.convert_to_tensor(x), ops.convert_to_tensor(y))
        tf_ans = self.evaluate(out)
      self.assertAllEqual(np_ans, tf_ans)
  
    def testTensorCompareTensor(self):
      x = np.linspace(-15, 15, 6).reshape(1, 3, 2)
      y = np.linspace(20, -10, 6).reshape(1, 3, 2)
      for t in [np.float16, np.float32, np.float64, np.int32, np.int64]:
        xt = x.astype(t)
        yt = y.astype(t)
        self._compare(xt, yt, np.less, math_ops.less)
        self._compare(xt, yt, np.less_equal, math_ops.less_equal)
        self._compare(xt, yt, np.greater, math_ops.greater)
        self._compare(xt, yt, np.greater_equal, math_ops.greater_equal)
        self._compare(xt, yt, np.equal, math_ops.equal)
        self._compare(xt, yt, np.not_equal, math_ops.not_equal)
      # Complex types do not support ordering but do support equality tests.
      for t in [np.complex64, np.complex128]:
        xt = x.astype(t)
        xt -= 1j * xt
        yt = y.astype(t)
        yt -= 1j * yt
        self._compare(xt, yt, np.equal, math_ops.equal)
        self._compare(xt, yt, np.not_equal, math_ops.not_equal)
  
    def _compareBCast(self, xs, ys, dtype, np_func, tf_func):
      x = np.linspace(-15, 15, np.prod(xs)).astype(dtype).reshape(xs)
      y = np.linspace(20, -10, np.prod(ys)).astype(dtype).reshape(ys)
      if dtype in (np.complex64, np.complex128):
        x -= 1j * x
        y -= 1j * y
      self._compare(x, y, np_func, tf_func)
      self._compare(y, x, np_func, tf_func)
  
    def _testBCastByFunc(self, np_func, tf_func, include_complex=False):
      shapes = [
          ([1, 3, 2], [1]),
          ([1, 3, 2], [2]),
          ([1, 3, 2], [3, 2]),
          ([1, 3, 2], [3, 1]),
          ([1, 3, 2], [1, 3, 2]),
          ([1, 3, 2], [2, 3, 1]),
          ([1, 3, 2], [2, 1, 1]),
          ([1, 3, 2], [1, 3, 1]),
          ([2, 1, 5], [2, 3, 1]),
          ([2, 0, 5], [2, 0, 1]),
          ([2, 3, 0], [2, 3, 1]),
      ]
      dtypes = [
          np.float16,
          np.float32,
          np.float64,
          np.int32,
          np.int64,
      ]
      if include_complex:
        dtypes.extend([np.complex64, np.complex128])
  
      for (xs, ys) in shapes:
        for dtype in dtypes:
          self._compareBCast(xs, ys, dtype, np_func, tf_func)
  
    def testBCastLess(self):
      self._testBCastByFunc(np.less, math_ops.less)
  
    def testBCastLessEqual(self):
      self._testBCastByFunc(np.less_equal, math_ops.less_equal)
  
    def testBCastGreater(self):
      self._testBCastByFunc(np.greater, math_ops.greater)
  
    def testBCastGreaterEqual(self):
      self._testBCastByFunc(np.greater_equal, math_ops.greater_equal)
  
    def testBCastEqual(self):
      self._testBCastByFunc(np.equal, math_ops.equal, include_complex=True)
  
    def testBCastNotEqual(self):
      self._testBCastByFunc(
          np.not_equal, math_ops.not_equal, include_complex=True)
  
    def testShapeMismatch(self):
      dtypes = [np.float16, np.float32, np.float64, np.int32, np.int64]
      funcs = [
          math_ops.less, math_ops.less_equal, math_ops.greater,
          math_ops.greater_equal, math_ops.equal, math_ops.not_equal
      ]
      x = np.arange(0, 10).reshape([2, 5])
      y = np.arange(0, 10).reshape([5, 2])
      for t in dtypes:
        for f in funcs:
          with self.assertRaisesRegexp(
              (ValueError, errors.InvalidArgumentError),
              "Incompatible shapes|Dimensions must be equal"):
            f(x.astype(t), y.astype(t))
  
  
  class LogicalOpTest(test.TestCase):
  
    def _compareBinary(self, x, y, np_func, tf_func, use_gpu=False):
      np_ans = np_func(x, y)
      with test_util.device(use_gpu=use_gpu):
        inx = ops.convert_to_tensor(x)
        iny = ops.convert_to_tensor(y)
        out = tf_func(inx, iny)
        tf_val = self.evaluate(out)
      self.assertEqual(out.dtype, dtypes_lib.bool)
      self.assertAllEqual(np_ans, tf_val)
      self.assertShapeEqual(np_ans, out)
  
    def _not(self, x, use_gpu=False):
      np_ans = np.logical_not(x)
      with test_util.device(use_gpu=use_gpu):
        out = math_ops.logical_not(ops.convert_to_tensor(x))
        tf_val = self.evaluate(out)
      self.assertEqual(out.dtype, dtypes_lib.bool)
      self.assertAllEqual(np_ans, tf_val)
      self.assertShapeEqual(np_ans, out)
  
    def testScalar(self):
      data = [np.array([True]), np.array([False])]
      for use_gpu in [True, False]:
        for x in data:
          self._not(x, use_gpu)
        for x in data:
          for y in data:
            self._compareBinary(x, y, np.logical_and, math_ops.logical_and,
                                use_gpu)
            self._compareBinary(x, y, np.logical_or, math_ops.logical_or, use_gpu)
            self._compareBinary(x, y, np.logical_xor, math_ops.logical_xor,
                                use_gpu)
  
    def testTensor(self):
      x = np.random.randint(0, 2, 6).astype(np.bool).reshape(1, 3, 2)
      y = np.random.randint(0, 2, 6).astype(np.bool).reshape(1, 3, 2)
      for use_gpu in [True, False]:
        self._not(x, use_gpu)
        self._compareBinary(x, y, np.logical_and, math_ops.logical_and, use_gpu)
        self._compareBinary(x, y, np.logical_or, math_ops.logical_or, use_gpu)
        self._compareBinary(x, y, np.logical_xor, math_ops.logical_xor, use_gpu)
  
    def testBCast(self):
      shapes = [
          ([1, 3, 2], [1]),
          ([1, 3, 2], [2]),
          ([1, 3, 2], [3, 2]),
          ([1, 3, 2], [3, 1]),
          ([1, 3, 2], [1, 3, 2]),
          ([1, 3, 2], [2, 3, 1]),
          ([1, 3, 2], [2, 1, 1]),
          ([1, 3, 2], [1, 3, 1]),
          ([2, 1, 5], [2, 3, 1]),
          ([2, 0, 5], [2, 0, 1]),
          ([2, 3, 0], [2, 3, 1]),
      ]
      for (xs, ys) in shapes:
        x = np.random.randint(0, 2, np.prod(xs)).astype(np.bool).reshape(xs)
        y = np.random.randint(0, 2, np.prod(ys)).astype(np.bool).reshape(ys)
        for use_gpu in [True, False]:
          self._compareBinary(x, y, np.logical_and, math_ops.logical_and, use_gpu)
          self._compareBinary(x, y, np.logical_or, math_ops.logical_or, use_gpu)
          self._compareBinary(x, y, np.logical_xor, math_ops.logical_xor, use_gpu)
  
    @test_util.run_deprecated_v1
    def testShapeMismatch(self):
      x = np.random.randint(0, 2, 6).astype(np.bool).reshape(1, 3, 2)
      y = np.random.randint(0, 2, 6).astype(np.bool).reshape(3, 2, 1)
      for f in [math_ops.logical_and, math_ops.logical_or, math_ops.logical_xor]:
        with self.assertRaisesWithPredicateMatch(
            ValueError, lambda e: "Dimensions must" in str(e)):
          f(x, y)
  
    @test_util.run_deprecated_v1
    def testUsingAsPythonValueFails(self):
      # Ensure that we raise an error when the user attempts to treat a
      # `Tensor` as a Python `bool`.
      b = constant_op.constant(False)
      with self.assertRaises(TypeError):
        if b:
          pass
  
      x = constant_op.constant(3)
      y = constant_op.constant(4)
      with self.assertRaises(TypeError):
        if x > y:
          pass
  
      z = constant_op.constant(7)
  
      # The chained comparison should fail because Python computes `x <
      # y` and short-circuits the comparison with `z` if it is `False`.
      with self.assertRaises(TypeError):
        _ = x < y < z
  
  
  class SelectOpTest(test.TestCase):
  
    def _compare(self, fn, c, x, y, use_gpu):
      np_ans = np.where(c, x, y)
      with test_util.device(use_gpu=use_gpu):
        out = fn(c, x, y)
        tf_ans = self.evaluate(out)
      self.assertAllEqual(np_ans, tf_ans)
      self.assertShapeEqual(np_ans, out)
  
    def _compareGradientX(self,
                          fn,
                          c,
                          x,
                          y,
                          numeric_gradient_type=None,
                          x_init_value=None):
      with self.cached_session():
        inx = ops.convert_to_tensor(x)
        iny = ops.convert_to_tensor(y)
        out = fn(c, inx, iny)
        s = list(np.shape(c))
        if x_init_value is None:
          x_init_value = x
        if x.shape != y.shape:
          x_init_value = np.broadcast_to(y, x.shape)
        jacob_t, jacob_n = gradient_checker.compute_gradient(
            inx, s, out, s, x_init_value=x_init_value)
        if numeric_gradient_type is not None:
          xf = x.astype(numeric_gradient_type)
          yf = y.astype(numeric_gradient_type)
          inxf = ops.convert_to_tensor(xf)
          inyf = ops.convert_to_tensor(yf)
          outf = fn(c, inxf, inyf)
          _, jacob_n = gradient_checker.compute_gradient(
              inxf, s, outf, s, x_init_value=xf)
          jacob_n = jacob_n.astype(x.dtype)
      if x.dtype == np.float16:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float32:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float64:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-5, atol=1e-5)
  
    def _compareGradientY(self, fn, c, x, y, numeric_gradient_type=None):
      with self.cached_session():
        inx = ops.convert_to_tensor(x)
        iny = ops.convert_to_tensor(y)
        out = fn(c, inx, iny)
        s = list(np.shape(c))
        jacob_t, jacob_n = gradient_checker.compute_gradient(
            iny, s, out, s, x_init_value=x, delta=1.0)
        if numeric_gradient_type is not None:
          xf = x.astype(numeric_gradient_type)
          yf = y.astype(numeric_gradient_type)
          inxf = ops.convert_to_tensor(xf)
          inyf = ops.convert_to_tensor(yf)
          outf = fn(c, inxf, inyf)
          _, jacob_n = gradient_checker.compute_gradient(
              inyf, s, outf, s, x_init_value=yf)
          jacob_n = jacob_n.astype(x.dtype)
      if x.dtype == np.float16:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float32:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float64:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-5, atol=1e-5)
  
    def _testScalar(self, fn):
      c = True
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(1, 3, 2) * 100
      for t in [
          np.float16, np.float32, np.float64, np.int32, np.int64, np.complex64,
          np.complex128
      ]:
        xt = x.astype(t)
        yt = y.astype(t)
        self._compare(fn, c, xt, yt, use_gpu=False)
        if t in [np.float16, np.float32, np.float64]:
          self._compare(fn, c, xt, yt, use_gpu=True)
  
    def testScalar(self):
      self._testScalar(array_ops.where)
      self._testScalar(array_ops.where_v2)
  
    def _testScalarBroadcast(self, fn, c, x, y):
      for t in [
          np.float16, np.float32, np.float64, np.int32, np.int64, np.complex64,
          np.complex128
      ]:
        xt = x.astype(t)
        yt = y.astype(t)
        self._compare(fn, c, xt, yt, use_gpu=False)
        if t in [np.float16, np.float32, np.float64]:
          self._compare(fn, c, xt, yt, use_gpu=True)
  
    def testScalarBroadcast(self):
      c = True
      # where_v2 only
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(1, 1, 1) * 100
      self._testScalarBroadcast(array_ops.where_v2, c, x, y)
      self._testScalarBroadcast(array_ops.where_v2, c, y, x)
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(1, 3, 1) * 100
      self._testScalarBroadcast(array_ops.where_v2, c, x, y)
      self._testScalarBroadcast(array_ops.where_v2, c, y, x)
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(1, 1, 2) * 100
      self._testScalarBroadcast(array_ops.where_v2, c, x, y)
      self._testScalarBroadcast(array_ops.where_v2, c, y, x)
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(1, 1) * 100
      self._testScalarBroadcast(array_ops.where_v2, c, x, y)
      self._testScalarBroadcast(array_ops.where_v2, c, y, x)
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(1) * 100
      self._testScalarBroadcast(array_ops.where_v2, c, x, y)
      self._testScalarBroadcast(array_ops.where_v2, c, y, x)
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(1, 2) * 100
      self._testScalarBroadcast(array_ops.where_v2, c, x, y)
      self._testScalarBroadcast(array_ops.where_v2, c, y, x)
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(3, 2) * 100
      self._testScalarBroadcast(array_ops.where_v2, c, x, y)
      self._testScalarBroadcast(array_ops.where_v2, c, y, x)
  
    def _testBasic(self, fn):
      c = np.random.randint(0, 2, 6).astype(np.bool).reshape(1, 3, 2)
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(1, 3, 2) * 100
      for t in [
          np.float16, np.float32, np.float64, np.int32, np.int64, np.complex64,
          np.complex128
      ]:
        xt = x.astype(t)
        yt = y.astype(t)
        self._compare(fn, c, xt, yt, use_gpu=False)
        if t in [np.float16, np.float32, np.float64]:
          self._compare(fn, c, xt, yt, use_gpu=True)
  
    def testBasic(self):
      self._testBasic(array_ops.where)
      self._testBasic(array_ops.where_v2)
  
    def _testBasicBroadcast(self, fn, c, x, y):
      for t in [
          np.float16, np.float32, np.float64, np.int32, np.int64, np.complex64,
          np.complex128
      ]:
        xt = x.astype(t)
        yt = y.astype(t)
        self._compare(fn, c, xt, yt, use_gpu=False)
        if t in [np.float16, np.float32, np.float64]:
          self._compare(fn, c, xt, yt, use_gpu=True)
  
    def testBasicBroadcast(self):
      c0 = np.random.randint(0, 2, 6).astype(np.bool).reshape(1, 3, 2)
      c1 = np.random.randint(0, 2, 2).astype(np.bool).reshape(1, 1, 2)
      c2 = np.random.randint(0, 2, 3).astype(np.bool).reshape(1, 3, 1)
      c3 = np.random.randint(0, 2, 1).astype(np.bool).reshape(1, 1, 1)
      for c in [c0, c1, c2, c3]:
        # where_v2 only
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1, 1, 1) * 100
        self._testBasicBroadcast(array_ops.where_v2, c, x, y)
        self._testBasicBroadcast(array_ops.where_v2, c, y, x)
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1, 3, 1) * 100
        self._testBasicBroadcast(array_ops.where_v2, c, x, y)
        self._testBasicBroadcast(array_ops.where_v2, c, y, x)
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1, 1, 2) * 100
        self._testBasicBroadcast(array_ops.where_v2, c, x, y)
        self._testBasicBroadcast(array_ops.where_v2, c, y, x)
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1, 1) * 100
        self._testBasicBroadcast(array_ops.where_v2, c, x, y)
        self._testBasicBroadcast(array_ops.where_v2, c, y, x)
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1) * 100
        self._testBasicBroadcast(array_ops.where_v2, c, x, y)
        self._testBasicBroadcast(array_ops.where_v2, c, y, x)
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1, 2) * 100
        self._testBasicBroadcast(array_ops.where_v2, c, x, y)
        self._testBasicBroadcast(array_ops.where_v2, c, y, x)
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(3, 2) * 100
        self._testBasicBroadcast(array_ops.where_v2, c, x, y)
        self._testBasicBroadcast(array_ops.where_v2, c, y, x)
  
    def _testGradients(self, fn):
      c = np.random.randint(0, 2, 6).astype(np.bool).reshape(1, 3, 2)
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(1, 3, 2) * 100
      for t in [np.float16, np.float32, np.float64]:
        xt = x.astype(t)
        yt = y.astype(t)
        if t == np.float16:
          # Compare fp16 theoretical gradients to fp32 numerical gradients,
          # since fp16 numerical gradients are too imprecise unless great
          # care is taken with choosing the inputs and the delta. This is
          # a weaker check (in particular, it does not test the op itself,
          # only its gradient), but it's much better than nothing.
          self._compareGradientX(fn, c, xt, yt, np.float)
          self._compareGradientY(fn, c, xt, yt, np.float)
        else:
          self._compareGradientX(fn, c, xt, yt)
          self._compareGradientY(fn, c, xt, yt)
  
    @test_util.run_deprecated_v1
    def testGradients(self):
      self._testGradients(array_ops.where)
      self._testGradients(array_ops.where_v2)
  
    @test_util.run_deprecated_v1
    def testGradientsBroadcast(self):
      c = np.random.randint(0, 2, 6).astype(np.bool).reshape(1, 3, 2)
      for t in [np.float32, np.float64]:
        # where_v2 only
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1, 1, 1) * 100
        self._compareGradientX(array_ops.where_v2, c, x.astype(t), y.astype(t))
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1, 3, 1) * 100
        self._compareGradientX(array_ops.where_v2, c, x.astype(t), y.astype(t))
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1, 1, 2) * 100
        self._compareGradientX(array_ops.where_v2, c, x.astype(t), y.astype(t))
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1, 1) * 100
        self._compareGradientX(array_ops.where_v2, c, x.astype(t), y.astype(t))
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1) * 100
        self._compareGradientX(array_ops.where_v2, c, x.astype(t), y.astype(t))
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(1, 2) * 100
        self._compareGradientX(array_ops.where_v2, c, x.astype(t), y.astype(t))
        x = np.random.rand(1, 3, 2) * 100
        y = np.random.rand(3, 2) * 100
        self._compareGradientX(array_ops.where_v2, c, x.astype(t), y.astype(t))
  
    def _testShapeMismatch(self, fn):
      c = np.random.randint(0, 2, 6).astype(np.bool).reshape(1, 3, 2)
      x = np.random.rand(1, 3, 2) * 100
      y = np.random.rand(2, 5, 3) * 100
      for t in [
          np.float16, np.float32, np.float64, np.int32, np.int64, np.complex64,
          np.complex128
      ]:
        xt = x.astype(t)
        yt = y.astype(t)
        with self.assertRaises(ValueError):
          fn(c, xt, yt)
  
    @test_util.run_deprecated_v1
    def testShapeMismatch(self):
      self._testShapeMismatch(array_ops.where)
      self._testShapeMismatch(array_ops.where_v2)
  
    def _testEmptyTensor(self, fn):
      c = np.random.randint(0, 3, 0).astype(np.bool).reshape(1, 3, 0)
      x = np.random.rand(1, 3, 0) * 100
      y = np.random.rand(1, 3, 0) * 100
      z_expected = np.zeros((1, 3, 0), dtype=np.float32)
      with self.cached_session():
        xt = x.astype(np.float32)
        yt = y.astype(np.float32)
        z = fn(c, xt, yt).eval()
        self.assertAllEqual(z_expected, z)
  
    @test_util.run_deprecated_v1
    def testEmptyTensor(self):
      self._testEmptyTensor(array_ops.where)
      self._testEmptyTensor(array_ops.where_v2)
  
    def _testNan(self, fn):
      with self.cached_session():
        for c in False, True:
          for a in 7.0, np.nan:
            for b in 5.0, np.nan:
              x = fn(c, a, b).eval()
              y = a if c else b
              self.assertEqual(np.isnan(x), np.isnan(y))
  
    @test_util.run_deprecated_v1
    def testNan(self):
      """Verify that nans don't propagate where they shouldn't."""
      self._testNan(array_ops.where)
      self._testNan(array_ops.where_v2)
  
  
  class BatchSelectOpTest(test.TestCase):
    """Test broadcasting of Select when 'c' is a vec and 't' &'e' are rank2+."""
  
    def _compare(self, c, x, y, use_gpu):
      np_ans = np.dstack(
          [x_i if c_i else y_i for c_i, x_i, y_i in zip(c, x, y)]).transpose(
              [2, 0, 1])
      with test_util.device(use_gpu=use_gpu):
        out = array_ops.where(c, x, y)
        tf_ans = self.evaluate(out)
      self.assertAllEqual(np_ans, tf_ans)
      self.assertShapeEqual(np_ans, out)
  
    def _compareGradientX(self, c, x, y, numeric_gradient_type=None):
      with self.cached_session():
        inx = ops.convert_to_tensor(x)
        iny = ops.convert_to_tensor(y)
        out = array_ops.where(c, inx, iny)
        s = list(np.shape(x))
        jacob_t, jacob_n = gradient_checker.compute_gradient(
            inx, s, out, s, x_init_value=x)
        if numeric_gradient_type is not None:
          xf = x.astype(numeric_gradient_type)
          yf = y.astype(numeric_gradient_type)
          inxf = ops.convert_to_tensor(xf)
          inyf = ops.convert_to_tensor(yf)
          outf = array_ops.where(c, inxf, inyf)
          _, jacob_n = gradient_checker.compute_gradient(
              inxf, s, outf, s, x_init_value=xf)
          jacob_n = jacob_n.astype(x.dtype)
      if x.dtype == np.float16:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float32:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float64:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-5, atol=1e-5)
  
    def _compareGradientY(self, c, x, y, numeric_gradient_type=None):
      with self.cached_session():
        inx = ops.convert_to_tensor(x)
        iny = ops.convert_to_tensor(y)
        out = array_ops.where(c, inx, iny)
        s = list(np.shape(x))
        jacob_t, jacob_n = gradient_checker.compute_gradient(
            iny, s, out, s, x_init_value=y)
        if numeric_gradient_type is not None:
          xf = x.astype(numeric_gradient_type)
          yf = y.astype(numeric_gradient_type)
          inxf = ops.convert_to_tensor(xf)
          inyf = ops.convert_to_tensor(yf)
          outf = array_ops.where(c, inxf, inyf)
          _, jacob_n = gradient_checker.compute_gradient(
              inyf, s, outf, s, x_init_value=yf)
          jacob_n = jacob_n.astype(x.dtype)
      if x.dtype == np.float16:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float32:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float64:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-5, atol=1e-5)
  
    def testBasic(self):
      c = np.random.randint(0, 2, 16).astype(np.bool)
      x = np.random.rand(16, 2, 8) * 100
      y = np.random.rand(16, 2, 8) * 100
      for t in [
          np.float16, np.float32, np.float64, np.int32, np.int64, np.complex64,
          np.complex128
      ]:
        xt = x.astype(t)
        yt = y.astype(t)
        self._compare(c, xt, yt, use_gpu=False)
        if t in [np.float16, np.float32, np.float64]:
          self._compare(c, xt, yt, use_gpu=True)
  
    @test_util.run_deprecated_v1
    def testGradients(self):
      c = np.random.randint(0, 2, 16).astype(np.bool)
      x = np.random.rand(16, 2, 8) * 100
      y = np.random.rand(16, 2, 8) * 100
      for t in [np.float16, np.float32, np.float64]:
        xt = x.astype(t)
        yt = y.astype(t)
        if t == np.float16:
          # Compare fp16 theoretical gradients to fp32 numerical gradients,
          # since fp16 numerical gradients are too imprecise unless great
          # care is taken with choosing the inputs and the delta. This is
          # a weaker check (in particular, it does not test the op itself,
          # only its gradient), but it's much better than nothing.
          self._compareGradientX(c, xt, yt, np.float)
          self._compareGradientY(c, xt, yt, np.float)
        else:
          self._compareGradientX(c, xt, yt)
          self._compareGradientY(c, xt, yt)
  
    @test_util.run_deprecated_v1
    def testShapeMismatch(self):
      c = np.random.randint(0, 2, 8).astype(np.bool)
      x = np.random.rand(16, 3, 2) * 100
      y = np.random.rand(16, 3, 2) * 100
      for t in [
          np.float16, np.float32, np.float64, np.int32, np.int64, np.complex64,
          np.complex128
      ]:
        xt = x.astype(t)
        yt = y.astype(t)
        with self.assertRaises(ValueError):
          array_ops.where(c, xt, yt)
  
  
  class MinMaxOpTest(test.TestCase):
  
    def _compare(self, x, y, use_gpu):
      np_min, np_max = np.minimum(x, y), np.maximum(x, y)
      with test_util.device(use_gpu=use_gpu):
        inx = ops.convert_to_tensor(x)
        iny = ops.convert_to_tensor(y)
        omin, omax = math_ops.minimum(inx, iny), math_ops.maximum(inx, iny)
        tf_min, tf_max = self.evaluate([omin, omax])
      self.assertAllEqual(np_min, tf_min)
      self.assertAllEqual(np_max, tf_max)
  
    def testBasic(self):
      x = np.random.rand(1, 3, 2) * 100.
      y = np.random.rand(1, 3, 2) * 100.
      for t in [np.float16, np.float32, np.float64, np.int32, np.int64]:
        self._compare(x.astype(t), y.astype(t), use_gpu=False)
        self._compare(x.astype(t), y.astype(t), use_gpu=True)
  
    def testDifferentShapes(self):
      x = np.random.rand(1, 3, 2) * 100.
      y = np.random.rand(2) * 100.  # should broadcast
      for t in [np.float16, np.float32, np.float64, np.int32, np.int64]:
        self._compare(x.astype(t), y.astype(t), use_gpu=False)
        self._compare(x.astype(t), y.astype(t), use_gpu=True)
  
    def testScalar(self):
      x = np.random.rand(1, 3, 2) * 100.
      y = np.random.rand(1).item() * 100.  # should broadcast
      # dropped np.float64, int64 because TF automatically converts to 32 bit
      for t in [np.float32, np.int32]:
        self._compare(x.astype(t), t(y), use_gpu=False)
        self._compare(x.astype(t), t(y), use_gpu=True)
  
    def _compareGradientX(self, func, x, y):
      with self.cached_session():
        inx = ops.convert_to_tensor(x)
        iny = ops.convert_to_tensor(y)
        out = func(inx, iny)
        s = list(np.shape(x))
        jacob_t, jacob_n = gradient_checker.compute_gradient(
            inx, s, out, s, x_init_value=x)
      if x.dtype == np.float16:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float32:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float64:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-5, atol=1e-5)
  
    def _compareGradientY(self, func, x, y):
      with self.cached_session():
        inx = ops.convert_to_tensor(x)
        iny = ops.convert_to_tensor(y)
        out = func(inx, iny)
        s = list(np.shape(x))
        jacob_t, jacob_n = gradient_checker.compute_gradient(
            iny, s, out, s, x_init_value=y)
      if x.dtype == np.float16:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float32:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-3, atol=1e-3)
      elif x.dtype == np.float64:
        self.assertAllClose(jacob_t, jacob_n, rtol=1e-5, atol=1e-5)
  
    @test_util.run_deprecated_v1
    def testGradients(self):
      x = np.random.rand(1, 3, 2) * 100.
      # ensure x != y
      y = x + (np.random.randint(2, size=x.shape) - .5) * 2  # -1 or +1
      self._compareGradientX(math_ops.maximum, x, y)
      self._compareGradientY(math_ops.maximum, x, y)
      self._compareGradientX(math_ops.minimum, x, y)
      self._compareGradientY(math_ops.minimum, x, y)
  
  
  class MathOpsOverloadTest(test.TestCase):
  
    def _computeTensorAndLiteral(self, x, y, dtype, func):
      with test_util.force_cpu():
        inx = ops.convert_to_tensor(x, dtype=dtype)
        z = func(inx, y)  # Should use __add__, __sub__, etc.
        return self.evaluate(z)
  
    def _computeLiteralAndTensor(self, x, y, dtype, func):
      with test_util.force_cpu():
        iny = ops.convert_to_tensor(y, dtype=dtype)
        z = func(x, iny)  # Should use __radd__, __rsub__, etc.
        return self.evaluate(z)
  
    def _compareBinary(self, x, y, dtype, np_func, tf_func):
      np_ans = np_func(x, y).astype(dtype.as_numpy_dtype)
      self.assertAllClose(np_ans,
                          self._computeTensorAndLiteral(x, y, dtype, tf_func))
      self.assertAllClose(np_ans,
                          self._computeLiteralAndTensor(x, y, dtype, tf_func))
  
    def _compareUnary(self, x, dtype, np_func, tf_func):
      np_ans = np_func(x).astype(dtype.as_numpy_dtype)
      with test_util.force_cpu():
        self.assertAllClose(
            np_ans, self.evaluate(tf_func(ops.convert_to_tensor(x, dtype=dtype))))
  
    def testOverload(self):
      dtypes = [
          dtypes_lib.float16,
          dtypes_lib.float32,
          dtypes_lib.float64,
          dtypes_lib.int32,
          dtypes_lib.int64,
          dtypes_lib.complex64,
          dtypes_lib.complex128,
      ]
      funcs = [
          (np.add, _ADD),
          (np.subtract, _SUB),
          (np.multiply, _MUL),
          (np.power, _POW),
          (np.true_divide, _TRUEDIV),
          (np.floor_divide, _FLOORDIV),
      ]
      for dtype in dtypes:
        for np_func, tf_func in funcs:
          if dtype in (dtypes_lib.complex64,
                       dtypes_lib.complex128) and tf_func == _FLOORDIV:
            continue  # floordiv makes no sense for complex
          self._compareBinary(10, 5, dtype, np_func, tf_func)
      # Mod only works for int32 and int64.
      for dtype in [dtypes_lib.int32, dtypes_lib.int64]:
        self._compareBinary(10, 3, dtype, np.mod, _MOD)
  
    def testOverloadComparisons(self):
      dtypes = [
          dtypes_lib.float16,
          dtypes_lib.float32,
          dtypes_lib.float64,
          dtypes_lib.int32,
          dtypes_lib.int64,
      ]
      funcs = [
          (np.less, _LT),
          (np.less_equal, _LE),
          (np.greater, _GT),
          (np.greater_equal, _GE),
      ]
      for dtype in dtypes:
        for np_func, tf_func in funcs:
          self._compareBinary(10, 5, dtype, np_func, tf_func)
      logical_funcs = [(np.logical_and, _AND), (np.logical_or, _OR),
                       (np.logical_xor, _XOR), (np.equal, math_ops.equal),
                       (np.not_equal, math_ops.not_equal)]
      for np_func, tf_func in logical_funcs:
        self._compareBinary(True, False, dtypes_lib.bool, np_func, tf_func)
        self._compareBinary(True, True, dtypes_lib.bool, np_func, tf_func)
        self._compareBinary(False, False, dtypes_lib.bool, np_func, tf_func)
        self._compareBinary(False, True, dtypes_lib.bool, np_func, tf_func)
        self._compareBinary([True, True, False, False],
                            [True, False, True, False], dtypes_lib.bool, np_func,
                            tf_func)
      self._compareUnary(True, dtypes_lib.bool, np.logical_not, _INV)
      self._compareUnary(False, dtypes_lib.bool, np.logical_not, _INV)
      self._compareUnary([True, False], dtypes_lib.bool, np.logical_not, _INV)
  
  
  class IsFiniteInfNanTest(test.TestCase):
  
    def _compare(self, x, use_gpu):
      np_finite, np_inf, np_nan = np.isfinite(x), np.isinf(x), np.isnan(x)
      with test_util.device(use_gpu=use_gpu):
        inx = ops.convert_to_tensor(x)
        ofinite, oinf, onan = math_ops.is_finite(inx), math_ops.is_inf(
            inx), math_ops.is_nan(inx)
        tf_finite, tf_inf, tf_nan = self.evaluate([ofinite, oinf, onan])
      self.assertAllEqual(np_inf, tf_inf)
      self.assertAllEqual(np_nan, tf_nan)
      self.assertAllEqual(np_finite, tf_finite)
      self.assertShapeEqual(np_inf, oinf)
      self.assertShapeEqual(np_nan, onan)
      self.assertShapeEqual(np_finite, ofinite)
  
    def _testDtype(self, dtype):
      fi = np.finfo(dtype)
      data = np.array([
          0, -1, 1, fi.resolution, -fi.resolution, fi.min, fi.max, -np.inf,
          np.inf, np.nan
      ]).astype(dtype)
      self._compare(data, use_gpu=False)
      self._compare(data, use_gpu=True)
  
    def testHalf(self):
      self._testDtype(np.float16)
  
    def testFloat(self):
      self._testDtype(np.float32)
  
    def testDouble(self):
      self._testDtype(np.float64)
  
    def testSqrt(self):
      for dtype in [np.float16, np.float32, np.float64]:
        fi = np.finfo(dtype)
        for size in [1, 3, 4, 7, 8, 63, 64, 65]:
          # For float32 Eigen uses Carmack's fast vectorized sqrt algorithm.
          # It is not accurate for very large arguments, so we test for
          # fi.max/100 instead of fi.max here.
          for value in [fi.min, -2, -1, 0, fi.tiny, 1, 2, 1000, fi.max / 100]:
            x = np.full((size,), value, dtype=dtype)
            np_y = np.sqrt(x)
            np_nan = np.isnan(np_y)
            with test_util.use_gpu():
              tf_y = math_ops.sqrt(x)
              tf_nan = math_ops.is_nan(tf_y)
              if value < 0:
                self.assertAllEqual(np_nan, self.evaluate(tf_nan))
              else:
                self.assertAllCloseAccordingToType(np_y, self.evaluate(tf_y))
  
  
  class RoundingTest(test.TestCase):
  
    def _compare_values(self, x, y=None):
      y = np.rint(x) if y is None else np.asarray(y)
  
      tf_rint = math_ops.rint(x)
      np_rint = self.evaluate(tf_rint)
  
      self.assertAllEqual(y, np_rint)
      self.assertShapeEqual(y, tf_rint)
  
    def _compare(self, x):
      np_floor, np_ceil = np.floor(x), np.ceil(x)
  
      inx = ops.convert_to_tensor(x)
      ofloor, oceil = math_ops.floor(inx), math_ops.ceil(inx)
      tf_floor, tf_ceil = self.evaluate([ofloor, oceil])
  
      self.assertAllEqual(np_floor, tf_floor)
      self.assertAllEqual(np_ceil, tf_ceil)
      self.assertShapeEqual(np_floor, ofloor)
      self.assertShapeEqual(np_ceil, oceil)
  
    def _testDtype(self, dtype):
      data = (np.arange(-3, 3) / 4.).reshape(1, 3, 2).astype(dtype)
      self._compare(data)
      # TODO: rint op is not supported for float16
      if dtype is np.float16:
        return
      self._compare_values(data)
      x = [0.5, 0.5000001]
      y = [0.0, 1.0]
      self._compare_values(x, y=y)
  
      # numpy example
      x = [-1.7, -1.5, -0.2, 0.2, 1.5, 1.7, 2.0]
      y = [-2., -2., -0., 0., 2., 2., 2.]
      self._compare_values(x, y=y)
  
    def testTypes(self):
      self.skipTest("b/131162241")
      for dtype in [np.float16, np.float32, np.float64]:
        self._testDtype(dtype)
  
  
  class ComplexMakeRealImagTest(test.TestCase):
  
    def _compareMake(self, real, imag, use_gpu):
      np_ans = real + (1j) * imag
  
      with test_util.device(use_gpu=use_gpu):
        real = ops.convert_to_tensor(real)
        imag = ops.convert_to_tensor(imag)
        tf_ans = math_ops.complex(real, imag)
        out = self.evaluate(tf_ans)
  
      self.assertAllEqual(np_ans, out)
      self.assertShapeEqual(np_ans, tf_ans)
  
    def testMake(self):
      real = (np.arange(-3, 3) / 4.).reshape([1, 3, 2]).astype(np.float32)
      imag = (np.arange(-3, 3) / 5.).reshape([1, 3, 2]).astype(np.float32)
      for use_gpu in [False, True]:
        self._compareMake(real, imag, use_gpu)
        self._compareMake(real, 12.0, use_gpu)
        self._compareMake(23.0, imag, use_gpu)
  
    def testRealImagNumericType(self):
      for use_gpu in [True, False]:
        for value in [1., 1j, 1. + 1j]:
          np_real, np_imag = np.real(value), np.imag(value)
          with test_util.device(use_gpu=use_gpu):
            tf_real = math_ops.real(value)
            tf_imag = math_ops.imag(value)
            self.assertAllEqual(np_real, self.evaluate(tf_real))
            self.assertAllEqual(np_imag, self.evaluate(tf_imag))
  
    def _compareRealImag(self, cplx, use_gpu):
      np_real, np_imag = np.real(cplx), np.imag(cplx)
      np_zeros = np_real * 0
  
      with test_util.device(use_gpu=use_gpu):
        inx = ops.convert_to_tensor(cplx)
        tf_real = math_ops.real(inx)
        tf_imag = math_ops.imag(inx)
        tf_real_real = math_ops.real(tf_real)
        tf_imag_real = math_ops.imag(tf_real)
        self.assertAllEqual(np_real, self.evaluate(tf_real))
        self.assertAllEqual(np_imag, self.evaluate(tf_imag))
        self.assertAllEqual(np_real, self.evaluate(tf_real_real))
        self.assertAllEqual(np_zeros, self.evaluate(tf_imag_real))
  
    def testRealImag64(self):
      real = (np.arange(-3, 3) / 4.).reshape([1, 3, 2]).astype(np.float32)
      imag = (np.arange(-3, 3) / 5.).reshape([1, 3, 2]).astype(np.float32)
      cplx = real + 1j * imag
      self._compareRealImag(cplx, use_gpu=False)
      self._compareRealImag(cplx, use_gpu=True)
  
    def testRealImag128(self):
      real = (np.arange(-3, 3) / 4.).reshape([1, 3, 2]).astype(np.float64)
      imag = (np.arange(-3, 3) / 5.).reshape([1, 3, 2]).astype(np.float64)
      cplx = real + 1j * imag
      self._compareRealImag(cplx, use_gpu=False)
      self._compareRealImag(cplx, use_gpu=True)
  
    def _compareAngle(self, cplx, use_gpu):
      np_angle = np.angle(cplx)
  
      with test_util.device(use_gpu=use_gpu):
        inx = ops.convert_to_tensor(cplx)
        tf_angle = math_ops.angle(inx)
        tf_angle_val = self.evaluate(tf_angle)
  
      self.assertAllClose(np_angle, tf_angle_val)
      self.assertShapeEqual(np_angle, tf_angle)
  
    def testAngle64(self):
      real = (np.arange(-3, 3) / 4.).reshape([1, 3, 2]).astype(np.float32)
      imag = (np.arange(-3, 3) / 5.).reshape([1, 3, 2]).astype(np.float32)
      cplx = real + 1j * imag
      self._compareAngle(cplx, use_gpu=False)
      self._compareAngle(cplx, use_gpu=True)
  
    def testAngle(self):
      real = (np.arange(-3, 3) / 4.).reshape([1, 3, 2]).astype(np.float64)
      imag = (np.arange(-3, 3) / 5.).reshape([1, 3, 2]).astype(np.float64)
      cplx = real + 1j * imag
      self._compareAngle(cplx, use_gpu=False)
      self._compareAngle(cplx, use_gpu=True)
  
    @test_util.run_deprecated_v1
    def testRealReal(self):
      for dtype in (dtypes_lib.int32, dtypes_lib.int64, dtypes_lib.float32,
                    dtypes_lib.float64):
        x = array_ops.placeholder(dtype)
        y = math_ops.real(x)
        self.assertEqual(x, y)
  
    def _compareConj(self, cplx, use_gpu):
      np_ans = np.conj(cplx)
      with test_util.device(use_gpu=use_gpu):
        inx = ops.convert_to_tensor(cplx)
        tf_conj = math_ops.conj(inx)
        tf_ans = self.evaluate(tf_conj)
      self.assertAllEqual(np_ans, tf_ans)
      self.assertShapeEqual(np_ans, tf_conj)
  
    def testConj64(self):
      real = (np.arange(-3, 3) / 4.).reshape([1, 3, 2]).astype(np.float32)
      imag = (np.arange(-3, 3) / 5.).reshape([1, 3, 2]).astype(np.float32)
      cplx = real + 1j * imag
      self._compareConj(cplx, use_gpu=False)
      self._compareConj(cplx, use_gpu=True)
  
    def testConj128(self):
      real = (np.arange(-3, 3) / 4.).reshape([1, 3, 2]).astype(np.float64)
      imag = (np.arange(-3, 3) / 5.).reshape([1, 3, 2]).astype(np.float64)
      cplx = real + 1j * imag
      self._compareConj(cplx, use_gpu=False)
      self._compareConj(cplx, use_gpu=True)
  
    @test_util.run_deprecated_v1
    def testConjReal(self):
      for dtype in (dtypes_lib.int32, dtypes_lib.int64, dtypes_lib.float16,
                    dtypes_lib.float32, dtypes_lib.float64):
        x = array_ops.placeholder(dtype)
        y = math_ops.conj(x)
        self.assertEqual(x, y)
  
    @test_util.run_deprecated_v1
    def testConjString(self):
      x = array_ops.placeholder(dtypes_lib.string)
      with self.assertRaisesRegexp(TypeError,
                                   r"Expected numeric or variant tensor"):
        math_ops.conj(x)
  
    def _compareGradient(self, x):
      # x[:, 0] is real, x[:, 1] is imag.  We combine real and imag into
      # complex numbers. Then, we extract real and imag parts and
      # computes the squared sum. This is obviously the same as sum(real
      # * real) + sum(imag * imag). We just want to make sure the
      # gradient function is checked.
      with self.cached_session():
        inx = ops.convert_to_tensor(x)
        real, imag = array_ops.split(value=inx, num_or_size_splits=2, axis=1)
        real, imag = array_ops.reshape(real, [-1]), array_ops.reshape(imag, [-1])
        cplx = math_ops.complex(real, imag)
        cplx = math_ops.conj(cplx)
        loss = math_ops.reduce_sum(math_ops.square(
            math_ops.real(cplx))) + math_ops.reduce_sum(
                math_ops.square(math_ops.imag(cplx)))
        epsilon = 1e-3
        jacob_t, jacob_n = gradient_checker.compute_gradient(
            inx, list(x.shape), loss, [1], x_init_value=x, delta=epsilon)
      self.assertAllClose(jacob_t, jacob_n, rtol=epsilon, atol=epsilon)
  
    def _compareBroadcastGradient(self, x):
      x_ = ops.convert_to_tensor(x)
      epsilon = 1e-3
      with self.cached_session():
        for args in [(x_, 0.), (0., x_)]:
          z = math_ops.reduce_sum(math_ops.abs(math_ops.complex(*args)))
          jacob_t, jacob_n = gradient_checker.compute_gradient(
              x_, list(x.shape), z, [1], x_init_value=x, delta=epsilon)
          self.assertAllClose(jacob_t, jacob_n, rtol=epsilon, atol=epsilon)
  
    @test_util.run_deprecated_v1
    def testGradient(self):
      # complex64
      data = np.arange(1, 2, 0.10).reshape([5, 2]).astype(np.float32)
      self._compareGradient(data)
      self._compareBroadcastGradient(data)
      # complex128
      data = np.arange(1, 2, 0.10).reshape([5, 2]).astype(np.float64)
      self._compareGradient(data)
  
    def _compareMulGradient(self, data):
      # data is a float matrix of shape [n, 4].  data[:, 0], data[:, 1],
      # data[:, 2], data[:, 3] are real parts of x, imaginary parts of
      # x, real parts of y and imaginary parts of y.
      with self.cached_session():
        inp = ops.convert_to_tensor(data)
        xr, xi, yr, yi = array_ops.split(value=inp, num_or_size_splits=4, axis=1)
  
        def vec(x):  # Reshape to a vector
          return array_ops.reshape(x, [-1])
  
        xr, xi, yr, yi = vec(xr), vec(xi), vec(yr), vec(yi)
  
        def cplx(r, i):  # Combine to a complex vector
          return math_ops.complex(r, i)
  
        x, y = cplx(xr, xi), cplx(yr, yi)
        # z is x times y in complex plane.
        z = x * y
        # Defines the loss function as the sum of all coefficients of z.
        loss = math_ops.reduce_sum(math_ops.real(z) + math_ops.imag(z))
        epsilon = 0.005
        jacob_t, jacob_n = gradient_checker.compute_gradient(
            inp, list(data.shape), loss, [1], x_init_value=data, delta=epsilon)
      self.assertAllClose(jacob_t, jacob_n, rtol=epsilon, atol=epsilon)
  
    @test_util.run_deprecated_v1
    def testMulGradient(self):
      data = np.arange(1, 2, 0.125).reshape([2, 4]).astype(np.float32)
      self._compareMulGradient(data)
  
  
  class PolyvalTest(test.TestCase):
  
    def _runtest(self, dtype, degree):
      x = np.random.rand(2, 2).astype(dtype)
      coeffs = [np.random.rand(2, 2).astype(dtype) for _ in range(degree + 1)]
      np_val = np.polyval(coeffs, x)
      with self.cached_session():
        tf_val = math_ops.polyval(coeffs, x)
        self.assertAllClose(np_val, self.evaluate(tf_val))
  
    def testSimple(self):
      for dtype in [
          np.int32, np.float32, np.float64, np.complex64, np.complex128
      ]:
        for degree in range(5):
          self._runtest(dtype, degree)
  
    def testBroadcast(self):
      dtype = np.float32
      degree = 3
      shapes = [(1,), (2, 1), (1, 2), (2, 2)]
      for x_shape in shapes:
        for coeff_shape in shapes:
          x = np.random.rand(*x_shape).astype(dtype)
          coeffs = [
              np.random.rand(*coeff_shape).astype(dtype)
              for _ in range(degree + 1)
          ]
          np_val = np.polyval(coeffs, x)
          with self.cached_session():
            tf_val = math_ops.polyval(coeffs, x)
            self.assertAllClose(np_val, self.evaluate(tf_val))
  
    def testEmpty(self):
      x = np.random.rand(2, 2).astype(np.float32)
      coeffs = []
      np_val = np.polyval(coeffs, x)
      with self.cached_session():
        tf_val = math_ops.polyval(coeffs, x)
        self.assertAllClose(np_val, self.evaluate(tf_val))
  
  
  class SingularGradientOpTest(test.TestCase):
  
    @test_util.run_deprecated_v1
    def testGradientAtSingularity(self):
      if not compat.forward_compatible(2020, 3, 14):
        self.skipTest("Skipping test for future functionality.")
  
      ops_and_singularity = [
          (gen_math_ops.reciprocal, (0.,)),
          (gen_math_ops.rsqrt, (0.,)),
          (gen_math_ops.sqrt, (0.,)),
          (gen_math_ops.sqrt_grad, (
              0.,
              0.,
          )),
          (gen_math_ops.reciprocal_grad, (
              1.,
              0.,
          )),
          (gen_math_ops.tan, (np.pi / 2,)),
          (gen_math_ops.log, (0.,)),
          (gen_math_ops.log1p, (-1.,)),
          (gen_math_ops.acosh, (0.,)),
          (gen_math_ops.asin, (1.,)),
          (gen_math_ops.acos, (1.,)),
          (gen_math_ops.atan2, (0., 0.)),
          (gen_math_ops.div, (1., 0.)),
          (gen_math_ops.div_no_nan, (1., 0.)),
          (gen_math_ops.real_div, (1., 0.)),
          (math_ops.pow, (0., -1.)),
      ]
      for op, singularity in ops_and_singularity:
        for dtype in (dtypes_lib.half, dtypes_lib.float32, dtypes_lib.float64,
                      dtypes_lib.complex64, dtypes_lib.complex128):
          if dtype.is_complex and op in [
              gen_math_ops.asin, gen_math_ops.acos, gen_math_ops.atan2
          ]:
            continue
          if dtype == dtypes_lib.half and op in [
              gen_math_ops.acosh, gen_math_ops.asin, gen_math_ops.acos,
              gen_math_ops.atan2
          ]:
            continue
          with self.cached_session():
            print("op = ", op, ", singularity = ", singularity, ", type = ",
                  dtype)
            args = [constant_op.constant(s, dtype=dtype) for s in singularity]
            grad_y = constant_op.constant(0, dtype=dtype)
            y = op(*args)
            g = gradients_impl.gradients(y, args, grad_ys=grad_y)
            g_val = self.evaluate(g)
            self.assertAllEqual(g_val, np.zeros(len(singularity)))
  
  
  if __name__ == "__main__":
    test.main()
   # Copyright 2017 The TensorFlow Authors. All Rights Reserved.
  #
  # 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 the experimental input pipeline statistics gathering ops."""
  from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  
  from absl.testing import parameterized
  import numpy as np
  
  from tensorflow.python.data.experimental.kernel_tests import reader_dataset_ops_test_base
  from tensorflow.python.data.experimental.kernel_tests import stats_dataset_test_base
  from tensorflow.python.data.experimental.ops import batching
  from tensorflow.python.data.experimental.ops import stats_aggregator
  from tensorflow.python.data.experimental.ops import stats_ops
  from tensorflow.python.data.kernel_tests import test_base
  from tensorflow.python.data.ops import dataset_ops
  from tensorflow.python.framework import combinations
  from tensorflow.python.framework import errors
  from tensorflow.python.framework import ops
  from tensorflow.python.ops import array_ops
  from tensorflow.python.ops import math_ops
  from tensorflow.python.platform import test
  
  
  # TODO(jsimsa): Figure out why are graph tests failing.
  class StatsDatasetTest(stats_dataset_test_base.StatsDatasetTestBase,
                         parameterized.TestCase):
  
    @combinations.generate(test_base.eager_only_combinations())
    def testBytesProduced(self):
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(100).map(
          lambda x: array_ops.tile([x], ops.convert_to_tensor([x]))).apply(
              stats_ops.bytes_produced_stats("bytes_produced"))
      dataset = self.datasetExperimentalStats(dataset, aggregator)
      next_element = self.getNext(dataset, requires_initialization=True)
  
      expected_sum = 0.0
      for i in range(100):
        self.assertAllEqual(
            np.array([i] * i, dtype=np.int64), self.evaluate(next_element()))
        handle = self.getHandle(aggregator)
        self.assertStatisticsHasCount(handle, "bytes_produced", float(i + 1),
                                      i + 2)
        expected_sum += i * 8.0
        self.assertStatisticsHasSum(handle, "bytes_produced", expected_sum, i + 2)
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element())
      handle = self.getHandle(aggregator)
      self.assertStatisticsHasCount(handle, "bytes_produced", 100.0, 101)
      self.assertStatisticsHasSum(handle, "bytes_produced", expected_sum, 101)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testLatencyStats(self):
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(100).apply(
          stats_ops.latency_stats("record_latency"))
  
      dataset = self.datasetExperimentalStats(dataset, aggregator)
  
      next_element = self.getNext(dataset, requires_initialization=True)
  
      for i in range(100):
        self.assertEqual(i, self.evaluate(next_element()))
        handle = self.getHandle(aggregator)
        self.assertStatisticsHasCount(handle, "record_latency", float(i + 1),
                                      i + 2)
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element())
      handle = self.getHandle(aggregator)
      self.assertStatisticsHasCount(handle, "record_latency", 100.0, 101)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testPrefetchBufferUtilization(self):
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(100).map(
          lambda x: array_ops.tile([x], ops.convert_to_tensor([x]))).prefetch(-1)
      dataset = self.datasetExperimentalStats(dataset, aggregator)
      next_element = self.getNext(dataset, requires_initialization=True)
      for i in range(100):
        self.assertAllEqual(
            np.array([i] * i, dtype=np.int64), self.evaluate(next_element()))
        handle = self.getHandle(aggregator)
        self.assertStatisticsHasCount(
            handle,
            self.regexForNodeName("PrefetchDataset", "buffer_utilization"),
            float(i + 1),
            3 * i + 4,
            offset=2)
        self.assertStatisticsContains(
            handle, self.regexForNodeName("PrefetchDataset", "buffer_capacity"),
            3 * i + 4)
        self.assertStatisticsContains(
            handle,
            self.regexForNodeName("PrefetchDataset", "buffer_size"),
            3 * i + 4,
            offset=1)
        self.assertStatisticsHasRange(
            handle,
            self.regexForNodeName("PrefetchDataset", "buffer_utilization"),
            0,
            1,
            3 * i + 4,
            offset=2)
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element())
      handle = self.getHandle(aggregator)
      self.assertStatisticsHasCount(
          handle,
          self.regexForNodeName("PrefetchDataset", "buffer_utilization"),
          100,
          301,
          offset=2)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testPrefetchBufferScalars(self):
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(10).map(
          lambda x: array_ops.tile([x], ops.convert_to_tensor([x]))).prefetch(1)
      dataset = self.datasetExperimentalStats(dataset, aggregator)
      next_element = self.getNext(dataset, requires_initialization=True)
  
      for i in range(10):
        self.assertAllEqual(
            np.array([i] * i, dtype=np.int64), self.evaluate(next_element()))
        handle = self.getHandle(aggregator)
        self.assertStatisticsHasScalarValue(
            handle, self.regexForNodeName("PrefetchDataset", "buffer_capacity"),
            1, 3 * i + 4)
        self.assertStatisticsHasScalarValue(
            handle,
            self.regexForNodeName("PrefetchDataset", "buffer_size"),
            1,
            3 * i + 4,
            offset=1)
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element())
  
    @combinations.generate(test_base.eager_only_combinations())
    def testFilteredElementsStats(self):
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(101).filter(
          lambda x: math_ops.equal(math_ops.mod(x, 3), 0))
      dataset = self.datasetExperimentalStats(dataset, aggregator)
      next_element = self.getNext(dataset, requires_initialization=True)
  
      for i in range(34):
        self.assertEqual(i * 3, self.evaluate(next_element()))
        handle = self.getHandle(aggregator)
        if i != 0:
          self.assertStatisticsHasScalarValue(
              handle, self.regexForNodeName("FilterDataset", "dropped_elements"),
              float(i * 2))
        self.assertStatisticsHasScalarValue(
            handle, self.regexForNodeName("FilterDataset", "filtered_elements"),
            float(i + 1))
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element())
      handle = self.getHandle(aggregator)
      self.assertStatisticsHasScalarValue(
          handle, self.regexForNodeName("FilterDataset", "dropped_elements"),
          67.0)
      self.assertStatisticsHasScalarValue(
          handle, self.regexForNodeName("FilterDataset", "filtered_elements"),
          34.0)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testReinitialize(self):
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(100).apply(
          stats_ops.latency_stats("record_latency"))
      dataset = self.datasetExperimentalStats(dataset, aggregator)
  
      for j in range(5):
        next_element = self.getNext(dataset, requires_initialization=True)
        for i in range(100):
          self.assertEqual(i, self.evaluate(next_element()))
          handle = self.getHandle(aggregator)
          self.assertStatisticsHasCount(handle, "record_latency",
                                        float((j * 100) + i + 1),
                                        (j * 100) + i + 2)
        with self.assertRaises(errors.OutOfRangeError):
          self.evaluate(next_element())
        handle = self.getHandle(aggregator)
        self.assertStatisticsHasCount(handle, "record_latency", (j + 1) * 100.0,
                                      (j * 100) + 101)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testNoAggregatorRegistered(self):
      dataset = dataset_ops.Dataset.range(100).apply(
          stats_ops.latency_stats("record_latency"))
  
      next_element = self.getNext(dataset, requires_initialization=True)
  
      for i in range(100):
        self.assertEqual(i, self.evaluate(next_element()))
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element())
  
    @combinations.generate(test_base.eager_only_combinations())
    def testMultipleTags(self):
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(100).apply(
          stats_ops.latency_stats("record_latency")).apply(
              stats_ops.latency_stats("record_latency_2"))
      dataset = self.datasetExperimentalStats(dataset, aggregator)
  
      next_element = self.getNext(dataset, requires_initialization=True)
  
      for i in range(100):
        self.assertEqual(i, self.evaluate(next_element()))
        handle = self.getHandle(aggregator)
        self.assertStatisticsHasCount(
            handle, "record_latency", float(i + 1), 2 * i + 3, offset=1)
        self.assertStatisticsHasCount(handle, "record_latency_2", float(i + 1),
                                      2 * i + 3)
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element())
      handle = self.getHandle(aggregator)
      self.assertStatisticsHasCount(
          handle, "record_latency", 100.0, 201, offset=1)
      self.assertStatisticsHasCount(handle, "record_latency_2", 100.0, 201)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testRepeatedTags(self):
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(100).apply(
          stats_ops.latency_stats("record_latency")).apply(
              stats_ops.latency_stats("record_latency"))
      dataset = self.datasetExperimentalStats(dataset, aggregator)
      next_element = self.getNext(dataset, requires_initialization=True)
  
      for i in range(100):
        self.assertEqual(i, self.evaluate(next_element()))
        handle = self.getHandle(aggregator)
        self.assertStatisticsHasCount(handle, "record_latency",
                                      float(2 * (i + 1)), 2 * i + 3)
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element())
      handle = self.getHandle(aggregator)
      self.assertStatisticsHasCount(handle, "record_latency", 200.0, 201)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testMultipleIteratorsSameAggregator(self):
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(100).apply(
          stats_ops.latency_stats("record_latency"))
      dataset = self.datasetExperimentalStats(dataset, aggregator)
      next_element1 = self.getNext(dataset, requires_initialization=True)
      next_element2 = self.getNext(dataset, requires_initialization=True)
  
      for i in range(100):
        self.assertEqual(i * 2, self.evaluate(next_element1() + next_element2()))
        handle = self.getHandle(aggregator)
        self.assertStatisticsHasCount(handle, "record_latency",
                                      float(2 * (i + 1)), 2 * i + 3)
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element1())
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element2())
      handle = self.getHandle(aggregator)
      self.assertStatisticsHasCount(handle, "record_latency", 200.0, 201)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testMultipleDatasetWithPrefixes(self):
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(100).apply(
          stats_ops.latency_stats("record_latency"))
      dataset = self.datasetExperimentalStats(
          dataset, aggregator, prefix="dataset1")
      dataset2 = dataset_ops.Dataset.range(100).apply(
          stats_ops.latency_stats("record_latency"))
      dataset2 = self.datasetExperimentalStats(
          dataset2, aggregator, prefix="dataset2")
      next_element1 = self.getNext(dataset, requires_initialization=True)
      next_element2 = self.getNext(dataset2, requires_initialization=True)
  
      for i in range(100):
        self.assertEqual(i * 2, self.evaluate(next_element1() + next_element2()))
        handle = self.getHandle(aggregator)
        self.assertStatisticsHasCount(
            handle, "dataset1::record_latency", float(i + 1), 2 * i + 3, offset=1)
        self.assertStatisticsHasCount(handle, "dataset2::record_latency",
                                      float(i + 1), 2 * i + 3)
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element1())
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element2())
      handle = self.getHandle(aggregator)
      self.assertStatisticsHasCount(
          handle, "dataset1::record_latency", 100.0, 201, offset=1)
      self.assertStatisticsHasCount(handle, "dataset2::record_latency", 100.0,
                                    201)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testMultiplePrefetchStats(self):
  
      aggregator = stats_aggregator.StatsAggregator()
      dataset = dataset_ops.Dataset.range(10).prefetch(
          2).filter(lambda x: math_ops.equal(math_ops.mod(x, 2), 0)).prefetch(1)
  
      dataset = self.datasetExperimentalStats(dataset, aggregator)
      next_element = self.getNext(dataset, requires_initialization=True)
  
      for i in range(5):
        self.assertEqual(i * 2, self.evaluate(next_element()))
        handle = self.getHandle(aggregator)
        # TODO(shivaniagarwal): using exact name of prefetch node than the regex,
        # to differentiate between two prefetch. This might break in future, at
        # which point, it would be best to disable this test.
        self.assertStatisticsHasScalarValue(
            handle, "PrefetchDataset/_5::buffer_capacity", 2)
        self.assertStatisticsContains(handle, "PrefetchDataset/_5::buffer_size")
        self.assertStatisticsHasScalarValue(
            handle, "PrefetchDataset/_8::buffer_capacity", 1)
        self.assertStatisticsContains(handle, "PrefetchDataset/_8::buffer_size")
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element())
  
  
  class ThreadUtilizationStatsTest(stats_dataset_test_base.StatsDatasetTestBase,
                                   parameterized.TestCase):
  
    @combinations.generate(test_base.eager_only_combinations())
    def testMapBufferUtilization(self):
  
      def dataset_fn():
        return dataset_ops.Dataset.range(10).map(
            lambda x: array_ops.tile([x], ops.convert_to_tensor([x])),
            num_parallel_calls=4)
  
      self.parallelCallsStats(
          dataset_fn, {"ParallelMapDataset"}, 10, function_processing_time=True)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testMapAutoTuneBufferUtilization(self):
  
      def dataset_fn():
        return dataset_ops.Dataset.range(10).map(
            lambda x: array_ops.tile([x], ops.convert_to_tensor([x])),
            num_parallel_calls=dataset_ops.AUTOTUNE)
  
      self.parallelCallsStats(
          dataset_fn, {"ParallelMapDataset"}, 10, function_processing_time=True)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testInterleaveAutoTuneBufferUtilization(self):
  
      def dataset_fn():
  
        def interleave_fn(_):
          return dataset_ops.Dataset.range(
              10).map(lambda x: array_ops.tile([x], ops.convert_to_tensor([x])))
  
        return dataset_ops.Dataset.range(1).interleave(
            interleave_fn,
            cycle_length=1,
            num_parallel_calls=dataset_ops.AUTOTUNE)
  
      self.parallelCallsStats(dataset_fn, {"ParallelInterleaveDatasetV2"}, 10)
  
    @combinations.generate(test_base.eager_only_combinations())
    def testMapAndBatchAutoTuneBufferUtilization(self):
  
      def dataset_fn():
        return dataset_ops.Dataset.range(100).apply(
            batching.map_and_batch(
                lambda x: array_ops.tile([x], ops.convert_to_tensor([2])),
                num_parallel_calls=dataset_ops.AUTOTUNE,
                batch_size=16))
  
      num_output = 100 // 16 + 1
      self.parallelCallsStats(
          dataset_fn, {"MapAndBatchDataset"},
          num_output,
          check_elements=False,
          function_processing_time=True)
  
  
  class FeatureStatsDatasetTest(
      stats_dataset_test_base.StatsDatasetTestBase,
      reader_dataset_ops_test_base.MakeBatchedFeaturesDatasetTestBase,
      parameterized.TestCase):
  
    @combinations.generate(test_base.eager_only_combinations())
    def testFeaturesStats(self):
      num_epochs = 5
      total_records = num_epochs * self._num_records
      batch_size = 2
  
      def dataset_fn():
        return self.make_batch_feature(
            filenames=self.test_filenames[0],
            num_epochs=num_epochs,
            batch_size=batch_size,
            shuffle=True,
            shuffle_seed=5,
            drop_final_batch=False)
  
      num_output = total_records // batch_size
      if total_records % batch_size:
        num_output = total_records // batch_size + 1
  
      self.parallelCallsStats(
          dataset_fn, {"ParseExampleDataset"},
          num_output,
          check_elements=False)
  
      aggregator = stats_aggregator.StatsAggregator()
      dataset = self.datasetExperimentalStats(
          dataset_fn(), aggregator, prefix="record_stats")
  
      next_element = self.getNext(dataset, requires_initialization=True)
  
      for _ in range(num_output):
        self.evaluate(next_element())
  
      with self.assertRaises(errors.OutOfRangeError):
        self.evaluate(next_element())
      handle = self.getHandle(aggregator)
      self.assertStatisticsHasCount(
          handle,
          self.regexForNodeName("record_stats::ParseExampleDataset",
                                "features_count"), total_records)
      self.assertStatisticsHasCount(
          handle,
          self.regexForNodeName("record_stats::ParseExampleDataset",
                                "feature_values_count"), total_records)
      self.assertStatisticsHasSum(
          handle,
          self.regexForNodeName("record_stats::ParseExampleDataset",
                                "features_count"), total_records * 4)
      self.assertStatisticsHasSum(
          handle,
          self.regexForNodeName("record_stats::ParseExampleDataset",
                                "feature_values_count"),
          self._sum_keywords(1) * num_epochs + 3 * total_records)
  
  
  if __name__ == "__main__":
    test.main()
  
apis: numpy.ones tensorflow.python.keras.regularizers.l1_l2 tensorflow.python.keras.testing_utils.should_run_eagerly tensorflow.python.keras.layers.Average tensorflow.python.keras.models.Sequential tensorflow.python.keras.Model tensorflow.python.keras.Input tensorflow.python.keras.testing_utils.get_test_data tensorflow.python.ops.math_ops.reduce_sum tensorflow.python.keras.regularizers.l2 tensorflow.python.keras.layers.Dense tensorflow.python.keras.utils.np_utils.to_categorical tensorflow.python.ops.math_ops.abs tensorflow.python.keras.backend.mean tensorflow.python.keras.models.Model tensorflow.python.platform.test.main tensorflow.python.keras.regularizers.l1 tensorflow.python.eager.context.executing_eagerly tensorflow.python.keras.testing_utils.should_run_tf_function tensorflow.python.keras.layers.Input
apis: tensorflow.python.ops.math_ops.maximum numpy.asarray tensorflow.python.ops.gradients_impl.gradients numpy.full tensorflow.python.ops.gradient_checker.compute_gradient numpy.polyval tensorflow.python.ops.math_ops.is_finite tensorflow.python.framework.constant_op.constant numpy.isfinite numpy.vstack tensorflow.python.ops.array_ops.reshape tensorflow.python.ops.array_ops.placeholder tensorflow.python.ops.math_ops.angle numpy.logical_not numpy.random.rand numpy.isnan tensorflow.python.ops.math_ops.is_inf tensorflow.python.framework.sparse_tensor.SparseTensor numpy.where tensorflow.python.ops.math_ops.imag tensorflow.python.compat.compat.forward_compatible numpy.linspace tensorflow.python.ops.math_ops.polyval numpy.minimum numpy.sqrt numpy.ceil numpy.zeros tensorflow.python.ops.math_ops.minimum tensorflow.python.framework.test_util.device numpy.arange tensorflow.python.ops.math_ops.rint tensorflow.python.framework.test_util.force_cpu tensorflow.python.ops.math_ops.ceil tensorflow.python.ops.math_ops.real numpy.prod numpy.maximum tensorflow.python.ops.math_ops.floor numpy.finfo tensorflow.python.framework.ops.convert_to_tensor numpy.broadcast_to numpy.rint numpy.conj numpy.isinf numpy.floor tensorflow.python.ops.math_ops.complex tensorflow.python.ops.math_ops.sqrt tensorflow.python.ops.array_ops.split tensorflow.python.platform.test.main numpy.shape numpy.angle tensorflow.python.framework.test_util.use_gpu tensorflow.python.ops.array_ops.where numpy.array tensorflow.python.ops.math_ops.is_nan numpy.random.randint tensorflow.python.ops.math_ops.conj numpy.real numpy.imag
apis: tensorflow.python.ops.math_ops.mod tensorflow.python.data.ops.dataset_ops.Dataset.range tensorflow.python.platform.test.main tensorflow.python.data.kernel_tests.test_base.eager_only_combinations numpy.array tensorflow.python.data.experimental.ops.stats_ops.bytes_produced_stats tensorflow.python.framework.ops.convert_to_tensor tensorflow.python.data.experimental.ops.stats_aggregator.StatsAggregator tensorflow.python.data.experimental.ops.stats_ops.latency_stats