# ==============================================================================
"""A simple network to use in tests and examples."""

import tensorflow.compat.v2 as tf
from keras.optimizer_v2 import optimizer_v2


def minimize_loss_example(optimizer, use_bias=False, use_callable_loss=True):
  """Example of non-distribution-aware legacy code."""

            

Reported by Pylint.

              
  def dataset_fn():
    dataset = tf.data.Dataset.from_tensors([[1.]]).repeat()
    # TODO(isaprykin): batch with drop_remainder causes shapes to be
    # fully defined for TPU.  Remove this when XLA supports dynamic shapes.
    return dataset.batch(1, drop_remainder=True)

  layer = tf.compat.v1.layers.Dense(1, use_bias=use_bias)


            

Reported by Pylint.

              

def minimize_loss_example(optimizer, use_bias=False, use_callable_loss=True):
  """Example of non-distribution-aware legacy code."""

  def dataset_fn():
    dataset = tf.data.Dataset.from_tensors([[1.]]).repeat()
    # TODO(isaprykin): batch with drop_remainder causes shapes to be
    # fully defined for TPU.  Remove this when XLA supports dynamic shapes.

            

Reported by Pylint.

              def minimize_loss_example(optimizer, use_bias=False, use_callable_loss=True):
  """Example of non-distribution-aware legacy code."""

  def dataset_fn():
    dataset = tf.data.Dataset.from_tensors([[1.]]).repeat()
    # TODO(isaprykin): batch with drop_remainder causes shapes to be
    # fully defined for TPU.  Remove this when XLA supports dynamic shapes.
    return dataset.batch(1, drop_remainder=True)


            

Reported by Pylint.

                """Example of non-distribution-aware legacy code."""

  def dataset_fn():
    dataset = tf.data.Dataset.from_tensors([[1.]]).repeat()
    # TODO(isaprykin): batch with drop_remainder causes shapes to be
    # fully defined for TPU.  Remove this when XLA supports dynamic shapes.
    return dataset.batch(1, drop_remainder=True)

  layer = tf.compat.v1.layers.Dense(1, use_bias=use_bias)

            

Reported by Pylint.

                  dataset = tf.data.Dataset.from_tensors([[1.]]).repeat()
    # TODO(isaprykin): batch with drop_remainder causes shapes to be
    # fully defined for TPU.  Remove this when XLA supports dynamic shapes.
    return dataset.batch(1, drop_remainder=True)

  layer = tf.compat.v1.layers.Dense(1, use_bias=use_bias)

  def model_fn(x):
    """A very simple model written by the user."""

            

Reported by Pylint.

                  # fully defined for TPU.  Remove this when XLA supports dynamic shapes.
    return dataset.batch(1, drop_remainder=True)

  layer = tf.compat.v1.layers.Dense(1, use_bias=use_bias)

  def model_fn(x):
    """A very simple model written by the user."""

    def loss_fn():

            

Reported by Pylint.

              
  layer = tf.compat.v1.layers.Dense(1, use_bias=use_bias)

  def model_fn(x):
    """A very simple model written by the user."""

    def loss_fn():
      y = tf.reshape(layer(x), []) - tf.constant(1.)
      return y * y

            

Reported by Pylint.

              
  layer = tf.compat.v1.layers.Dense(1, use_bias=use_bias)

  def model_fn(x):
    """A very simple model written by the user."""

    def loss_fn():
      y = tf.reshape(layer(x), []) - tf.constant(1.)
      return y * y

            

Reported by Pylint.

                layer = tf.compat.v1.layers.Dense(1, use_bias=use_bias)

  def model_fn(x):
    """A very simple model written by the user."""

    def loss_fn():
      y = tf.reshape(layer(x), []) - tf.constant(1.)
      return y * y


            

Reported by Pylint.

              # ==============================================================================
"""Benchmark for KPL implementation of vocabulary columns from files with dense inputs."""

import tensorflow as tf

import os

import keras
from tensorflow.python.eager.def_function import function as tf_function

            

Reported by Pylint.

              
import os

import keras
from tensorflow.python.eager.def_function import function as tf_function
from keras.layers.preprocessing import string_lookup
from keras.layers.preprocessing.benchmarks import feature_column_benchmark as fc_bm

