# ==============================================================================
"""Tests for metrics_utils."""

import tensorflow.compat.v2 as tf

from absl.testing import parameterized
from keras import combinations
from keras.utils import metrics_utils


            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

from absl.testing import parameterized
from keras import combinations
from keras.utils import metrics_utils


@combinations.generate(combinations.combine(mode=['graph', 'eager']))

            

Reported by Pylint.

                def test_failing_different_ragged_and_dense_ranks(self, x_list, y_list):
    x = tf.ragged.constant(x_list)
    y = tf.ragged.constant(y_list)
    with self.assertRaises(ValueError):  # pylint: disable=g-error-prone-assert-raises
      [x, y
      ], _ = metrics_utils.ragged_assert_compatible_and_get_flat_values([x, y])

  @parameterized.parameters([
      {

            

Reported by Pylint.

                  x = tf.ragged.constant(x_list)
    y = tf.ragged.constant(y_list)
    mask = tf.ragged.constant(mask_list)
    with self.assertRaises(ValueError):  # pylint: disable=g-error-prone-assert-raises
      [x, y
      ], _ = metrics_utils.ragged_assert_compatible_and_get_flat_values([x, y],
                                                                        mask)

  # we do not support such cases that ragged_ranks are different but overall

            

Reported by Pylint.

                  # adding a ragged dimension
    x = tf.RaggedTensor.from_row_splits(dt, row_splits=[0, 1])
    y = tf.ragged.constant([[[[1, 2]]]])
    with self.assertRaises(ValueError):  # pylint: disable=g-error-prone-assert-raises
      [x, y], _ = \
          metrics_utils.ragged_assert_compatible_and_get_flat_values([x, y])


@combinations.generate(combinations.combine(mode=['graph', 'eager']))

            

Reported by Pylint.

              
  def test_one_dimensional(self):
    x = tf.constant([.3, .1, .2, -.5, 42.])
    top_1 = self.evaluate(metrics_utils._filter_top_k(x=x, k=1))
    top_2 = self.evaluate(metrics_utils._filter_top_k(x=x, k=2))
    top_3 = self.evaluate(metrics_utils._filter_top_k(x=x, k=3))

    self.assertAllClose(top_1, [
        metrics_utils.NEG_INF, metrics_utils.NEG_INF, metrics_utils.NEG_INF,

            

Reported by Pylint.

                def test_one_dimensional(self):
    x = tf.constant([.3, .1, .2, -.5, 42.])
    top_1 = self.evaluate(metrics_utils._filter_top_k(x=x, k=1))
    top_2 = self.evaluate(metrics_utils._filter_top_k(x=x, k=2))
    top_3 = self.evaluate(metrics_utils._filter_top_k(x=x, k=3))

    self.assertAllClose(top_1, [
        metrics_utils.NEG_INF, metrics_utils.NEG_INF, metrics_utils.NEG_INF,
        metrics_utils.NEG_INF, 42.

            

Reported by Pylint.

                  x = tf.constant([.3, .1, .2, -.5, 42.])
    top_1 = self.evaluate(metrics_utils._filter_top_k(x=x, k=1))
    top_2 = self.evaluate(metrics_utils._filter_top_k(x=x, k=2))
    top_3 = self.evaluate(metrics_utils._filter_top_k(x=x, k=3))

    self.assertAllClose(top_1, [
        metrics_utils.NEG_INF, metrics_utils.NEG_INF, metrics_utils.NEG_INF,
        metrics_utils.NEG_INF, 42.
    ])

            

Reported by Pylint.

                def test_three_dimensional(self):
    x = tf.constant([[[.3, .1, .2], [-.3, -.2, -.1]],
                              [[5., .2, 42.], [-.3, -.6, -.99]]])
    top_2 = self.evaluate(metrics_utils._filter_top_k(x=x, k=2))

    self.assertAllClose(
        top_2,
        [[[.3, metrics_utils.NEG_INF, .2], [metrics_utils.NEG_INF, -.2, -.1]],
         [[5., metrics_utils.NEG_INF, 42.], [-.3, -.6, metrics_utils.NEG_INF]]])

            

Reported by Pylint.

                    # This loses the static shape.
      x = tf.numpy_function(_identity, (x,), tf.float32)

      return metrics_utils._filter_top_k(x=x, k=2)

    x = tf.constant([.3, .1, .2, -.5, 42.])
    top_2 = self.evaluate(_filter_top_k(x))
    self.assertAllClose(top_2, [
        .3, metrics_utils.NEG_INF, metrics_utils.NEG_INF, metrics_utils.NEG_INF,

            

Reported by Pylint.

              # ==============================================================================
"""Keras text vectorization preprocessing layer."""

# pylint: disable=g-classes-have-attributes
# pylint: disable=g-direct-tensorflow-import

from keras import backend
from keras.engine import base_preprocessing_layer
from keras.layers.preprocessing import index_lookup

            

Reported by Pylint.

              """Keras text vectorization preprocessing layer."""

# pylint: disable=g-classes-have-attributes
# pylint: disable=g-direct-tensorflow-import

from keras import backend
from keras.engine import base_preprocessing_layer
from keras.layers.preprocessing import index_lookup
from keras.layers.preprocessing import preprocessing_utils as utils

            

Reported by Pylint.

              from keras.utils import layer_utils
from keras.utils import tf_utils
import numpy as np
import tensorflow.compat.v2 as tf
from tensorflow.python.util.tf_export import keras_export

LOWER_AND_STRIP_PUNCTUATION = "lower_and_strip_punctuation"

SPLIT_ON_WHITESPACE = "whitespace"

            

Reported by Pylint.

              from keras.utils import tf_utils
import numpy as np
import tensorflow.compat.v2 as tf
from tensorflow.python.util.tf_export import keras_export

LOWER_AND_STRIP_PUNCTUATION = "lower_and_strip_punctuation"

SPLIT_ON_WHITESPACE = "whitespace"


            

Reported by Pylint.

                  # expression to evaluate to False instead of True if the shape is undefined;
    # the expression needs to evaluate to True in that case.
    if self._split is not None:
      if input_shape.ndims > 1 and not input_shape[-1] == 1:  # pylint: disable=g-comparison-negation
        raise RuntimeError(
            "When using TextVectorization to tokenize strings, the innermost "
            "dimension of the input array must be 1, got shape "
            "{}".format(input_shape))


            

Reported by Pylint.

                ['', '[UNK]', 'earth', 'wind', 'and', 'fire']

  """
  # TODO(momernick): Add an examples section to the docstring.

  def __init__(self,
               max_tokens=None,
               standardize="lower_and_strip_punctuation",
               split="whitespace",

            

Reported by Pylint.

                    input_shape = tuple(input_shape) + (None,)
    return self._lookup_layer.compute_output_shape(input_shape)

  def compute_output_signature(self, input_spec):
    output_shape = self.compute_output_shape(input_spec.shape.as_list())
    output_dtype = (tf.int64 if self._output_mode == INT
                    else backend.floatx())
    return tf.TensorSpec(shape=output_shape, dtype=output_dtype)


            

Reported by Pylint.

              
    return inputs

  def call(self, inputs):
    if isinstance(inputs, (list, tuple, np.ndarray)):
      inputs = tf.convert_to_tensor(inputs)

    inputs = self._preprocess(inputs)


            

Reported by Pylint.

                  "keras.layers.TextVectorization",
    "keras.layers.experimental.preprocessing.TextVectorization",
    v1=[])
class TextVectorization(base_preprocessing_layer.PreprocessingLayer):
  """Text vectorization layer.

  This layer has basic options for managing text in a Keras model. It
  transforms a batch of strings (one example = one string) into either a list of
  token indices (one example = 1D tensor of integer token indices) or a dense

            

Reported by Pylint.

                  "keras.layers.experimental.preprocessing.TextVectorization",
    v1=[])
class TextVectorization(base_preprocessing_layer.PreprocessingLayer):
  """Text vectorization layer.

  This layer has basic options for managing text in a Keras model. It
  transforms a batch of strings (one example = one string) into either a list of
  token indices (one example = 1D tensor of integer token indices) or a dense
  representation (one example = 1D tensor of float values representing data

            

Reported by Pylint.

              # limitations under the License.
# ==============================================================================
"""TFDecorator-aware replacements for the inspect module."""
# pylint: disable=g-classes-have-attributes
import tensorflow.compat.v2 as tf

import collections
import functools
import inspect as _inspect

            

Reported by Pylint.

              # ==============================================================================
"""TFDecorator-aware replacements for the inspect module."""
# pylint: disable=g-classes-have-attributes
import tensorflow.compat.v2 as tf

import collections
import functools
import inspect as _inspect


            

Reported by Pylint.

              
  try:
    # Python3 will handle most callables here (not partial).
    return _getargspec(target)
  except TypeError:
    pass

  if isinstance(target, type):
    try:

            

Reported by Pylint.

              
  if isinstance(target, type):
    try:
      return _getargspec(target.__init__)
    except TypeError:
      pass

    try:
      return _getargspec(target.__new__)

            

Reported by Pylint.

                    pass

    try:
      return _getargspec(target.__new__)
    except TypeError:
      pass

  # The `type(target)` ensures that if a class is received we don't return
  # the signature of its __call__ method.

            

Reported by Pylint.

              
  # The `type(target)` ensures that if a class is received we don't return
  # the signature of its __call__ method.
  return _getargspec(type(target).__call__)


def _get_argspec_for_partial(obj):
  """Implements `getargspec` for `functools.partial` objects.


            

Reported by Pylint.

              # pylint: disable=g-classes-have-attributes
import tensorflow.compat.v2 as tf

import collections
import functools
import inspect as _inspect

ArgSpec = _inspect.ArgSpec


            

Reported by Pylint.

              import tensorflow.compat.v2 as tf

import collections
import functools
import inspect as _inspect

ArgSpec = _inspect.ArgSpec



            

Reported by Pylint.

              
import collections
import functools
import inspect as _inspect

ArgSpec = _inspect.ArgSpec


if hasattr(_inspect, 'FullArgSpec'):

            

Reported by Pylint.

              

if hasattr(_inspect, 'FullArgSpec'):
  FullArgSpec = _inspect.FullArgSpec  # pylint: disable=invalid-name
else:
  FullArgSpec = collections.namedtuple('FullArgSpec', [
      'args', 'varargs', 'varkw', 'defaults', 'kwonlyargs', 'kwonlydefaults',
      'annotations'
  ])

            

Reported by Pylint.

              from __future__ import division
from __future__ import print_function

import tensorflow as tf

from keras.benchmarks import benchmark_util


class TextWithTransformerBenchmark(tf.test.Benchmark):

            

Reported by Pylint.

              
import tensorflow as tf

from keras.benchmarks import benchmark_util


class TextWithTransformerBenchmark(tf.test.Benchmark):
  """Benchmarks for Text classification with Transformer
  using `tf.test.Benchmark`.

            

Reported by Pylint.

              

class TextWithTransformerBenchmark(tf.test.Benchmark):
  """Benchmarks for Text classification with Transformer
  using `tf.test.Benchmark`.
  """

  def __init__(self):
    super(TextWithTransformerBenchmark, self).__init__()

            

Reported by Pylint.

                using `tf.test.Benchmark`.
  """

  def __init__(self):
    super(TextWithTransformerBenchmark, self).__init__()
    self.max_feature = 20000
    self.max_len = 200
    (self.imdb_x, self.imdb_y), _ = tf.keras.datasets.imdb.load_data(
        num_words=self.max_feature)

            

Reported by Pylint.

                """

  def __init__(self):
    super(TextWithTransformerBenchmark, self).__init__()
    self.max_feature = 20000
    self.max_len = 200
    (self.imdb_x, self.imdb_y), _ = tf.keras.datasets.imdb.load_data(
        num_words=self.max_feature)
    self.imdb_x = tf.keras.preprocessing.sequence.pad_sequences(

            

Reported by Pylint.

                """

  def __init__(self):
    super(TextWithTransformerBenchmark, self).__init__()
    self.max_feature = 20000
    self.max_len = 200
    (self.imdb_x, self.imdb_y), _ = tf.keras.datasets.imdb.load_data(
        num_words=self.max_feature)
    self.imdb_x = tf.keras.preprocessing.sequence.pad_sequences(

            

Reported by Pylint.

              
  def __init__(self):
    super(TextWithTransformerBenchmark, self).__init__()
    self.max_feature = 20000
    self.max_len = 200
    (self.imdb_x, self.imdb_y), _ = tf.keras.datasets.imdb.load_data(
        num_words=self.max_feature)
    self.imdb_x = tf.keras.preprocessing.sequence.pad_sequences(
        self.imdb_x, maxlen=self.max_len)

            

Reported by Pylint.

                def __init__(self):
    super(TextWithTransformerBenchmark, self).__init__()
    self.max_feature = 20000
    self.max_len = 200
    (self.imdb_x, self.imdb_y), _ = tf.keras.datasets.imdb.load_data(
        num_words=self.max_feature)
    self.imdb_x = tf.keras.preprocessing.sequence.pad_sequences(
        self.imdb_x, maxlen=self.max_len)


            

Reported by Pylint.

                  super(TextWithTransformerBenchmark, self).__init__()
    self.max_feature = 20000
    self.max_len = 200
    (self.imdb_x, self.imdb_y), _ = tf.keras.datasets.imdb.load_data(
        num_words=self.max_feature)
    self.imdb_x = tf.keras.preprocessing.sequence.pad_sequences(
        self.imdb_x, maxlen=self.max_len)

  def _build_model(self):

            

Reported by Pylint.

                  self.max_len = 200
    (self.imdb_x, self.imdb_y), _ = tf.keras.datasets.imdb.load_data(
        num_words=self.max_feature)
    self.imdb_x = tf.keras.preprocessing.sequence.pad_sequences(
        self.imdb_x, maxlen=self.max_len)

  def _build_model(self):
    """Model from https://keras.io/examples/nlp/text_classification_with_transformer/."""
    embed_dim = 32

            

Reported by Pylint.

              # ==============================================================================
"""Test multi-worker Keras."""

import tensorflow.compat.v2 as tf

import collections
import copy
import functools
import json

            

Reported by Pylint.

              import sys
import threading

from absl.testing import parameterized

# pylint: disable=g-direct-tensorflow-import
import keras
from keras import backend
from keras import callbacks

            

Reported by Pylint.

              
from absl.testing import parameterized

# pylint: disable=g-direct-tensorflow-import
import keras
from keras import backend
from keras import callbacks
from keras import metrics as metrics_module
from keras import models

            

Reported by Pylint.

              from keras.optimizer_v2 import rmsprop
from keras.utils import kpl_test_utils

# pylint: disable=g-direct-tensorflow-import


def _clone_and_build_model(model, strategy):
  # The new "original" model in worker 0.
  with strategy.scope():

            

Reported by Pylint.

                # Compile and build model.
  if isinstance(model.optimizer, optimizer_v1.TFOptimizer):
    optimizer = model.optimizer
    # TODO(yuefengz): figure out why the optimizer here is still a
    # TFOptimizer.
    while isinstance(optimizer, optimizer_v1.TFOptimizer):
      optimizer = optimizer.optimizer
    optimizer = copy.deepcopy(optimizer)
  else:

            

Reported by Pylint.

                cloned_model.compile(
      optimizer,
      model.loss,
      metrics=metrics_module.clone_metrics(model._compile_metrics),
      loss_weights=model.loss_weights,
      sample_weight_mode=model.sample_weight_mode,
      weighted_metrics=metrics_module.clone_metrics(
          model._compile_weighted_metrics))
  return cloned_model

            

Reported by Pylint.

                    loss_weights=model.loss_weights,
      sample_weight_mode=model.sample_weight_mode,
      weighted_metrics=metrics_module.clone_metrics(
          model._compile_weighted_metrics))
  return cloned_model


# TODO(b/123918215): Possibly merge this Callback with keras_test.Counter.
class MultiWorkerVerificationCallback(callbacks.Callback):

            

Reported by Pylint.

                return cloned_model


# TODO(b/123918215): Possibly merge this Callback with keras_test.Counter.
class MultiWorkerVerificationCallback(callbacks.Callback):
  """MultiWorkerVerificationCallback verifies the callbacks in multi-worker scheme.

  This Callback is intended to be used for verifying the callback is indeed
  called the correct number of times in various task types.

            

Reported by Pylint.

                              of the two indices, and likewise for worker task.
  """

  # TODO(rchao): Add other method calls to verify.
  METHODS_TO_VERIFY = ['on_epoch_begin']

  def __init__(self, num_epoch, num_worker):
    """Initialize a MultiWorkerVerificationCallback.


            

Reported by Pylint.

                  }
    assert self._is_between_graph is not None
    if self._is_between_graph:
      # TODO(b/124171024): In between-graph replication, by default only the
      # chief calls callback. Fix this test to cover that, as well as the rare
      # cases where all workers call.
      worker_call_count = {
          i: method_count_dict for i in range(0, self._num_worker)
      }

            

Reported by Pylint.

              # ==============================================================================
"""Tests for Keras Premade Linear models."""

import tensorflow.compat.v2 as tf

import numpy as np
from keras import backend
from keras import keras_parameterized
from keras import losses

            

Reported by Pylint.

              

@keras_parameterized.run_all_keras_modes(always_skip_v1=True)
class LinearModelTest(keras_parameterized.TestCase):

  def test_linear_model_with_single_input(self):
    model = linear.LinearModel()
    inp = np.random.uniform(low=-5., high=5., size=(64, 2))
    output = .3 * inp[:, 0] + .2 * inp[:, 1]

            

Reported by Pylint.

              @keras_parameterized.run_all_keras_modes(always_skip_v1=True)
class LinearModelTest(keras_parameterized.TestCase):

  def test_linear_model_with_single_input(self):
    model = linear.LinearModel()
    inp = np.random.uniform(low=-5., high=5., size=(64, 2))
    output = .3 * inp[:, 0] + .2 * inp[:, 1]
    model.compile('sgd', 'mse', [])
    model.fit(inp, output, epochs=5)

            

Reported by Pylint.

              @keras_parameterized.run_all_keras_modes(always_skip_v1=True)
class LinearModelTest(keras_parameterized.TestCase):

  def test_linear_model_with_single_input(self):
    model = linear.LinearModel()
    inp = np.random.uniform(low=-5., high=5., size=(64, 2))
    output = .3 * inp[:, 0] + .2 * inp[:, 1]
    model.compile('sgd', 'mse', [])
    model.fit(inp, output, epochs=5)

            

Reported by Pylint.

              class LinearModelTest(keras_parameterized.TestCase):

  def test_linear_model_with_single_input(self):
    model = linear.LinearModel()
    inp = np.random.uniform(low=-5., high=5., size=(64, 2))
    output = .3 * inp[:, 0] + .2 * inp[:, 1]
    model.compile('sgd', 'mse', [])
    model.fit(inp, output, epochs=5)
    self.assertTrue(model.built)

            

Reported by Pylint.

              
  def test_linear_model_with_single_input(self):
    model = linear.LinearModel()
    inp = np.random.uniform(low=-5., high=5., size=(64, 2))
    output = .3 * inp[:, 0] + .2 * inp[:, 1]
    model.compile('sgd', 'mse', [])
    model.fit(inp, output, epochs=5)
    self.assertTrue(model.built)


            

Reported by Pylint.

                def test_linear_model_with_single_input(self):
    model = linear.LinearModel()
    inp = np.random.uniform(low=-5., high=5., size=(64, 2))
    output = .3 * inp[:, 0] + .2 * inp[:, 1]
    model.compile('sgd', 'mse', [])
    model.fit(inp, output, epochs=5)
    self.assertTrue(model.built)

  def test_linear_model_with_list_input(self):

            

Reported by Pylint.

                  model = linear.LinearModel()
    inp = np.random.uniform(low=-5., high=5., size=(64, 2))
    output = .3 * inp[:, 0] + .2 * inp[:, 1]
    model.compile('sgd', 'mse', [])
    model.fit(inp, output, epochs=5)
    self.assertTrue(model.built)

  def test_linear_model_with_list_input(self):
    model = linear.LinearModel()

            

Reported by Pylint.

                  inp = np.random.uniform(low=-5., high=5., size=(64, 2))
    output = .3 * inp[:, 0] + .2 * inp[:, 1]
    model.compile('sgd', 'mse', [])
    model.fit(inp, output, epochs=5)
    self.assertTrue(model.built)

  def test_linear_model_with_list_input(self):
    model = linear.LinearModel()
    input_a = np.random.uniform(low=-5., high=5., size=(64, 1))

            

Reported by Pylint.

                  output = .3 * inp[:, 0] + .2 * inp[:, 1]
    model.compile('sgd', 'mse', [])
    model.fit(inp, output, epochs=5)
    self.assertTrue(model.built)

  def test_linear_model_with_list_input(self):
    model = linear.LinearModel()
    input_a = np.random.uniform(low=-5., high=5., size=(64, 1))
    input_b = np.random.uniform(low=-5., high=5., size=(64, 1))

            

Reported by Pylint.

              # ==============================================================================
"""Tests for Nadam."""

import tensorflow.compat.v2 as tf

import numpy as np
from keras.optimizer_v2 import nadam



            

Reported by Pylint.

              import tensorflow.compat.v2 as tf

import numpy as np
from keras.optimizer_v2 import nadam


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)

            

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)
  beta_2_t = tf.cast(opt._get_hyper("beta_2"), dtype)
  beta_2_power = tf.pow(beta_2_t, local_step)
  return (beta_1_power, beta_2_power)


            

Reported by Pylint.

                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)
  beta_2_t = tf.cast(opt._get_hyper("beta_2"), dtype)
  beta_2_power = tf.pow(beta_2_t, local_step)
  return (beta_1_power, beta_2_power)


def update_m_cache(m_cache, t, beta1=0.9):

            

Reported by Pylint.

              class NadamOptimizerTest(tf.test.TestCase):

  def testSparse(self):
    # TODO(tanzheny, omalleyt): Fix test in eager mode.
    sparse_epsilon = 1e-7
    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, mcache = 0.0, 0.0, 0.0, 0.0, 1.0

            

Reported by Pylint.

                        self.assertAllCloseAccordingToType(var1_np, var1)

  def testBasic(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, mcache = 0.0, 0.0, 0.0, 0.0, 1.0
        var0_np = np.array([1.0, 2.0], dtype=dtype.as_numpy_dtype)

            

Reported by Pylint.

              from keras.optimizer_v2 import nadam


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)
  beta_2_t = tf.cast(opt._get_hyper("beta_2"), dtype)
  beta_2_power = tf.pow(beta_2_t, local_step)

            

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)
  beta_2_t = tf.cast(opt._get_hyper("beta_2"), dtype)
  beta_2_power = tf.pow(beta_2_t, local_step)
  return (beta_1_power, beta_2_power)

            

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)
  beta_2_t = tf.cast(opt._get_hyper("beta_2"), dtype)
  beta_2_power = tf.pow(beta_2_t, local_step)
  return (beta_1_power, beta_2_power)


            

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)
  beta_2_t = tf.cast(opt._get_hyper("beta_2"), dtype)
  beta_2_power = tf.pow(beta_2_t, local_step)
  return (beta_1_power, beta_2_power)



            

Reported by Pylint.

              # ==============================================================================
"""Tests for evaluation using Keras model and ParameterServerStrategy."""

import tensorflow.compat.v2 as tf

import time

import keras
from tensorflow.python.distribute import multi_worker_test_base

            

Reported by Pylint.

              import time

import keras
from tensorflow.python.distribute import multi_worker_test_base
from tensorflow.python.distribute.cluster_resolver import SimpleClusterResolver
from tensorflow.python.ops import resource_variable_ops


# TODO(yuefengz): move the following implementation to Keras core.

            

Reported by Pylint.

              
import keras
from tensorflow.python.distribute import multi_worker_test_base
from tensorflow.python.distribute.cluster_resolver import SimpleClusterResolver
from tensorflow.python.ops import resource_variable_ops


# TODO(yuefengz): move the following implementation to Keras core.
class KerasMetricTypeSpec(tf.TypeSpec):

            

Reported by Pylint.

              import keras
from tensorflow.python.distribute import multi_worker_test_base
from tensorflow.python.distribute.cluster_resolver import SimpleClusterResolver
from tensorflow.python.ops import resource_variable_ops


# TODO(yuefengz): move the following implementation to Keras core.
class KerasMetricTypeSpec(tf.TypeSpec):


            

Reported by Pylint.

              from tensorflow.python.ops import resource_variable_ops


# TODO(yuefengz): move the following implementation to Keras core.
class KerasMetricTypeSpec(tf.TypeSpec):

  def __init__(self, cls, config, weights):
    self._cls = cls
    self._config = config

            

Reported by Pylint.

                  return self._cls

  def most_specific_compatible_type(self, other):
    if (type(self) is not type(other) or self._cls != other._cls or
        self._config != other._config):
      raise ValueError("No TypeSpec is compatible with both %s and %s" %
                       (self, other))
    return KerasMetricTypeSpec(self._cls, self._config, self._weights)


            

Reported by Pylint.

              
  def most_specific_compatible_type(self, other):
    if (type(self) is not type(other) or self._cls != other._cls or
        self._config != other._config):
      raise ValueError("No TypeSpec is compatible with both %s and %s" %
                       (self, other))
    return KerasMetricTypeSpec(self._cls, self._config, self._weights)

  @property

            

Reported by Pylint.

              
    def fetch_variable(next_creator, **kwargs):
      del next_creator, kwargs
      # TODO(yuefengz): verify the var creation order matches the weights
      # property
      var = weights[counter[0]]
      counter[0] += 1
      return var


            

Reported by Pylint.

                  self.assertEqual(metric2.result(), 0.0)

    tf.nest.assert_same_structure(
        metric1, metric2._type_spec, expand_composites=True)
    tf.nest.assert_same_structure(
        metric1._type_spec, metric2, expand_composites=True)

    @tf.function
    def func(m):

            

Reported by Pylint.

                  tf.nest.assert_same_structure(
        metric1, metric2._type_spec, expand_composites=True)
    tf.nest.assert_same_structure(
        metric1._type_spec, metric2, expand_composites=True)

    @tf.function
    def func(m):
      m.update_state([1.0, 2.0])


            

Reported by Pylint.

              # ==============================================================================
"""Tests for imagenet_utils."""

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 keras_parameterized
from keras.applications import imagenet_utils as utils

            

Reported by Pylint.

              from keras.mixed_precision.policy import set_policy


class TestImageNetUtils(keras_parameterized.TestCase):

  def test_preprocess_input(self):
    # Test invalid mode check
    x = np.random.uniform(0, 255, (10, 10, 3))
    with self.assertRaises(ValueError):

            

Reported by Pylint.

              
class TestImageNetUtils(keras_parameterized.TestCase):

  def test_preprocess_input(self):
    # Test invalid mode check
    x = np.random.uniform(0, 255, (10, 10, 3))
    with self.assertRaises(ValueError):
      utils.preprocess_input(x, mode='some_unknown_mode')


            

Reported by Pylint.

              
class TestImageNetUtils(keras_parameterized.TestCase):

  def test_preprocess_input(self):
    # Test invalid mode check
    x = np.random.uniform(0, 255, (10, 10, 3))
    with self.assertRaises(ValueError):
      utils.preprocess_input(x, mode='some_unknown_mode')


            

Reported by Pylint.

              
  def test_preprocess_input(self):
    # Test invalid mode check
    x = np.random.uniform(0, 255, (10, 10, 3))
    with self.assertRaises(ValueError):
      utils.preprocess_input(x, mode='some_unknown_mode')

    # Test image batch with float and int image input
    x = np.random.uniform(0, 255, (2, 10, 10, 3))

            

Reported by Pylint.

              
  def test_preprocess_input(self):
    # Test invalid mode check
    x = np.random.uniform(0, 255, (10, 10, 3))
    with self.assertRaises(ValueError):
      utils.preprocess_input(x, mode='some_unknown_mode')

    # Test image batch with float and int image input
    x = np.random.uniform(0, 255, (2, 10, 10, 3))

            

Reported by Pylint.

                def test_preprocess_input(self):
    # Test invalid mode check
    x = np.random.uniform(0, 255, (10, 10, 3))
    with self.assertRaises(ValueError):
      utils.preprocess_input(x, mode='some_unknown_mode')

    # Test image batch with float and int image input
    x = np.random.uniform(0, 255, (2, 10, 10, 3))
    xint = x.astype('int32')

            

Reported by Pylint.

                  # Test invalid mode check
    x = np.random.uniform(0, 255, (10, 10, 3))
    with self.assertRaises(ValueError):
      utils.preprocess_input(x, mode='some_unknown_mode')

    # Test image batch with float and int image input
    x = np.random.uniform(0, 255, (2, 10, 10, 3))
    xint = x.astype('int32')
    self.assertEqual(utils.preprocess_input(x).shape, x.shape)

            

Reported by Pylint.

                    utils.preprocess_input(x, mode='some_unknown_mode')

    # Test image batch with float and int image input
    x = np.random.uniform(0, 255, (2, 10, 10, 3))
    xint = x.astype('int32')
    self.assertEqual(utils.preprocess_input(x).shape, x.shape)
    self.assertEqual(utils.preprocess_input(xint).shape, xint.shape)

    out1 = utils.preprocess_input(x, 'channels_last')

            

Reported by Pylint.

              # pylint: disable=protected-access
"""Input layer code (`Input` and `InputLayer`)."""

import tensorflow.compat.v2 as tf
from keras import backend
from keras.distribute import distributed_training_utils
from keras.engine import base_layer
from keras.engine import keras_tensor
from keras.engine import node as node_module

            

Reported by Pylint.

              from keras.saving.saved_model import layer_serialization
from keras.utils import tf_utils
from keras.utils import traceback_utils
from tensorflow.python.util.tf_export import keras_export


def _assert_other_arg_none(arg_name, arg):
  if arg is not None:
    raise ValueError('When `type_spec` is not None, all other args '

            

Reported by Pylint.

              

def _assert_other_arg_none(arg_name, arg):
  if arg is not None:
    raise ValueError('When `type_spec` is not None, all other args '
                     'except `name` must be None, '
                     'but %s is not None.' % arg_name)



            

Reported by Pylint.

              
def _assert_other_arg_none(arg_name, arg):
  if arg is not None:
    raise ValueError('When `type_spec` is not None, all other args '
                     'except `name` must be None, '
                     'but %s is not None.' % arg_name)


@keras_export('keras.layers.InputLayer')

            

Reported by Pylint.

              

@keras_export('keras.layers.InputLayer')
class InputLayer(base_layer.Layer):
  """Layer to be used as an entry point into a Network (a graph of layers).

  It can either wrap an existing tensor (pass an `input_tensor` argument)
  or create a placeholder tensor (pass arguments `input_shape`, and
  optionally, `dtype`).

            

Reported by Pylint.

              
@keras_export('keras.layers.InputLayer')
class InputLayer(base_layer.Layer):
  """Layer to be used as an entry point into a Network (a graph of layers).

  It can either wrap an existing tensor (pass an `input_tensor` argument)
  or create a placeholder tensor (pass arguments `input_shape`, and
  optionally, `dtype`).


            

Reported by Pylint.

                    name: Optional name of the layer (string).
  """

  @traceback_utils.filter_traceback
  def __init__(self,
               input_shape=None,
               batch_size=None,
               dtype=None,
               input_tensor=None,

            

Reported by Pylint.

                """

  @traceback_utils.filter_traceback
  def __init__(self,
               input_shape=None,
               batch_size=None,
               dtype=None,
               input_tensor=None,
               sparse=None,

            

Reported by Pylint.

                """

  @traceback_utils.filter_traceback
  def __init__(self,
               input_shape=None,
               batch_size=None,
               dtype=None,
               input_tensor=None,
               sparse=None,

            

Reported by Pylint.

                """

  @traceback_utils.filter_traceback
  def __init__(self,
               input_shape=None,
               batch_size=None,
               dtype=None,
               input_tensor=None,
               sparse=None,

            

Reported by Pylint.