# ==============================================================================
"""Tests for Keras callbacks."""

import tensorflow.compat.v2 as tf

import os
import shutil
import tempfile


            

Reported by Pylint.

              import shutil
import tempfile

from absl.testing import parameterized
import numpy as np
from keras import callbacks
from keras import callbacks_v1
from keras import combinations
from keras import layers

            

Reported by Pylint.

                        loss='categorical_crossentropy',
          optimizer='sgd',
          metrics=['accuracy'])
      callbacks_v1.TensorBoard._init_writer = _init_writer
      tsb = callbacks_v1.TensorBoard(
          log_dir=tmpdir,
          histogram_freq=1,
          write_images=True,
          write_grads=True,

            

Reported by Pylint.

              
      def add_summary(self, summary, step):
        if 'batch_' in summary.value[0].tag:
          self.batch_summary = (step, summary)
        elif 'epoch_' in summary.value[0].tag:
          self.epoch_summary = (step, summary)

      def flush(self):
        pass

            

Reported by Pylint.

                      if 'batch_' in summary.value[0].tag:
          self.batch_summary = (step, summary)
        elif 'epoch_' in summary.value[0].tag:
          self.epoch_summary = (step, summary)

      def flush(self):
        pass

      def close(self):

            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

import os
import shutil
import tempfile

from absl.testing import parameterized
import numpy as np

            

Reported by Pylint.

              import tensorflow.compat.v2 as tf

import os
import shutil
import tempfile

from absl.testing import parameterized
import numpy as np
from keras import callbacks

            

Reported by Pylint.

              
import os
import shutil
import tempfile

from absl.testing import parameterized
import numpy as np
from keras import callbacks
from keras import callbacks_v1

            

Reported by Pylint.

              BATCH_SIZE = 5


class TestTensorBoardV1(tf.test.TestCase, parameterized.TestCase):

  def test_TensorBoard(self):
    np.random.seed(1337)

    temp_dir = self.get_temp_dir()

            

Reported by Pylint.

              
class TestTensorBoardV1(tf.test.TestCase, parameterized.TestCase):

  def test_TensorBoard(self):
    np.random.seed(1337)

    temp_dir = self.get_temp_dir()
    self.addCleanup(shutil.rmtree, temp_dir, ignore_errors=True)


            

Reported by Pylint.

              from __future__ import division
from __future__ import print_function

import tensorflow.compat.v2 as tf

import copy
import json
import os
import threading

            

Reported by Pylint.

              import os
import threading
import time
from tensorflow.core.protobuf import cluster_pb2
from tensorflow.python.platform import tf_logging as logging

_worker_context = threading.local()
_thread_local = threading.local()


            

Reported by Pylint.

              import threading
import time
from tensorflow.core.protobuf import cluster_pb2
from tensorflow.python.platform import tf_logging as logging

_worker_context = threading.local()
_thread_local = threading.local()



            

Reported by Pylint.

              def _configure_session_config_for_std_servers(strategy, eval_strategy,
                                              session_config, cluster_spec,
                                              task_type, task_id):
  # pylint: disable=g-doc-args
  """Call strategy's `configure` to mutate the session_config.

  The session_config is currently needed as default config for a TensorFlow
  server. In the future, we should be able to remove this method and only pass
  the session config to a client session.

            

Reported by Pylint.

                    ValueError: if `worker_barrier` is not passed to the __init__ method.
    """
    if not self._worker_barrier:
      # TODO(yuefengz): we should throw an error in independent worker mode.
      return
    self._worker_barrier.wait()

  def session_creator(self,
                      scaffold=None,

            

Reported by Pylint.

                del session_config.device_filters[:]


# TODO(yuefengz): propagate cluster_spec in the STANDALONE_CLIENT mode.
# TODO(yuefengz): we may need a smart way to figure out whether the current task
# is the special task when we support cluster_spec propagation.
def run_distribute_coordinator(worker_fn,
                               strategy,
                               eval_fn=None,

            

Reported by Pylint.

              

# TODO(yuefengz): propagate cluster_spec in the STANDALONE_CLIENT mode.
# TODO(yuefengz): we may need a smart way to figure out whether the current task
# is the special task when we support cluster_spec propagation.
def run_distribute_coordinator(worker_fn,
                               strategy,
                               eval_fn=None,
                               eval_strategy=None,

            

Reported by Pylint.

                    task_id = int(task_env.get("index", task_id))

  if cluster_spec:
    # TODO(yuefengz): validate cluster_spec.
    cluster_spec = normalize_cluster_spec(cluster_spec)
  elif hasattr(strategy.extended, "_cluster_resolver"):
    cluster_resolver = strategy.extended._cluster_resolver  # pylint: disable=protected-access
    task_type = cluster_resolver.task_type
    task_id = cluster_resolver.task_id

            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

import copy
import json
import os
import threading
import time
from tensorflow.core.protobuf import cluster_pb2

            

Reported by Pylint.

              import tensorflow.compat.v2 as tf

import copy
import json
import os
import threading
import time
from tensorflow.core.protobuf import cluster_pb2
from tensorflow.python.platform import tf_logging as logging

            

Reported by Pylint.

              # ==============================================================================
"""Tests for Adamax."""

import tensorflow.compat.v2 as tf

from absl.testing import parameterized
import numpy as np
from keras import combinations
from keras.optimizer_v2 import adamax

            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

from absl.testing import parameterized
import numpy as np
from keras import combinations
from keras.optimizer_v2 import adamax



            

Reported by Pylint.

              
from absl.testing import parameterized
import numpy as np
from keras import combinations
from keras.optimizer_v2 import adamax


def adamax_update_numpy(param,
                        g_t,

            

Reported by Pylint.

              from absl.testing import parameterized
import numpy as np
from keras import combinations
from keras.optimizer_v2 import adamax


def adamax_update_numpy(param,
                        g_t,
                        t,

            

Reported by Pylint.

              
def get_beta_accumulators(opt, dtype):
  local_step = tf.cast(opt.iterations + 1, dtype)
  beta_1_t = tf.cast(opt._get_hyper("beta_1"), dtype)
  beta_1_power = tf.pow(beta_1_t, local_step)
  return beta_1_power


class AdamaxOptimizerTest(tf.test.TestCase, parameterized.TestCase):

            

Reported by Pylint.

              class AdamaxOptimizerTest(tf.test.TestCase, parameterized.TestCase):

  def testResourceSparse(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32, tf.float64]:
      with tf.Graph().as_default(), self.cached_session():
        # Initialize variables for numpy implementation.
        zero_slots = lambda: np.zeros((3), dtype=dtype.as_numpy_dtype)  # pylint: disable=cell-var-from-loop
        m0, v0, m1, v1 = zero_slots(), zero_slots(), zero_slots(), zero_slots()

            

Reported by Pylint.

                        self.assertAllCloseAccordingToType(var1_np, var1)

  def testSparseDevicePlacement(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for index_dtype in [tf.int32, tf.int64]:
      with tf.Graph().as_default(), self.cached_session(
          force_gpu=tf.test.is_gpu_available()):
        # If a GPU is available, tests that all optimizer ops can be placed on
        # it (i.e. they have GPU kernels).

            

Reported by Pylint.

                      minimize_op.run()

  def testSparseRepeatedIndices(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32, tf.float64]:
      with tf.Graph().as_default(), self.cached_session():
        repeated_index_update_var = tf.Variable(
            [[1.0], [2.0]], dtype=dtype)
        aggregated_update_var = tf.Variable(

            

Reported by Pylint.

                                                           rtol=1e-2)

  def testTensorLearningRate(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32, tf.float64]:
      with tf.Graph().as_default(), self.cached_session():
        # Initialize variables for numpy implementation.
        m0, v0, m1, v1 = 0.0, 0.0, 0.0, 0.0
        var0_np = np.array([1.0, 2.0], dtype=dtype.as_numpy_dtype)

            

Reported by Pylint.

                        self.assertAllCloseAccordingToType(var1_np, var1)

  def testSharing(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32, tf.float64]:
      with tf.Graph().as_default(), self.cached_session():
        # Initialize variables for numpy implementation.
        m0, v0, m1, v1 = 0.0, 0.0, 0.0, 0.0
        var0_np = np.array([1.0, 2.0], dtype=dtype.as_numpy_dtype)

            

Reported by Pylint.

              # ==============================================================================
"""Tests for cudnn recurrent layers."""

import tensorflow.compat.v2 as tf

import os
import tempfile

from absl.testing import parameterized

            

Reported by Pylint.

              import os
import tempfile

from absl.testing import parameterized
import numpy as np

import keras
from tensorflow.python.framework import test_util
from keras import combinations

            

Reported by Pylint.

              import numpy as np

import keras
from tensorflow.python.framework import test_util
from keras import combinations
from keras import keras_parameterized
from keras import testing_utils
from keras.optimizer_v2.rmsprop import RMSprop


            

Reported by Pylint.

                    output = layer(inputs, initial_state=initial_state)
    self.assertTrue(
        any(initial_state[0] is t
            for t in layer._inbound_nodes[0].input_tensors))

    model = keras.models.Model([inputs] + initial_state, output)
    model.compile(
        loss='categorical_crossentropy',
        optimizer=RMSprop(learning_rate=0.001),

            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

import os
import tempfile

from absl.testing import parameterized
import numpy as np


            

Reported by Pylint.

              import tensorflow.compat.v2 as tf

import os
import tempfile

from absl.testing import parameterized
import numpy as np

import keras

            

Reported by Pylint.

              

@keras_parameterized.run_all_keras_modes
class CuDNNTest(keras_parameterized.TestCase):

  @parameterized.named_parameters(
      *testing_utils.generate_combinations_with_testcase_name(
          layer_class=[keras.layers.CuDNNGRU, keras.layers.CuDNNLSTM],
          return_sequences=[True, False]))

            

Reported by Pylint.

              @keras_parameterized.run_all_keras_modes
class CuDNNTest(keras_parameterized.TestCase):

  @parameterized.named_parameters(
      *testing_utils.generate_combinations_with_testcase_name(
          layer_class=[keras.layers.CuDNNGRU, keras.layers.CuDNNLSTM],
          return_sequences=[True, False]))
  @test_util.run_gpu_only
  def test_cudnn_rnn_return_sequence(self, layer_class, return_sequences):

            

Reported by Pylint.

                    *testing_utils.generate_combinations_with_testcase_name(
          layer_class=[keras.layers.CuDNNGRU, keras.layers.CuDNNLSTM],
          return_sequences=[True, False]))
  @test_util.run_gpu_only
  def test_cudnn_rnn_return_sequence(self, layer_class, return_sequences):
    input_size = 10
    timesteps = 6
    units = 2
    num_samples = 32

            

Reported by Pylint.

                        layer_class=[keras.layers.CuDNNGRU, keras.layers.CuDNNLSTM],
          return_sequences=[True, False]))
  @test_util.run_gpu_only
  def test_cudnn_rnn_return_sequence(self, layer_class, return_sequences):
    input_size = 10
    timesteps = 6
    units = 2
    num_samples = 32
    testing_utils.layer_test(

            

Reported by Pylint.

              # ==============================================================================
"""Tests for Keras text category_encoding preprocessing layer."""

import tensorflow.compat.v2 as tf

from absl.testing import parameterized
import numpy as np

import keras

            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

from absl.testing import parameterized
import numpy as np

import keras
from keras import backend
from keras import keras_parameterized

            

Reported by Pylint.

              

@keras_parameterized.run_all_keras_modes(always_skip_v1=True)
class CategoryEncodingInputTest(keras_parameterized.TestCase,
                                preprocessing_test_utils.PreprocessingLayerTest
                               ):

  def test_dense_input_sparse_output(self):
    input_array = tf.constant([[1, 2, 3], [3, 3, 0]])

            

Reported by Pylint.

                                              preprocessing_test_utils.PreprocessingLayerTest
                               ):

  def test_dense_input_sparse_output(self):
    input_array = tf.constant([[1, 2, 3], [3, 3, 0]])

    # The expected output should be (X for missing value):
    # [[X, 1, 1, 1, X, X]
    #  [1, X, X, 2, X, X]]

            

Reported by Pylint.

                                              preprocessing_test_utils.PreprocessingLayerTest
                               ):

  def test_dense_input_sparse_output(self):
    input_array = tf.constant([[1, 2, 3], [3, 3, 0]])

    # The expected output should be (X for missing value):
    # [[X, 1, 1, 1, X, X]
    #  [1, X, X, 2, X, X]]

            

Reported by Pylint.

                                             ):

  def test_dense_input_sparse_output(self):
    input_array = tf.constant([[1, 2, 3], [3, 3, 0]])

    # The expected output should be (X for missing value):
    # [[X, 1, 1, 1, X, X]
    #  [1, X, X, 2, X, X]]
    expected_indices = [[0, 1], [0, 2], [0, 3], [1, 0], [1, 3]]

            

Reported by Pylint.

                  # The expected output should be (X for missing value):
    # [[X, 1, 1, 1, X, X]
    #  [1, X, X, 2, X, X]]
    expected_indices = [[0, 1], [0, 2], [0, 3], [1, 0], [1, 3]]
    expected_values = [1, 1, 1, 1, 2]
    num_tokens = 6

    input_data = keras.Input(shape=(None,), dtype=tf.int32)
    layer = category_encoding.CategoryEncoding(

            

Reported by Pylint.

                  # [[X, 1, 1, 1, X, X]
    #  [1, X, X, 2, X, X]]
    expected_indices = [[0, 1], [0, 2], [0, 3], [1, 0], [1, 3]]
    expected_values = [1, 1, 1, 1, 2]
    num_tokens = 6

    input_data = keras.Input(shape=(None,), dtype=tf.int32)
    layer = category_encoding.CategoryEncoding(
        num_tokens=num_tokens, output_mode=category_encoding.COUNT, sparse=True)

            

Reported by Pylint.

                  #  [1, X, X, 2, X, X]]
    expected_indices = [[0, 1], [0, 2], [0, 3], [1, 0], [1, 3]]
    expected_values = [1, 1, 1, 1, 2]
    num_tokens = 6

    input_data = keras.Input(shape=(None,), dtype=tf.int32)
    layer = category_encoding.CategoryEncoding(
        num_tokens=num_tokens, output_mode=category_encoding.COUNT, sparse=True)
    int_data = layer(input_data)

            

Reported by Pylint.

                  expected_values = [1, 1, 1, 1, 2]
    num_tokens = 6

    input_data = keras.Input(shape=(None,), dtype=tf.int32)
    layer = category_encoding.CategoryEncoding(
        num_tokens=num_tokens, output_mode=category_encoding.COUNT, sparse=True)
    int_data = layer(input_data)

    model = keras.Model(inputs=input_data, outputs=int_data)

            

Reported by Pylint.

              # ==============================================================================
"""Pooling layers."""

import tensorflow.compat.v2 as tf

import functools
from keras import backend
from keras.engine.base_layer import Layer
from keras.engine.input_spec import InputSpec

            

Reported by Pylint.

              from keras.engine.base_layer import Layer
from keras.engine.input_spec import InputSpec
from keras.utils import conv_utils
from tensorflow.python.util.tf_export import keras_export


class Pooling1D(Layer):
  """Pooling layer for arbitrary pooling functions, for 1D inputs.


            

Reported by Pylint.

                  self.data_format = conv_utils.normalize_data_format(data_format)
    self.input_spec = InputSpec(ndim=3)

  def call(self, inputs):
    pad_axis = 2 if self.data_format == 'channels_last' else 3
    inputs = tf.expand_dims(inputs, pad_axis)
    outputs = self.pool_function(
        inputs,
        self.pool_size + (1,),

            

Reported by Pylint.

                  self.data_format = conv_utils.normalize_data_format(data_format)
    self.input_spec = InputSpec(ndim=4)

  def call(self, inputs):
    if self.data_format == 'channels_last':
      pool_shape = (1,) + self.pool_size + (1,)
      strides = (1,) + self.strides + (1,)
    else:
      pool_shape = (1, 1) + self.pool_size

            

Reported by Pylint.

                  self.data_format = conv_utils.normalize_data_format(data_format)
    self.input_spec = InputSpec(ndim=5)

  def call(self, inputs):
    pool_shape = (1,) + self.pool_size + (1,)
    strides = (1,) + self.strides + (1,)

    if self.data_format == 'channels_first':
      # TF does not support `channels_first` with 3D pooling operations,

            

Reported by Pylint.

                  if self.data_format == 'channels_first':
      # TF does not support `channels_first` with 3D pooling operations,
      # so we must handle this case manually.
      # TODO(fchollet): remove this when TF pooling is feature-complete.
      inputs = tf.transpose(inputs, (0, 2, 3, 4, 1))

    outputs = self.pool_function(
        inputs,
        ksize=pool_shape,

            

Reported by Pylint.

                    else:
        return tf.TensorShape([input_shape[0], input_shape[2]])

  def call(self, inputs):
    raise NotImplementedError

  def get_config(self):
    config = {'data_format': self.data_format, 'keepdims': self.keepdims}
    base_config = super(GlobalPooling1D, self).get_config()

            

Reported by Pylint.

                                                               **kwargs)
    self.supports_masking = True

  def call(self, inputs, mask=None):
    steps_axis = 1 if self.data_format == 'channels_last' else 2
    if mask is not None:
      mask = tf.cast(mask, inputs[0].dtype)
      mask = tf.expand_dims(
          mask, 2 if self.data_format == 'channels_last' else 1)

            

Reported by Pylint.

                    else:
        return tf.TensorShape([input_shape[0], input_shape[1]])

  def call(self, inputs):
    raise NotImplementedError

  def get_config(self):
    config = {'data_format': self.data_format, 'keepdims': self.keepdims}
    base_config = super(GlobalPooling2D, self).get_config()

            

Reported by Pylint.

                    else:
        return tf.TensorShape([input_shape[0], input_shape[1]])

  def call(self, inputs):
    raise NotImplementedError

  def get_config(self):
    config = {'data_format': self.data_format, 'keepdims': self.keepdims}
    base_config = super(GlobalPooling3D, self).get_config()

            

Reported by Pylint.

              # ==============================================================================
"""Tests for tf.keras models with callbacks, checkpointing with dist strategy."""

import tensorflow.compat.v2 as tf

import collections
import tempfile

from absl.testing import parameterized

            

Reported by Pylint.

              import collections
import tempfile

from absl.testing import parameterized
import numpy as np

import keras
from keras import losses
from keras.distribute import distribute_strategy_test as keras_test_lib

            

Reported by Pylint.

                    run.
  """

  def __init__(self):
    self.method_counts = collections.defaultdict(int)
    methods_to_count = [
        'on_batch_begin', 'on_batch_end', 'on_epoch_begin', 'on_epoch_end',
        'on_predict_batch_begin', 'on_predict_batch_end', 'on_predict_begin',
        'on_predict_end', 'on_test_batch_begin', 'on_test_batch_end',

            

Reported by Pylint.

                            tf.__internal__.distribute.combinations.mirrored_strategy_with_gpu_and_cpu,
          ],
          mode=['graph', 'eager']))
  def test_unsupported_features(self, distribution, mode):
    with self.cached_session():
      with distribution.scope():
        model = keras_test_lib.get_model()
        optimizer = tf.compat.v1.train.GradientDescentOptimizer(0.001)
        loss = 'mse'

            

Reported by Pylint.

                    self, distribution):
    with distribution.scope():

      class _SimpleMLP(keras.Model):

        def __init__(self, num_labels):
          super(_SimpleMLP, self).__init__()
          self.dense = keras.layers.Dense(num_labels)


            

Reported by Pylint.

                        super(_SimpleMLP, self).__init__()
          self.dense = keras.layers.Dense(num_labels)

        def call(self, inputs):
          return self.dense(inputs)

      model = _SimpleMLP(3)

      if not tf.executing_eagerly():

            

Reported by Pylint.

              class TestDistributionStrategyWithLossMasking(tf.test.TestCase,
                                              parameterized.TestCase):

  # TODO(priyag): Enable all strategies for this test. Currently it does not
  # work for TPU due to some invalid datatype.
  @tf.__internal__.distribute.combinations.generate(
      tf.__internal__.test.combinations.combine(
          distribution=[
              tf.__internal__.distribute.combinations.mirrored_strategy_with_gpu_and_cpu,

            

Reported by Pylint.

                          'mse')
        model.fit(dataset, epochs=1, steps_per_epoch=1)

        weights_file = tempfile.mktemp('.h5')
        model.save_weights(weights_file)

        model_2 = keras_test_lib.get_model()
        model_2.compile(
            optimizer(),

            

Reported by Bandit.

                        tf.__internal__.test.combinations.combine(
              optimizer=optimizer_combinations.rmsprop_optimizer_keras_v2_fn)))
  def test_save_load_trackable(self, distribution, optimizer):
    # TODO(b/123533246): Enable the test for TPU once bug is fixed
    if (isinstance(distribution,
                   (tf.distribute.experimental.TPUStrategy, tf.compat.v1.distribute.experimental.TPUStrategy)) and
        distribution.extended.steps_per_run > 1):
      self.skipTest('MultiStep TPU Strategy deadlocks with optimizer restore.')
    with self.cached_session():

            

Reported by Pylint.

                          'mse')
        model.fit(dataset, epochs=1, steps_per_epoch=1)

        weights_file = tempfile.mktemp()
        model.save_weights(weights_file)

        model_2 = keras_test_lib.get_model()
        model_2.compile(
            optimizer(),

            

Reported by Bandit.

              # ==============================================================================
"""Tests keras.layers.Layer works properly with mixed precision."""

import tensorflow.compat.v2 as tf

import os

from absl.testing import parameterized
import numpy as np

            

Reported by Pylint.

              
import os

from absl.testing import parameterized
import numpy as np
from keras import combinations
from keras import keras_parameterized
from keras import layers
from keras import models

            

Reported by Pylint.

                  class LayerWithIntVar(base_layer.Layer):

      def build(self, _):
        self.v = self.add_weight('v', dtype='int32', trainable=False)

      def call(self, inputs):
        # Only float variables should be autocasted. This will fail if self.v is
        # autocasted to float32
        return tf.cast(inputs, 'int32') + self.v

            

Reported by Pylint.

                    def build(self, _):
        self.v = self.add_weight('v', dtype='int32', trainable=False)

      def call(self, inputs):
        # Only float variables should be autocasted. This will fail if self.v is
        # autocasted to float32
        return tf.cast(inputs, 'int32') + self.v

    x = tf.constant([1.])

            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

import os

from absl.testing import parameterized
import numpy as np
from keras import combinations
from keras import keras_parameterized

            

Reported by Pylint.

              

class MultiplyLayerWithFunction(mp_test_util.MultiplyLayer):
  """Same as MultiplyLayer, but _multiply is decorated with a tf.function."""

  @tf.function
  def _multiply(self, x, y):
    return super(MultiplyLayerWithFunction, self)._multiply(x, y)


            

Reported by Pylint.

              class MultiplyLayerWithFunction(mp_test_util.MultiplyLayer):
  """Same as MultiplyLayer, but _multiply is decorated with a tf.function."""

  @tf.function
  def _multiply(self, x, y):
    return super(MultiplyLayerWithFunction, self)._multiply(x, y)


# If called outside any strategy.scope() calls, this will return the default

            

Reported by Pylint.

                """Same as MultiplyLayer, but _multiply is decorated with a tf.function."""

  @tf.function
  def _multiply(self, x, y):
    return super(MultiplyLayerWithFunction, self)._multiply(x, y)


# If called outside any strategy.scope() calls, this will return the default
# strategy.

            

Reported by Pylint.

              
  @tf.function
  def _multiply(self, x, y):
    return super(MultiplyLayerWithFunction, self)._multiply(x, y)


# If called outside any strategy.scope() calls, this will return the default
# strategy.
default_strategy_fn = tf.distribute.get_strategy

            

Reported by Pylint.

              
  @tf.function
  def _multiply(self, x, y):
    return super(MultiplyLayerWithFunction, self)._multiply(x, y)


# If called outside any strategy.scope() calls, this will return the default
# strategy.
default_strategy_fn = tf.distribute.get_strategy

            

Reported by Pylint.

              from __future__ import division
from __future__ import print_function

import tensorflow.compat.v2 as tf

import hashlib
import numbers
import sys
import types as python_types

            

Reported by Pylint.

              import sys
import types as python_types
import warnings
from keras.utils import generic_utils


class DropoutWrapperBase:
  """Operator adding dropout to inputs and outputs of the given cell."""


            

Reported by Pylint.

                      shape = convert_to_batch_shape(s)
        return tf.random.uniform(shape, seed=inner_seed, dtype=dtype)

      if (not isinstance(self._input_keep_prob, numbers.Real) or
          self._input_keep_prob < 1.0):
        if input_size is None:
          raise ValueError(
              "When variational_recurrent=True and input_keep_prob < 1.0 or "
              "is unknown, input_size must be provided")

            

Reported by Pylint.

                      return tf.random.uniform(shape, seed=inner_seed, dtype=dtype)

      if (not isinstance(self._input_keep_prob, numbers.Real) or
          self._input_keep_prob < 1.0):
        if input_size is None:
          raise ValueError(
              "When variational_recurrent=True and input_keep_prob < 1.0 or "
              "is unknown, input_size must be provided")
        self._recurrent_input_noise = _enumerated_map_structure_up_to(

            

Reported by Pylint.

              
  @property
  def wrapped_cell(self):
    return self.cell

  @property
  def state_size(self):
    return self.cell.state_size


            

Reported by Pylint.

              
  @property
  def state_size(self):
    return self.cell.state_size

  @property
  def output_size(self):
    return self.cell.output_size


            

Reported by Pylint.

              
  @property
  def output_size(self):
    return self.cell.output_size

  def build(self, inputs_shape):
    self.cell.build(inputs_shape)
    self.built = True


            

Reported by Pylint.

                  return self.cell.output_size

  def build(self, inputs_shape):
    self.cell.build(inputs_shape)
    self.built = True

  def zero_state(self, batch_size, dtype):
    with tf.name_scope(type(self).__name__ + "ZeroState"):
      return self.cell.zero_state(batch_size, dtype)

            

Reported by Pylint.

              
  def zero_state(self, batch_size, dtype):
    with tf.name_scope(type(self).__name__ + "ZeroState"):
      return self.cell.zero_state(batch_size, dtype)

  def _variational_recurrent_dropout_value(
      self, unused_index, value, noise, keep_prob):
    """Performs dropout given the pre-calculated noise tensor."""
    # uniform [keep_prob, 1.0 + keep_prob)

            

Reported by Pylint.

                  def _should_dropout(p):
      return (not isinstance(p, float)) or p < 1

    if _should_dropout(self._input_keep_prob):
      inputs = self._dropout(inputs, "input", self._recurrent_input_noise,
                             self._input_keep_prob)
    output, new_state = cell_call_fn(inputs, state, **kwargs)
    if _should_dropout(self._state_keep_prob):
      # Identify which subsets of the state to perform dropout on and

            

Reported by Pylint.

              # ==============================================================================
"""Functional tests for Ftrl operations."""

import tensorflow.compat.v2 as tf

import numpy as np
from keras.optimizer_v2 import ftrl



            

Reported by Pylint.

              import tensorflow.compat.v2 as tf

import numpy as np
from keras.optimizer_v2 import ftrl


class FtrlOptimizerTest(tf.test.TestCase):

  def doTestFtrlwithoutRegularization(self, use_resource=False):

            

Reported by Pylint.

              class FtrlOptimizerTest(tf.test.TestCase):

  def doTestFtrlwithoutRegularization(self, use_resource=False):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.float32]:
      with tf.Graph().as_default(), self.cached_session():
        if use_resource:
          var0 = tf.Variable([0.0, 0.0], dtype=dtype)
          var1 = tf.Variable([0.0, 0.0], dtype=dtype)

            

Reported by Pylint.

                  self.doTestFtrlwithoutRegularization(use_resource=True)

  def testFtrlwithoutRegularization2(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32]:
      with tf.Graph().as_default(), self.cached_session():
        var0 = tf.Variable([1.0, 2.0], dtype=dtype)
        var1 = tf.Variable([4.0, 3.0], dtype=dtype)
        grads0 = tf.constant([0.1, 0.2], dtype=dtype)

            

Reported by Pylint.

                          np.array([-0.28232238, -0.56096673]), v1_val)

  def testMinimizeSparseResourceVariable(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32, tf.float64]:
      with tf.Graph().as_default(), self.cached_session():
        var0 = tf.Variable([[1.0, 2.0]], dtype=dtype)
        x = tf.constant([[4.0], [5.0]], dtype=dtype)


            

Reported by Pylint.

                                                         atol=0.01)

  def testFtrlWithL1(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32]:
      with tf.Graph().as_default(), self.cached_session():
        var0 = tf.Variable([1.0, 2.0], dtype=dtype)
        var1 = tf.Variable([4.0, 3.0], dtype=dtype)
        grads0 = tf.constant([0.1, 0.2], dtype=dtype)

            