# This is required as of 3/2021 because otherwise we drop into graph mode.

            

Reported by Pylint.

              import os

import keras
from tensorflow.python.eager.def_function import function as tf_function
from keras.layers.preprocessing import string_lookup
from keras.layers.preprocessing.benchmarks import feature_column_benchmark as fc_bm

# This is required as of 3/2021 because otherwise we drop into graph mode.
tf.compat.v1.enable_v2_behavior()

            

Reported by Pylint.

              
import keras
from tensorflow.python.eager.def_function import function as tf_function
from keras.layers.preprocessing import string_lookup
from keras.layers.preprocessing.benchmarks import feature_column_benchmark as fc_bm

# This is required as of 3/2021 because otherwise we drop into graph mode.
tf.compat.v1.enable_v2_behavior()


            

Reported by Pylint.

              import keras
from tensorflow.python.eager.def_function import function as tf_function
from keras.layers.preprocessing import string_lookup
from keras.layers.preprocessing.benchmarks import feature_column_benchmark as fc_bm

# This is required as of 3/2021 because otherwise we drop into graph mode.
tf.compat.v1.enable_v2_behavior()

NUM_REPEATS = 10

            

Reported by Pylint.

              
import tensorflow as tf

import os

import keras
from tensorflow.python.eager.def_function import function as tf_function
from keras.layers.preprocessing import string_lookup
from keras.layers.preprocessing.benchmarks import feature_column_benchmark as fc_bm

            

Reported by Pylint.

              

class BenchmarkLayer(tf.test.TestCase, fc_bm.LayerBenchmark):
  """Benchmark the layer forward pass."""

  def _write_to_temp_file(self, file_name, vocab_list):
    vocab_path = os.path.join(self.get_temp_dir(), file_name + ".txt")
    with tf.io.gfile.GFile(vocab_path, "w") as writer:
      for vocab in vocab_list:

            

Reported by Pylint.

              class BenchmarkLayer(tf.test.TestCase, fc_bm.LayerBenchmark):
  """Benchmark the layer forward pass."""

  def _write_to_temp_file(self, file_name, vocab_list):
    vocab_path = os.path.join(self.get_temp_dir(), file_name + ".txt")
    with tf.io.gfile.GFile(vocab_path, "w") as writer:
      for vocab in vocab_list:
        writer.write(vocab + "\n")
      writer.flush()

            

Reported by Pylint.

                """Benchmark the layer forward pass."""

  def _write_to_temp_file(self, file_name, vocab_list):
    vocab_path = os.path.join(self.get_temp_dir(), file_name + ".txt")
    with tf.io.gfile.GFile(vocab_path, "w") as writer:
      for vocab in vocab_list:
        writer.write(vocab + "\n")
      writer.flush()
      writer.close()

            

Reported by Pylint.

              
  def _write_to_temp_file(self, file_name, vocab_list):
    vocab_path = os.path.join(self.get_temp_dir(), file_name + ".txt")
    with tf.io.gfile.GFile(vocab_path, "w") as writer:
      for vocab in vocab_list:
        writer.write(vocab + "\n")
      writer.flush()
      writer.close()
    return vocab_path

            

Reported by Pylint.

                  return iter(self._dict)

  # abstract from collections.MutableSet
  def add(self, item):
    self._dict[item] = None

  # abstract from collections.MutableSet
  def discard(self, value):
    del self._dict[value]

            

Reported by Pylint.

              

class OrderPreservingSet(collections.abc.MutableSet):
  """A set based on dict so that it preserves key insertion order."""

  def __init__(self, iterable=None):
    self._dict = {item: None for item in (iterable or [])}

  # abstract from collections.MutableSet

            

Reported by Pylint.

              class OrderPreservingSet(collections.abc.MutableSet):
  """A set based on dict so that it preserves key insertion order."""

  def __init__(self, iterable=None):
    self._dict = {item: None for item in (iterable or [])}

  # abstract from collections.MutableSet
  def __len__(self):
    return len(self._dict)

            

Reported by Pylint.

                """A set based on dict so that it preserves key insertion order."""

  def __init__(self, iterable=None):
    self._dict = {item: None for item in (iterable or [])}

  # abstract from collections.MutableSet
  def __len__(self):
    return len(self._dict)


            

Reported by Pylint.

                  self._dict = {item: None for item in (iterable or [])}

  # abstract from collections.MutableSet
  def __len__(self):
    return len(self._dict)

  # abstract from collections.MutableSet
  def __contains__(self, value):
    return value in self._dict

            

Reported by Pylint.

              
  # abstract from collections.MutableSet
  def __len__(self):
    return len(self._dict)

  # abstract from collections.MutableSet
  def __contains__(self, value):
    return value in self._dict


            

Reported by Pylint.

                  return len(self._dict)

  # abstract from collections.MutableSet
  def __contains__(self, value):
    return value in self._dict

  # override from collections.MutableSet
  def __iter__(self):
    return iter(self._dict)

            

Reported by Pylint.

              
  # abstract from collections.MutableSet
  def __contains__(self, value):
    return value in self._dict

  # override from collections.MutableSet
  def __iter__(self):
    return iter(self._dict)


            

Reported by Pylint.

                  return value in self._dict

  # override from collections.MutableSet
  def __iter__(self):
    return iter(self._dict)

  # abstract from collections.MutableSet
  def add(self, item):
    self._dict[item] = None

            

Reported by Pylint.

              
  # override from collections.MutableSet
  def __iter__(self):
    return iter(self._dict)

  # abstract from collections.MutableSet
  def add(self, item):
    self._dict[item] = None


            

Reported by Pylint.

              # ==============================================================================
"""Distribution tests for keras.layers.preprocessing.text_vectorization."""

import tensorflow.compat.v2 as tf

import numpy as np

import keras
from keras import backend

            

Reported by Pylint.

                  preprocessing_test_utils.PreprocessingLayerTest):

  def test_distribution_strategy_output(self, strategy):
    # TODO(b/180614455): remove this check when MLIR bridge is always enabled.
    if backend.is_tpu_strategy(strategy):
      self.skipTest("This test needs MLIR bridge on TPU.")

    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],

            

Reported by Pylint.

                  self.assertAllEqual(expected_output, output_dataset)

  def test_distribution_strategy_output_with_adapt(self, strategy):
    # TODO(b/180614455): remove this check when MLIR bridge is always enabled.
    if backend.is_tpu_strategy(strategy):
      self.skipTest("This test needs MLIR bridge on TPU.")

    vocab_data = [[
        "earth", "earth", "earth", "earth", "wind", "wind", "wind", "and",

            

Reported by Pylint.

                      strategy=strategy_combinations.all_strategies +
        strategy_combinations.multi_worker_mirrored_strategies,
        mode=["eager"]))
class TextVectorizationDistributionTest(
    keras_parameterized.TestCase,
    preprocessing_test_utils.PreprocessingLayerTest):

  def test_distribution_strategy_output(self, strategy):
    # TODO(b/180614455): remove this check when MLIR bridge is always enabled.

            

Reported by Pylint.

                  keras_parameterized.TestCase,
    preprocessing_test_utils.PreprocessingLayerTest):

  def test_distribution_strategy_output(self, strategy):
    # TODO(b/180614455): remove this check when MLIR bridge is always enabled.
    if backend.is_tpu_strategy(strategy):
      self.skipTest("This test needs MLIR bridge on TPU.")

    vocab_data = ["earth", "wind", "and", "fire"]

            

Reported by Pylint.

                  keras_parameterized.TestCase,
    preprocessing_test_utils.PreprocessingLayerTest):

  def test_distribution_strategy_output(self, strategy):
    # TODO(b/180614455): remove this check when MLIR bridge is always enabled.
    if backend.is_tpu_strategy(strategy):
      self.skipTest("This test needs MLIR bridge on TPU.")

    vocab_data = ["earth", "wind", "and", "fire"]

            

Reported by Pylint.

              
  def test_distribution_strategy_output(self, strategy):
    # TODO(b/180614455): remove this check when MLIR bridge is always enabled.
    if backend.is_tpu_strategy(strategy):
      self.skipTest("This test needs MLIR bridge on TPU.")

    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])

            