Reported by Pylint.

                          np.array([-0.93460727, -1.86147261]), v1_val)

  def testFtrlWithBeta(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32]:
      with tf.Graph().as_default(), self.cached_session():
        var0 = tf.Variable([1.0, 2.0], dtype=dtype)
        var1 = tf.Variable([4.0, 3.0], dtype=dtype)
        grads0 = tf.constant([0.1, 0.2], dtype=dtype)

            

Reported by Pylint.

                          np.array([-0.717741, -1.425132]), v1_val)

  def testFtrlWithL2_Beta(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32]:
      with tf.Graph().as_default(), self.cached_session():
        var0 = tf.Variable([1.0, 2.0], dtype=dtype)
        var1 = tf.Variable([4.0, 3.0], dtype=dtype)
        grads0 = tf.constant([0.1, 0.2], dtype=dtype)

            

Reported by Pylint.

                          np.array([-0.294335, -0.586556]), v1_val)

  def testFtrlWithL1_L2(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32]:
      with tf.Graph().as_default(), self.cached_session():
        var0 = tf.Variable([1.0, 2.0], dtype=dtype)
        var1 = tf.Variable([4.0, 3.0], dtype=dtype)
        grads0 = tf.constant([0.1, 0.2], dtype=dtype)

            

Reported by Pylint.

                  towards the origin causes the gradient descent trajectory to differ. The
    weights will tend to have smaller magnitudes with this parameter set.
    """
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    for dtype in [tf.half, tf.float32]:
      with tf.Graph().as_default(), self.cached_session():
        var0 = tf.Variable([1.0, 2.0], dtype=dtype)
        var1 = tf.Variable([4.0, 3.0], dtype=dtype)
        grads0 = tf.constant([0.1, 0.2], dtype=dtype)

            

Reported by Pylint.