Reported by Pylint.

                def test_distribution_strategy_output(self, strategy):
    # TODO(b/180614455): remove this check when MLIR bridge is always enabled.
    if backend.is_tpu_strategy(strategy):
      self.skipTest("This test needs MLIR bridge on TPU.")

    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    input_dataset = tf.data.Dataset.from_tensor_slices(input_array).batch(

            

Reported by Pylint.

                  if backend.is_tpu_strategy(strategy):
      self.skipTest("This test needs MLIR bridge on TPU.")

    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    input_dataset = tf.data.Dataset.from_tensor_slices(input_array).batch(
        2, drop_remainder=True)


            

Reported by Pylint.

                    self.skipTest("This test needs MLIR bridge on TPU.")

    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    input_dataset = tf.data.Dataset.from_tensor_slices(input_array).batch(
        2, drop_remainder=True)

    expected_output = [[2, 3, 4, 5], [5, 4, 2, 1]]

            

Reported by Pylint.

              # ==============================================================================
"""Benchmark for KPL implementation of vocabulary columns from files with varying-length inputs."""

import tensorflow as tf

import os

import keras
from tensorflow.python.eager.def_function import function as tf_function

            

Reported by Pylint.

              
import os

import keras
from tensorflow.python.eager.def_function import function as tf_function
from keras.layers.preprocessing import string_lookup
from keras.layers.preprocessing.benchmarks import feature_column_benchmark as fc_bm

# This is required as of 3/2021 because otherwise we drop into graph mode.

            

Reported by Pylint.

              import os

import keras
from tensorflow.python.eager.def_function import function as tf_function
from keras.layers.preprocessing import string_lookup
from keras.layers.preprocessing.benchmarks import feature_column_benchmark as fc_bm

# This is required as of 3/2021 because otherwise we drop into graph mode.
tf.compat.v1.enable_v2_behavior()

            

Reported by Pylint.

              
import keras
from tensorflow.python.eager.def_function import function as tf_function
from keras.layers.preprocessing import string_lookup
from keras.layers.preprocessing.benchmarks import feature_column_benchmark as fc_bm

# This is required as of 3/2021 because otherwise we drop into graph mode.
tf.compat.v1.enable_v2_behavior()


            

Reported by Pylint.

              import keras
from tensorflow.python.eager.def_function import function as tf_function
from keras.layers.preprocessing import string_lookup
from keras.layers.preprocessing.benchmarks import feature_column_benchmark as fc_bm

# This is required as of 3/2021 because otherwise we drop into graph mode.
tf.compat.v1.enable_v2_behavior()

NUM_REPEATS = 10

            

Reported by Pylint.

              
import tensorflow as tf

import os

import keras
from tensorflow.python.eager.def_function import function as tf_function
from keras.layers.preprocessing import string_lookup
from keras.layers.preprocessing.benchmarks import feature_column_benchmark as fc_bm

            

Reported by Pylint.

              

class BenchmarkLayer(tf.test.TestCase, fc_bm.LayerBenchmark):
  """Benchmark the layer forward pass."""

  def _write_to_temp_file(self, file_name, vocab_list):
    vocab_path = os.path.join(self.get_temp_dir(), file_name + ".txt")
    with tf.io.gfile.GFile(vocab_path, "w") as writer:
      for vocab in vocab_list:

            

Reported by Pylint.

              class BenchmarkLayer(tf.test.TestCase, fc_bm.LayerBenchmark):
  """Benchmark the layer forward pass."""

  def _write_to_temp_file(self, file_name, vocab_list):
    vocab_path = os.path.join(self.get_temp_dir(), file_name + ".txt")
    with tf.io.gfile.GFile(vocab_path, "w") as writer:
      for vocab in vocab_list:
        writer.write(vocab + "\n")
      writer.flush()

            

Reported by Pylint.

                """Benchmark the layer forward pass."""

  def _write_to_temp_file(self, file_name, vocab_list):
    vocab_path = os.path.join(self.get_temp_dir(), file_name + ".txt")
    with tf.io.gfile.GFile(vocab_path, "w") as writer:
      for vocab in vocab_list:
        writer.write(vocab + "\n")
      writer.flush()
      writer.close()

            

Reported by Pylint.

              
  def _write_to_temp_file(self, file_name, vocab_list):
    vocab_path = os.path.join(self.get_temp_dir(), file_name + ".txt")
    with tf.io.gfile.GFile(vocab_path, "w") as writer:
      for vocab in vocab_list:
        writer.write(vocab + "\n")
      writer.flush()
      writer.close()
    return vocab_path

            

Reported by Pylint.

              # ==============================================================================
"""Benchmark for Keras category_encoding preprocessing layer."""

import tensorflow as tf

import time

import numpy as np


            

Reported by Pylint.

              
import numpy as np

import keras
from keras.layers.preprocessing import category_encoding

tf.compat.v1.enable_v2_behavior()



            

Reported by Pylint.

              import numpy as np

import keras
from keras.layers.preprocessing import category_encoding

tf.compat.v1.enable_v2_behavior()


class BenchmarkLayer(tf.test.Benchmark):

            

Reported by Pylint.

              
import tensorflow as tf

import time

import numpy as np

import keras
from keras.layers.preprocessing import category_encoding

            

Reported by Pylint.

              

class BenchmarkLayer(tf.test.Benchmark):
  """Benchmark the layer forward pass."""

  def run_dataset_implementation(self, output_mode, batch_size, sequence_length,
                                 max_tokens):
    input_t = keras.Input(shape=(sequence_length,), dtype=tf.int32)
    layer = category_encoding.CategoryEncoding(

            

Reported by Pylint.

              class BenchmarkLayer(tf.test.Benchmark):
  """Benchmark the layer forward pass."""

  def run_dataset_implementation(self, output_mode, batch_size, sequence_length,
                                 max_tokens):
    input_t = keras.Input(shape=(sequence_length,), dtype=tf.int32)
    layer = category_encoding.CategoryEncoding(
        max_tokens=max_tokens, output_mode=output_mode)
    _ = layer(input_t)

            

Reported by Pylint.

              class BenchmarkLayer(tf.test.Benchmark):
  """Benchmark the layer forward pass."""

  def run_dataset_implementation(self, output_mode, batch_size, sequence_length,
                                 max_tokens):
    input_t = keras.Input(shape=(sequence_length,), dtype=tf.int32)
    layer = category_encoding.CategoryEncoding(
        max_tokens=max_tokens, output_mode=output_mode)
    _ = layer(input_t)

            

Reported by Pylint.

              class BenchmarkLayer(tf.test.Benchmark):
  """Benchmark the layer forward pass."""

  def run_dataset_implementation(self, output_mode, batch_size, sequence_length,
                                 max_tokens):
    input_t = keras.Input(shape=(sequence_length,), dtype=tf.int32)
    layer = category_encoding.CategoryEncoding(
        max_tokens=max_tokens, output_mode=output_mode)
    _ = layer(input_t)

            

Reported by Pylint.

              
  def run_dataset_implementation(self, output_mode, batch_size, sequence_length,
                                 max_tokens):
    input_t = keras.Input(shape=(sequence_length,), dtype=tf.int32)
    layer = category_encoding.CategoryEncoding(
        max_tokens=max_tokens, output_mode=output_mode)
    _ = layer(input_t)

    num_repeats = 5

            

Reported by Pylint.

                def run_dataset_implementation(self, output_mode, batch_size, sequence_length,
                                 max_tokens):
    input_t = keras.Input(shape=(sequence_length,), dtype=tf.int32)
    layer = category_encoding.CategoryEncoding(
        max_tokens=max_tokens, output_mode=output_mode)
    _ = layer(input_t)

    num_repeats = 5
    starts = []

            

Reported by Pylint.

              # ==============================================================================
"""Utility methods related to kernelized layers."""

import tensorflow.compat.v2 as tf


def _to_matrix(u):
  """If input tensor is a vector (i.e., has rank 1), converts it to matrix."""
  u_rank = len(u.shape)

            

Reported by Pylint.

              import tensorflow.compat.v2 as tf


def _to_matrix(u):
  """If input tensor is a vector (i.e., has rank 1), converts it to matrix."""
  u_rank = len(u.shape)
  if u_rank not in [1, 2]:
    raise ValueError('The input tensor should have rank 1 or 2. '
                     f'Received rank: {u_rank}')

            

Reported by Pylint.

              

def _to_matrix(u):
  """If input tensor is a vector (i.e., has rank 1), converts it to matrix."""
  u_rank = len(u.shape)
  if u_rank not in [1, 2]:
    raise ValueError('The input tensor should have rank 1 or 2. '
                     f'Received rank: {u_rank}')
  if u_rank == 1:

            

Reported by Pylint.

              
def _to_matrix(u):
  """If input tensor is a vector (i.e., has rank 1), converts it to matrix."""
  u_rank = len(u.shape)
  if u_rank not in [1, 2]:
    raise ValueError('The input tensor should have rank 1 or 2. '
                     f'Received rank: {u_rank}')
  if u_rank == 1:
    return tf.expand_dims(u, 0)

            

Reported by Pylint.

              def _to_matrix(u):
  """If input tensor is a vector (i.e., has rank 1), converts it to matrix."""
  u_rank = len(u.shape)
  if u_rank not in [1, 2]:
    raise ValueError('The input tensor should have rank 1 or 2. '
                     f'Received rank: {u_rank}')
  if u_rank == 1:
    return tf.expand_dims(u, 0)
  return u

            

Reported by Pylint.

                """If input tensor is a vector (i.e., has rank 1), converts it to matrix."""
  u_rank = len(u.shape)
  if u_rank not in [1, 2]:
    raise ValueError('The input tensor should have rank 1 or 2. '
                     f'Received rank: {u_rank}')
  if u_rank == 1:
    return tf.expand_dims(u, 0)
  return u


            

Reported by Pylint.

                if u_rank not in [1, 2]:
    raise ValueError('The input tensor should have rank 1 or 2. '
                     f'Received rank: {u_rank}')
  if u_rank == 1:
    return tf.expand_dims(u, 0)
  return u


def _align_matrices(x, y):

            

Reported by Pylint.

                  raise ValueError('The input tensor should have rank 1 or 2. '
                     f'Received rank: {u_rank}')
  if u_rank == 1:
    return tf.expand_dims(u, 0)
  return u


def _align_matrices(x, y):
  """Aligns x and y tensors to allow computations over pairs of their rows."""

            

Reported by Pylint.

                                   f'Received rank: {u_rank}')
  if u_rank == 1:
    return tf.expand_dims(u, 0)
  return u


def _align_matrices(x, y):
  """Aligns x and y tensors to allow computations over pairs of their rows."""
  x_matrix = _to_matrix(x)

            

Reported by Pylint.

                return u


def _align_matrices(x, y):
  """Aligns x and y tensors to allow computations over pairs of their rows."""
  x_matrix = _to_matrix(x)
  y_matrix = _to_matrix(y)
  x_shape = x_matrix.shape
  y_shape = y_matrix.shape

            

Reported by Pylint.

              # ==============================================================================
"""Tests for trackable object SavedModel save."""

import tensorflow.compat.v2 as tf

import os
from tensorflow.python.framework import test_util
from keras.layers import core
from keras.optimizer_v2 import adam

            

Reported by Pylint.

              import tensorflow.compat.v2 as tf

import os
from tensorflow.python.framework import test_util
from keras.layers import core
from keras.optimizer_v2 import adam


class _ModelWithOptimizerUsingDefun(tf.train.Checkpoint):

            

Reported by Pylint.

              
import os
from tensorflow.python.framework import test_util
from keras.layers import core
from keras.optimizer_v2 import adam


class _ModelWithOptimizerUsingDefun(tf.train.Checkpoint):


            

Reported by Pylint.

              import os
from tensorflow.python.framework import test_util
from keras.layers import core
from keras.optimizer_v2 import adam


class _ModelWithOptimizerUsingDefun(tf.train.Checkpoint):

  def __init__(self):

            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

import os
from tensorflow.python.framework import test_util
from keras.layers import core
from keras.optimizer_v2 import adam



            

Reported by Pylint.

              from keras.optimizer_v2 import adam


class _ModelWithOptimizerUsingDefun(tf.train.Checkpoint):

  def __init__(self):
    self.dense = core.Dense(1)
    self.optimizer = adam.Adam(0.01)


            

Reported by Pylint.

              
class _ModelWithOptimizerUsingDefun(tf.train.Checkpoint):

  def __init__(self):
    self.dense = core.Dense(1)
    self.optimizer = adam.Adam(0.01)

  @tf.function(
      input_signature=(tf.TensorSpec([None, 2], tf.float32),

            

Reported by Pylint.

              class _ModelWithOptimizerUsingDefun(tf.train.Checkpoint):

  def __init__(self):
    self.dense = core.Dense(1)
    self.optimizer = adam.Adam(0.01)

  @tf.function(
      input_signature=(tf.TensorSpec([None, 2], tf.float32),
                       tf.TensorSpec([None], tf.float32)),

            

Reported by Pylint.

              
  def __init__(self):
    self.dense = core.Dense(1)
    self.optimizer = adam.Adam(0.01)

  @tf.function(
      input_signature=(tf.TensorSpec([None, 2], tf.float32),
                       tf.TensorSpec([None], tf.float32)),
  )

            

Reported by Pylint.

                  self.dense = core.Dense(1)
    self.optimizer = adam.Adam(0.01)

  @tf.function(
      input_signature=(tf.TensorSpec([None, 2], tf.float32),
                       tf.TensorSpec([None], tf.float32)),
  )
  def call(self, x, y):
    with tf.GradientTape() as tape:

            

Reported by Pylint.

              # limitations under the License.
# ==============================================================================
"""Tests for pickling / deepcopying of Keras Models."""
# pylint: disable=g-bad-import-order
import tensorflow.compat.v2 as tf

import copy
import pickle
import numpy as np

            

Reported by Pylint.

              # ==============================================================================
"""Tests for pickling / deepcopying of Keras Models."""
# pylint: disable=g-bad-import-order
import tensorflow.compat.v2 as tf

import copy
import pickle
import numpy as np


            

Reported by Pylint.

                @keras_parameterized.parameterized.named_parameters(
      ('copy', copy.copy), ('deepcopy', copy.deepcopy),
      *((f'pickle_protocol_level_{protocol}',
         lambda model: pickle.loads(pickle.dumps(model, protocol=protocol)))  # pylint: disable=cell-var-from-loop
        for protocol in range(pickle.HIGHEST_PROTOCOL + 1)))
  def test_built_models(self, serializer):
    """Built models should be copyable and picklable for all model types."""

    model = testing_utils.get_small_mlp(

            

Reported by Bandit.

              # pylint: disable=g-bad-import-order
import tensorflow.compat.v2 as tf

import copy
import pickle
import numpy as np

from keras import keras_parameterized
from keras import testing_utils

            

Reported by Pylint.

              import tensorflow.compat.v2 as tf

import copy
import pickle
import numpy as np

from keras import keras_parameterized
from keras import testing_utils


            

Reported by Bandit.

              import tensorflow.compat.v2 as tf

import copy
import pickle
import numpy as np

from keras import keras_parameterized
from keras import testing_utils


            

Reported by Pylint.

              

class TestPickleProtocol(keras_parameterized.TestCase):
  """Tests pickle protoocol support."""

  @keras_parameterized.run_with_all_model_types
  @keras_parameterized.parameterized.named_parameters(
      ('copy', copy.copy), ('deepcopy', copy.deepcopy),
      *((f'pickle_protocol_level_{protocol}',

            

Reported by Pylint.

              class TestPickleProtocol(keras_parameterized.TestCase):
  """Tests pickle protoocol support."""

  @keras_parameterized.run_with_all_model_types
  @keras_parameterized.parameterized.named_parameters(
      ('copy', copy.copy), ('deepcopy', copy.deepcopy),
      *((f'pickle_protocol_level_{protocol}',
         lambda model: pickle.loads(pickle.dumps(model, protocol=protocol)))  # pylint: disable=cell-var-from-loop
        for protocol in range(pickle.HIGHEST_PROTOCOL + 1)))

            

Reported by Pylint.

                """Tests pickle protoocol support."""

  @keras_parameterized.run_with_all_model_types
  @keras_parameterized.parameterized.named_parameters(
      ('copy', copy.copy), ('deepcopy', copy.deepcopy),
      *((f'pickle_protocol_level_{protocol}',
         lambda model: pickle.loads(pickle.dumps(model, protocol=protocol)))  # pylint: disable=cell-var-from-loop
        for protocol in range(pickle.HIGHEST_PROTOCOL + 1)))
  def test_built_models(self, serializer):

            

Reported by Pylint.

                    *((f'pickle_protocol_level_{protocol}',
         lambda model: pickle.loads(pickle.dumps(model, protocol=protocol)))  # pylint: disable=cell-var-from-loop
        for protocol in range(pickle.HIGHEST_PROTOCOL + 1)))
  def test_built_models(self, serializer):
    """Built models should be copyable and picklable for all model types."""

    model = testing_utils.get_small_mlp(
        num_hidden=1, num_classes=2, input_dim=3)
    model.compile(optimizer='sgd', loss='sparse_categorical_crossentropy')

            

Reported by Pylint.

              # ==============================================================================
"""Benchmark tests for Keras metrics memory consumption."""

import tensorflow as tf

import numpy as np

try:
  import memory_profiler  # pylint:disable=g-import-not-at-top

            

Reported by Pylint.

              import numpy as np

try:
  import memory_profiler  # pylint:disable=g-import-not-at-top
except ImportError:
  memory_profiler = None


class KerasMetricMemoryBenchmark(tf.test.Benchmark):

            

Reported by Pylint.

                    self.skipTest('Skip test since memory_profiler is not available.')

    with tf.compat.forward_compatibility_horizon(2021, 6, 9):
      self.y_true = np.random.randint(2, size=(1024, 1024))
      self.y_pred = np.random.rand(1024, 1024)

      memory_usage_1 = memory_profiler.memory_usage((self.even_thresholds_auc))
      memory_usage_2 = memory_profiler.memory_usage(
          (self.uneven_thresholds_auc))

            

Reported by Pylint.

              
    with tf.compat.forward_compatibility_horizon(2021, 6, 9):
      self.y_true = np.random.randint(2, size=(1024, 1024))
      self.y_pred = np.random.rand(1024, 1024)

      memory_usage_1 = memory_profiler.memory_usage((self.even_thresholds_auc))
      memory_usage_2 = memory_profiler.memory_usage(
          (self.uneven_thresholds_auc))
      # memory usage is a list of number which sampled when running the function

            

Reported by Pylint.

              
  def even_thresholds_auc(self):
    auc = tf.keras.metrics.AUC(num_thresholds=200)
    self.assertTrue(auc._thresholds_distributed_evenly)

    auc(self.y_true, self.y_pred)

  def uneven_thresholds_auc(self):
    num_thresholds = 200

            

Reported by Pylint.

                  thresholds = thresholds[1:-1]

    auc = tf.keras.metrics.AUC(thresholds=thresholds)
    self.assertFalse(auc._thresholds_distributed_evenly)
    self.assertEqual(auc.num_thresholds, num_thresholds)

    auc(self.y_true, self.y_pred)



            

Reported by Pylint.

              import numpy as np

try:
  import memory_profiler  # pylint:disable=g-import-not-at-top
except ImportError:
  memory_profiler = None


class KerasMetricMemoryBenchmark(tf.test.Benchmark):

            

Reported by Pylint.

              try:
  import memory_profiler  # pylint:disable=g-import-not-at-top
except ImportError:
  memory_profiler = None


class KerasMetricMemoryBenchmark(tf.test.Benchmark):

  # This test is added to measure the memory footprint for

            

Reported by Pylint.

                memory_profiler = None


class KerasMetricMemoryBenchmark(tf.test.Benchmark):

  # This test is added to measure the memory footprint for
  # metrics_utils._update_confusion_matrix_variables_optimized().

  def benchmark_auc_memory_usage(self):

            

Reported by Pylint.

                # This test is added to measure the memory footprint for
  # metrics_utils._update_confusion_matrix_variables_optimized().

  def benchmark_auc_memory_usage(self):
    if memory_profiler is None:
      self.skipTest('Skip test since memory_profiler is not available.')

    with tf.compat.forward_compatibility_horizon(2021, 6, 9):
      self.y_true = np.random.randint(2, size=(1024, 1024))

            

Reported by Pylint.