# ==============================================================================
"""Tests for RNN cell wrapper v2 implementation."""

import tensorflow.compat.v2 as tf

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

            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

from absl.testing import parameterized
import numpy as np
from keras import combinations
from keras import layers
from keras.layers import rnn_cell_wrapper_v2
from keras.layers.legacy_rnn import rnn_cell_impl

            

Reported by Pylint.

                def testWrapperV2Caller(self, wrapper):
    """Tests that wrapper V2 is using the LayerRNNCell's caller."""

    with legacy_base_layer.keras_style_scope():
      base_cell = rnn_cell_impl.MultiRNNCell(
          [rnn_cell_impl.BasicRNNCell(1) for _ in range(2)])
    rnn_cell = wrapper(base_cell)
    inputs = tf.convert_to_tensor([[1]], dtype=tf.float32)
    state = tf.convert_to_tensor([[1]], dtype=tf.float32)

            

Reported by Pylint.

                      bias_initializer=tf.compat.v1.constant_initializer(0.5))
    g, m_new = base_cell(x, m)
    wrapper_object = wrapper_type(base_cell)
    (name, dep), = wrapper_object._checkpoint_dependencies
    wrapper_object.get_config()  # Should not throw an error
    self.assertIs(dep, base_cell)
    self.assertEqual("cell", name)

    g_res, m_new_res = wrapper_object(x, m)

            

Reported by Pylint.

                  m = tf.zeros([1, 3])
    cell = rnn_cell_impl.GRUCell(3)
    wrapped_cell = wrapper_type(cell, "/cpu:0")
    (name, dep), = wrapped_cell._checkpoint_dependencies
    wrapped_cell.get_config()  # Should not throw an error
    self.assertIs(dep, cell)
    self.assertEqual("cell", name)

    outputs, _ = wrapped_cell(x, m)

            

Reported by Pylint.

                  self.assertIsNone(getattr(wrapped_cell_v2, "_keras_style", None))

    wrapped_cell = wrapper(rnn_cell_impl.BasicRNNCell(1))
    self.assertFalse(wrapped_cell._keras_style)

  @parameterized.parameters(
      [rnn_cell_wrapper_v2.DropoutWrapper, rnn_cell_wrapper_v2.ResidualWrapper])
  def testWrapperWeights(self, wrapper):
    """Tests that wrapper weights contain wrapped cells weights."""

            

Reported by Pylint.

                  inputs = tf.convert_to_tensor([[1]], dtype=tf.float32)
    state = tf.convert_to_tensor([[1]], dtype=tf.float32)
    _ = rnn_cell(inputs, [state, state])
    weights = base_cell._cells[0].weights
    self.assertLen(weights, expected_len=2)
    self.assertTrue(all("_wrapper" in v.name for v in weights))

  @parameterized.parameters(
      [rnn_cell_wrapper_v2.DropoutWrapper, rnn_cell_wrapper_v2.ResidualWrapper])

            

Reported by Pylint.

              
    reconstructed_wrapper = wrapper_cls.from_config(config)
    # Assert the reconstructed function will perform the math correctly.
    self.assertEqual(reconstructed_wrapper._residual_fn(1, 2), 4)

    def residual_fn(inputs, outputs):
      return inputs * 3 + outputs

    wrapper = wrapper_cls(cell, residual_fn=residual_fn)

            

Reported by Pylint.

              
    reconstructed_wrapper = wrapper_cls.from_config(config)
    # Assert the reconstructed function will perform the math correctly.
    self.assertEqual(reconstructed_wrapper._residual_fn(1, 2), 5)

  def testDropoutWrapperSerialization(self):
    wrapper_cls = rnn_cell_wrapper_v2.DropoutWrapper
    cell = layers.GRUCell(10)
    wrapper = wrapper_cls(cell)

            

Reported by Pylint.

                  config = wrapper.get_config()

    reconstructed_wrapper = wrapper_cls.from_config(config)
    self.assertTrue(reconstructed_wrapper._dropout_state_filter(None))

    def dropout_state_filter_visitor(unused_state):
      return False

    wrapper = wrapper_cls(

            

Reported by Pylint.

                    https://arxiv.org/abs/1512.03385) (CVPR 2015)
"""

import tensorflow.compat.v2 as tf

from keras import backend
from keras.applications import imagenet_utils
from keras.engine import training
from keras.layers import VersionAwareLayers

            

Reported by Pylint.

              from keras.layers import VersionAwareLayers
from keras.utils import data_utils
from keras.utils import layer_utils
from tensorflow.python.util.tf_export import keras_export


BASE_WEIGHTS_PATH = (
    'https://storage.googleapis.com/tensorflow/keras-applications/resnet/')
WEIGHTS_HASHES = {

            

Reported by Pylint.

                Returns:
    A `keras.Model` instance.
  """
  global layers
  if 'layers' in kwargs:
    layers = kwargs.pop('layers')
  else:
    layers = VersionAwareLayers()
  if kwargs:

            

Reported by Pylint.

              layers = None


def ResNet(stack_fn,
           preact,
           use_bias,
           model_name='resnet',
           include_top=True,
           weights='imagenet',

            

Reported by Pylint.

              layers = None


def ResNet(stack_fn,
           preact,
           use_bias,
           model_name='resnet',
           include_top=True,
           weights='imagenet',

            

Reported by Pylint.

              layers = None


def ResNet(stack_fn,
           preact,
           use_bias,
           model_name='resnet',
           include_top=True,
           weights='imagenet',

            

Reported by Pylint.

              layers = None


def ResNet(stack_fn,
           preact,
           use_bias,
           model_name='resnet',
           include_top=True,
           weights='imagenet',

            

Reported by Pylint.

                         classes=1000,
           classifier_activation='softmax',
           **kwargs):
  """Instantiates the ResNet, ResNetV2, and ResNeXt architecture.

  Args:
    stack_fn: a function that returns output tensor for the
      stacked residual blocks.
    preact: whether to use pre-activation or not

            

Reported by Pylint.

                Returns:
    A `keras.Model` instance.
  """
  global layers
  if 'layers' in kwargs:
    layers = kwargs.pop('layers')
  else:
    layers = VersionAwareLayers()
  if kwargs:

            

Reported by Pylint.

                  A `keras.Model` instance.
  """
  global layers
  if 'layers' in kwargs:
    layers = kwargs.pop('layers')
  else:
    layers = VersionAwareLayers()
  if kwargs:
    raise ValueError('Unknown argument(s): %s' % (kwargs,))

            

Reported by Pylint.

              # ==============================================================================
"""Correctness tests for tf.keras DNN model using DistributionStrategy."""

import tensorflow.compat.v2 as tf

import numpy as np

import keras
from keras import backend

            

Reported by Pylint.

              class TestDistributionStrategyDnnCorrectness(
    keras_correctness_test_base.TestDistributionStrategyCorrectnessBase):

  def get_model(self,
                initial_weights=None,
                distribution=None,
                input_shapes=None):
    with keras_correctness_test_base.MaybeDistributionScope(distribution):
      # We add few non-linear layers to make it non-trivial.

            

Reported by Pylint.

                  self.run_dynamic_lr_test(distribution)


class TestDistributionStrategyDnnMetricCorrectness(
    keras_correctness_test_base.TestDistributionStrategyCorrectnessBase):

  def get_model(self,
                distribution=None,
                input_shapes=None):

            

Reported by Pylint.

                  self.run_dynamic_lr_test(distribution)


class TestDistributionStrategyDnnMetricCorrectness(
    keras_correctness_test_base.TestDistributionStrategyCorrectnessBase):

  def get_model(self,
                distribution=None,
                input_shapes=None):

            

Reported by Pylint.

                  self.run_metric_correctness_test(distribution)


class TestDistributionStrategyDnnMetricEvalCorrectness(
    keras_correctness_test_base.TestDistributionStrategyCorrectnessBase):

  def get_model(self,
                distribution=None,
                input_shapes=None):

            

Reported by Pylint.

                  self.run_metric_correctness_test(distribution)


class TestDistributionStrategyDnnMetricEvalCorrectness(
    keras_correctness_test_base.TestDistributionStrategyCorrectnessBase):

  def get_model(self,
                distribution=None,
                input_shapes=None):

            

Reported by Pylint.

                  self.run_eval_metrics_correctness_test(distribution)


class SubclassedModel(keras.Model):

  def __init__(self, initial_weights, input_shapes):
    super(SubclassedModel, self).__init__()
    self.dense1 = keras.layers.Dense(10, activation='relu', input_shape=(1,))
    self.dense2 = keras.layers.Dense(

            

Reported by Pylint.

                  if initial_weights:
      self.set_weights(initial_weights)

  def call(self, inputs):
    x = self.dense1(inputs)
    x = self.dense2(x)
    x = self.dense3(x)
    return self.dense4(x)


            

Reported by Pylint.

              from keras.optimizer_v2 import gradient_descent as gradient_descent_keras


def all_strategy_combinations_with_eager_and_graph_modes():
  return (tf.__internal__.test.combinations.combine(
      distribution=strategy_combinations.all_strategies,
      mode=['graph', 'eager']) + tf.__internal__.test.combinations.combine(
          distribution=strategy_combinations.multi_worker_mirrored_strategies,
          mode='eager'))

            

Reported by Pylint.

              

def all_strategy_combinations_with_eager_and_graph_modes():
  return (tf.__internal__.test.combinations.combine(
      distribution=strategy_combinations.all_strategies,
      mode=['graph', 'eager']) + tf.__internal__.test.combinations.combine(
          distribution=strategy_combinations.multi_worker_mirrored_strategies,
          mode='eager'))


            

Reported by Pylint.

              # ==============================================================================
"""Tests for categorical preprocessing layers."""

import tensorflow.compat.v2 as tf

import numpy as np
from keras import keras_parameterized
from keras import testing_utils
from keras.engine import input_layer

            

Reported by Pylint.

              

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

  def test_crossing_sparse_inputs(self):
    layer = category_crossing.CategoryCrossing()
    inputs_0 = tf.SparseTensor(
        indices=[[0, 0], [1, 0], [1, 1]],

            

Reported by Pylint.

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

  def test_crossing_sparse_inputs(self):
    layer = category_crossing.CategoryCrossing()
    inputs_0 = tf.SparseTensor(
        indices=[[0, 0], [1, 0], [1, 1]],
        values=['a', 'b', 'c'],
        dense_shape=[2, 2])

            

Reported by Pylint.

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

  def test_crossing_sparse_inputs(self):
    layer = category_crossing.CategoryCrossing()
    inputs_0 = tf.SparseTensor(
        indices=[[0, 0], [1, 0], [1, 1]],
        values=['a', 'b', 'c'],
        dense_shape=[2, 2])

            

Reported by Pylint.

              class CategoryCrossingTest(keras_parameterized.TestCase):

  def test_crossing_sparse_inputs(self):
    layer = category_crossing.CategoryCrossing()
    inputs_0 = tf.SparseTensor(
        indices=[[0, 0], [1, 0], [1, 1]],
        values=['a', 'b', 'c'],
        dense_shape=[2, 2])
    inputs_1 = tf.SparseTensor(

            

Reported by Pylint.

              
  def test_crossing_sparse_inputs(self):
    layer = category_crossing.CategoryCrossing()
    inputs_0 = tf.SparseTensor(
        indices=[[0, 0], [1, 0], [1, 1]],
        values=['a', 'b', 'c'],
        dense_shape=[2, 2])
    inputs_1 = tf.SparseTensor(
        indices=[[0, 1], [1, 2]], values=['d', 'e'], dense_shape=[2, 3])

            

Reported by Pylint.

                      indices=[[0, 0], [1, 0], [1, 1]],
        values=['a', 'b', 'c'],
        dense_shape=[2, 2])
    inputs_1 = tf.SparseTensor(
        indices=[[0, 1], [1, 2]], values=['d', 'e'], dense_shape=[2, 3])
    output = layer([inputs_0, inputs_1])
    self.assertAllClose(np.asarray([[0, 0], [1, 0], [1, 1]]), output.indices)
    self.assertAllEqual([b'a_X_d', b'b_X_e', b'c_X_e'], output.values)


            

Reported by Pylint.

                      dense_shape=[2, 2])
    inputs_1 = tf.SparseTensor(
        indices=[[0, 1], [1, 2]], values=['d', 'e'], dense_shape=[2, 3])
    output = layer([inputs_0, inputs_1])
    self.assertAllClose(np.asarray([[0, 0], [1, 0], [1, 1]]), output.indices)
    self.assertAllEqual([b'a_X_d', b'b_X_e', b'c_X_e'], output.values)

  def test_crossing_sparse_inputs_custom_sep(self):
    layer = category_crossing.CategoryCrossing(separator='_Y_')

            

Reported by Pylint.

                  inputs_1 = tf.SparseTensor(
        indices=[[0, 1], [1, 2]], values=['d', 'e'], dense_shape=[2, 3])
    output = layer([inputs_0, inputs_1])
    self.assertAllClose(np.asarray([[0, 0], [1, 0], [1, 1]]), output.indices)
    self.assertAllEqual([b'a_X_d', b'b_X_e', b'c_X_e'], output.values)

  def test_crossing_sparse_inputs_custom_sep(self):
    layer = category_crossing.CategoryCrossing(separator='_Y_')
    inputs_0 = tf.SparseTensor(

            

Reported by Pylint.

                      indices=[[0, 1], [1, 2]], values=['d', 'e'], dense_shape=[2, 3])
    output = layer([inputs_0, inputs_1])
    self.assertAllClose(np.asarray([[0, 0], [1, 0], [1, 1]]), output.indices)
    self.assertAllEqual([b'a_X_d', b'b_X_e', b'c_X_e'], output.values)

  def test_crossing_sparse_inputs_custom_sep(self):
    layer = category_crossing.CategoryCrossing(separator='_Y_')
    inputs_0 = tf.SparseTensor(
        indices=[[0, 0], [1, 0], [1, 1]],

            

Reported by Pylint.

              from __future__ import division
from __future__ import print_function

import tensorflow.compat.v2 as tf

from tensorflow.python.framework import test_util
from keras.legacy_tf_layers import pooling as pooling_layers



            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

from tensorflow.python.framework import test_util
from keras.legacy_tf_layers import pooling as pooling_layers


class PoolingTest(tf.test.TestCase):


            

Reported by Pylint.

              from keras.legacy_tf_layers import pooling as pooling_layers


class PoolingTest(tf.test.TestCase):

  def testInvalidDataFormat(self):
    height, width = 7, 9
    images = tf.random.uniform((5, height, width, 3), seed=1)
    with self.assertRaisesRegex(ValueError, 'data_format'):

            

Reported by Pylint.

              
class PoolingTest(tf.test.TestCase):

  def testInvalidDataFormat(self):
    height, width = 7, 9
    images = tf.random.uniform((5, height, width, 3), seed=1)
    with self.assertRaisesRegex(ValueError, 'data_format'):
      pooling_layers.max_pooling2d(images, 3, strides=2, data_format='invalid')


            

Reported by Pylint.

              
class PoolingTest(tf.test.TestCase):

  def testInvalidDataFormat(self):
    height, width = 7, 9
    images = tf.random.uniform((5, height, width, 3), seed=1)
    with self.assertRaisesRegex(ValueError, 'data_format'):
      pooling_layers.max_pooling2d(images, 3, strides=2, data_format='invalid')


            

Reported by Pylint.

              
class PoolingTest(tf.test.TestCase):

  def testInvalidDataFormat(self):
    height, width = 7, 9
    images = tf.random.uniform((5, height, width, 3), seed=1)
    with self.assertRaisesRegex(ValueError, 'data_format'):
      pooling_layers.max_pooling2d(images, 3, strides=2, data_format='invalid')


            

Reported by Pylint.

              class PoolingTest(tf.test.TestCase):

  def testInvalidDataFormat(self):
    height, width = 7, 9
    images = tf.random.uniform((5, height, width, 3), seed=1)
    with self.assertRaisesRegex(ValueError, 'data_format'):
      pooling_layers.max_pooling2d(images, 3, strides=2, data_format='invalid')

  def testInvalidStrides(self):

            

Reported by Pylint.

              
  def testInvalidDataFormat(self):
    height, width = 7, 9
    images = tf.random.uniform((5, height, width, 3), seed=1)
    with self.assertRaisesRegex(ValueError, 'data_format'):
      pooling_layers.max_pooling2d(images, 3, strides=2, data_format='invalid')

  def testInvalidStrides(self):
    height, width = 7, 9

            

Reported by Pylint.

                def testInvalidDataFormat(self):
    height, width = 7, 9
    images = tf.random.uniform((5, height, width, 3), seed=1)
    with self.assertRaisesRegex(ValueError, 'data_format'):
      pooling_layers.max_pooling2d(images, 3, strides=2, data_format='invalid')

  def testInvalidStrides(self):
    height, width = 7, 9
    images = tf.random.uniform((5, height, width, 3), seed=1)

            

Reported by Pylint.

                  height, width = 7, 9
    images = tf.random.uniform((5, height, width, 3), seed=1)
    with self.assertRaisesRegex(ValueError, 'data_format'):
      pooling_layers.max_pooling2d(images, 3, strides=2, data_format='invalid')

  def testInvalidStrides(self):
    height, width = 7, 9
    images = tf.random.uniform((5, height, width, 3), seed=1)
    with self.assertRaisesRegex(ValueError, 'strides'):

            

Reported by Pylint.

              # ==============================================================================
"""Tests specific to deferred-build `Sequential` models."""

import tensorflow.compat.v2 as tf

import os
import unittest
import numpy as np


            

Reported by Pylint.

              from keras import testing_utils

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


class TestDeferredSequential(keras_parameterized.TestCase):

            

Reported by Pylint.

                  model = get_model()
    model(np.random.random((2, 6)))
    self.assertLen(model.weights, 4)
    self.assertTrue(model._is_graph_network)
    self.assertLen(model.inputs, 1)
    self.assertLen(model.outputs, 1)
    self.assertEqual(model.inputs[0].shape.as_list(), [2, 6])
    self.assertEqual(model.outputs[0].shape.as_list(), [2, 2])


            

Reported by Pylint.

                  model = get_model()
    model.build((None, 6))
    self.assertLen(model.weights, 4)
    self.assertTrue(model._is_graph_network)
    self.assertLen(model.inputs, 1)
    self.assertLen(model.outputs, 1)
    self.assertEqual(model.inputs[0].shape.as_list(), [None, 6])
    self.assertEqual(model.outputs[0].shape.as_list(), [None, 2])


            

Reported by Pylint.

                      run_eagerly=testing_utils.should_run_eagerly())
    model.fit(np.zeros((2, 6)), np.zeros((2, 2)))
    self.assertLen(model.weights, 4)
    self.assertTrue(model._is_graph_network)
    self.assertLen(model.inputs, 1)
    self.assertLen(model.outputs, 1)
    # Inconsistency here: with eager `fit`, the model is built with shape
    # (2, 6), but with graph function `fit`, it is built with shape `(None, 6)`.
    # This is likely due to our assumption "the batch size should be dynamic"

            

Reported by Pylint.

                  model = get_model()
    model.build((None, 6))
    self.assertTrue(model.built)
    self.assertTrue(model._is_graph_network)
    self.assertLen(model.layers, 3)
    self.assertLen(model.weights, 4)
    model.pop()
    self.assertTrue(model.built)
    self.assertTrue(model._is_graph_network)

            

Reported by Pylint.

                  self.assertLen(model.weights, 4)
    model.pop()
    self.assertTrue(model.built)
    self.assertTrue(model._is_graph_network)
    self.assertLen(model.layers, 2)
    self.assertLen(model.weights, 2)
    model.add(keras.layers.Dense(2))
    self.assertTrue(model.built)
    self.assertTrue(model._is_graph_network)

            

Reported by Pylint.

                  self.assertLen(model.weights, 2)
    model.add(keras.layers.Dense(2))
    self.assertTrue(model.built)
    self.assertTrue(model._is_graph_network)
    self.assertLen(model.layers, 3)
    self.assertLen(model.weights, 4)

  @keras_parameterized.run_all_keras_modes(always_skip_v1=True)
  def test_feature_extraction(self):

            

Reported by Pylint.

                  path = os.path.join(self.get_temp_dir(), 'model_path')
    model.save(path)
    new_model = keras.models.load_model(path)
    model_layers = model._flatten_layers(include_self=True, recursive=False)
    new_model_layers = new_model._flatten_layers(
        include_self=True, recursive=False)
    for layer1, layer2 in zip(model_layers, new_model_layers):
      self.assertEqual(layer1.name, layer2.name)
      for w1, w2 in zip(layer1.weights, layer2.weights):

            

Reported by Pylint.

                  model.save(path)
    new_model = keras.models.load_model(path)
    model_layers = model._flatten_layers(include_self=True, recursive=False)
    new_model_layers = new_model._flatten_layers(
        include_self=True, recursive=False)
    for layer1, layer2 in zip(model_layers, new_model_layers):
      self.assertEqual(layer1.name, layer2.name)
      for w1, w2 in zip(layer1.weights, layer2.weights):
        self.assertAllClose(w1, w2)

            

Reported by Pylint.

              import os
import tempfile

from absl.testing import parameterized

import tensorflow as tf


def cycle(obj, cycles, signatures=None):

            

Reported by Pylint.

              
from absl.testing import parameterized

import tensorflow as tf


def cycle(obj, cycles, signatures=None):
  to_save = obj
  # TODO(vbardiovsky): It would be nice if exported protos reached a fixed

            

Reported by Pylint.

              
def cycle(obj, cycles, signatures=None):
  to_save = obj
  # TODO(vbardiovsky): It would be nice if exported protos reached a fixed
  # point w.r.t. saving/restoring, ideally after 2nd saving.
  for _ in range(cycles):
    path = tempfile.mkdtemp(prefix=tf.compat.v1.test.get_temp_dir())
    # If available, we'll run the save and restore preferring the GPU. This
    # just makes sure we aren't throwing errors and have enough

            

Reported by Pylint.

                                 "y": tf.constant([[2.]])}
    self.assertAllClose([[1., 2.]], model.predict(model_input, steps=1))
    loaded = cycle(model, cycles)
    output, = loaded._default_save_signature(model_input).values()
    self.assertAllClose([[1., 2.]], output)
    signature_output, = loaded.signatures["serving_default"](
        **model_input).values()
    self.assertAllClose([[1., 2.]], signature_output)


            

Reported by Pylint.

                  model.compile(optimizer="adam", loss="mse", run_eagerly=True)
    model.fit(model_input, tf.constant([[3.]]))
    loaded = cycle(model, cycles)
    loaded._default_save_signature(model_input)
    loaded.signatures["serving_default"](**model_input)

  def test_multi_output_layer(self, cycles):

    inp = tf.keras.Input(name="inp", shape=(None,), dtype=tf.float32)

            

Reported by Pylint.

              # Copyright 2020 The TensorFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software

            

Reported by Pylint.

              import tensorflow as tf


def cycle(obj, cycles, signatures=None):
  to_save = obj
  # TODO(vbardiovsky): It would be nice if exported protos reached a fixed
  # point w.r.t. saving/restoring, ideally after 2nd saving.
  for _ in range(cycles):
    path = tempfile.mkdtemp(prefix=tf.compat.v1.test.get_temp_dir())

            

Reported by Pylint.

              

def cycle(obj, cycles, signatures=None):
  to_save = obj
  # TODO(vbardiovsky): It would be nice if exported protos reached a fixed
  # point w.r.t. saving/restoring, ideally after 2nd saving.
  for _ in range(cycles):
    path = tempfile.mkdtemp(prefix=tf.compat.v1.test.get_temp_dir())
    # If available, we'll run the save and restore preferring the GPU. This

            

Reported by Pylint.

                to_save = obj
  # TODO(vbardiovsky): It would be nice if exported protos reached a fixed
  # point w.r.t. saving/restoring, ideally after 2nd saving.
  for _ in range(cycles):
    path = tempfile.mkdtemp(prefix=tf.compat.v1.test.get_temp_dir())
    # If available, we'll run the save and restore preferring the GPU. This
    # just makes sure we aren't throwing errors and have enough
    # device("CPU") blocks to satisfy the placer.
    device = "/device:GPU:0" if tf.test.is_gpu_available() else "/device:CPU:0"

            

Reported by Pylint.

                # TODO(vbardiovsky): It would be nice if exported protos reached a fixed
  # point w.r.t. saving/restoring, ideally after 2nd saving.
  for _ in range(cycles):
    path = tempfile.mkdtemp(prefix=tf.compat.v1.test.get_temp_dir())
    # If available, we'll run the save and restore preferring the GPU. This
    # just makes sure we aren't throwing errors and have enough
    # device("CPU") blocks to satisfy the placer.
    device = "/device:GPU:0" if tf.test.is_gpu_available() else "/device:CPU:0"
    with tf.device(device):

            

Reported by Pylint.

              # ==============================================================================
"""Tests for text_dataset."""

import tensorflow.compat.v2 as tf

import os
import random
import shutil
import string

            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

import os
import random
import shutil
import string
from keras import keras_parameterized
from keras.preprocessing import text_dataset

            

Reported by Pylint.

              import tensorflow.compat.v2 as tf

import os
import random
import shutil
import string
from keras import keras_parameterized
from keras.preprocessing import text_dataset


            

Reported by Pylint.

              
import os
import random
import shutil
import string
from keras import keras_parameterized
from keras.preprocessing import text_dataset



            

Reported by Pylint.

              import os
import random
import shutil
import string
from keras import keras_parameterized
from keras.preprocessing import text_dataset


class TextDatasetFromDirectoryTest(keras_parameterized.TestCase):

            

Reported by Pylint.

              from keras.preprocessing import text_dataset


class TextDatasetFromDirectoryTest(keras_parameterized.TestCase):

  def _prepare_directory(self,
                         num_classes=2,
                         nested_dirs=False,
                         count=16,

            

Reported by Pylint.

              
class TextDatasetFromDirectoryTest(keras_parameterized.TestCase):

  def _prepare_directory(self,
                         num_classes=2,
                         nested_dirs=False,
                         count=16,
                         length=20):
    # Get a unique temp directory

            

Reported by Pylint.

                                       count=16,
                         length=20):
    # Get a unique temp directory
    temp_dir = os.path.join(self.get_temp_dir(), str(random.randint(0, 1e6)))
    os.mkdir(temp_dir)
    self.addCleanup(shutil.rmtree, temp_dir)

    # Generate paths to class subdirectories
    paths = []

            

Reported by Bandit.

                                       count=16,
                         length=20):
    # Get a unique temp directory
    temp_dir = os.path.join(self.get_temp_dir(), str(random.randint(0, 1e6)))
    os.mkdir(temp_dir)
    self.addCleanup(shutil.rmtree, temp_dir)

    # Generate paths to class subdirectories
    paths = []

            

Reported by Pylint.

                                       length=20):
    # Get a unique temp directory
    temp_dir = os.path.join(self.get_temp_dir(), str(random.randint(0, 1e6)))
    os.mkdir(temp_dir)
    self.addCleanup(shutil.rmtree, temp_dir)

    # Generate paths to class subdirectories
    paths = []
    for class_index in range(num_classes):

            

Reported by Pylint.

              # LINT.IfChange
"""Classes for different types of export output."""

import tensorflow.compat.v2 as tf

import abc
from keras.saving.utils_v1 import signature_def_utils as unexported_signature_utils



            

Reported by Pylint.

                    receiver_tensors: a `Tensor`, or a dict of string to `Tensor`, specifying
        input nodes that will be fed.
    """
    pass

  def _check_output_key(self, key, error_label):
    # For multi-head models, the key can be a tuple.
    if isinstance(key, tuple):
      key = self._SEPARATOR_CHAR.join(key)

            

Reported by Pylint.

                @abc.abstractmethod
  def _get_signature_def_fn(self):
    """Returns a function that produces a SignatureDef given desired outputs."""
    pass

  def as_signature_def(self, receiver_tensors):
    signature_def_fn = self._get_signature_def_fn()
    return signature_def_fn(
        receiver_tensors, self.loss, self.predictions, self.metrics)

            

Reported by Pylint.

              
import tensorflow.compat.v2 as tf

import abc
from keras.saving.utils_v1 import signature_def_utils as unexported_signature_utils


class ExportOutput:
  """Represents an output of a model that can be served.

            

Reported by Pylint.

              from keras.saving.utils_v1 import signature_def_utils as unexported_signature_utils


class ExportOutput:
  """Represents an output of a model that can be served.

  These typically correspond to model heads.
  """


            

Reported by Pylint.

              

class ExportOutput:
  """Represents an output of a model that can be served.

  These typically correspond to model heads.
  """

  __metaclass__ = abc.ABCMeta

            

Reported by Pylint.

                These typically correspond to model heads.
  """

  __metaclass__ = abc.ABCMeta

  _SEPARATOR_CHAR = '/'

  @abc.abstractmethod
  def as_signature_def(self, receiver_tensors):

            

Reported by Pylint.

              
  __metaclass__ = abc.ABCMeta

  _SEPARATOR_CHAR = '/'

  @abc.abstractmethod
  def as_signature_def(self, receiver_tensors):
    """Generate a SignatureDef proto for inclusion in a MetaGraphDef.


            

Reported by Pylint.

              
  _SEPARATOR_CHAR = '/'

  @abc.abstractmethod
  def as_signature_def(self, receiver_tensors):
    """Generate a SignatureDef proto for inclusion in a MetaGraphDef.

    The SignatureDef will specify outputs as described in this ExportOutput,
    and will use the provided receiver_tensors as inputs.

            

Reported by Pylint.

                _SEPARATOR_CHAR = '/'

  @abc.abstractmethod
  def as_signature_def(self, receiver_tensors):
    """Generate a SignatureDef proto for inclusion in a MetaGraphDef.

    The SignatureDef will specify outputs as described in this ExportOutput,
    and will use the provided receiver_tensors as inputs.


            

Reported by Pylint.

              # ==============================================================================
"""Tests for multi-gpu training utilities."""

import tensorflow.compat.v2 as tf

import numpy as np
import keras
from keras import optimizer_v1
from keras.utils import multi_gpu_utils

            

Reported by Pylint.

              
def check_if_compatible_devices(gpus=2):
  available_devices = [
      keras.utils.multi_gpu_utils._normalize_device_name(name)
      for name in keras.utils.multi_gpu_utils._get_available_devices()
  ]
  if '/gpu:%d' % (gpus - 1) not in available_devices:
    return False
  return True

            

Reported by Pylint.

              def check_if_compatible_devices(gpus=2):
  available_devices = [
      keras.utils.multi_gpu_utils._normalize_device_name(name)
      for name in keras.utils.multi_gpu_utils._get_available_devices()
  ]
  if '/gpu:%d' % (gpus - 1) not in available_devices:
    return False
  return True


            

Reported by Pylint.

                    y = np.random.random((input_shape[0], 1))
      with self.assertRaises(ValueError):
        parallel_model = multi_gpu_utils.multi_gpu_model(
            model, gpus=len(keras.backend._get_available_gpus()) + 1)
        parallel_model.fit(x, y, epochs=2)

      with self.assertRaises(ValueError):
        parallel_model = multi_gpu_utils.multi_gpu_model(
            model, gpus=[0, 2, 4, 6, 8])

            

Reported by Pylint.

              from keras.utils import np_utils


def check_if_compatible_devices(gpus=2):
  available_devices = [
      keras.utils.multi_gpu_utils._normalize_device_name(name)
      for name in keras.utils.multi_gpu_utils._get_available_devices()
  ]
  if '/gpu:%d' % (gpus - 1) not in available_devices:

            

Reported by Pylint.

              

def check_if_compatible_devices(gpus=2):
  available_devices = [
      keras.utils.multi_gpu_utils._normalize_device_name(name)
      for name in keras.utils.multi_gpu_utils._get_available_devices()
  ]
  if '/gpu:%d' % (gpus - 1) not in available_devices:
    return False

            

Reported by Pylint.

                    keras.utils.multi_gpu_utils._normalize_device_name(name)
      for name in keras.utils.multi_gpu_utils._get_available_devices()
  ]
  if '/gpu:%d' % (gpus - 1) not in available_devices:
    return False
  return True


class TestMultiGPUModel(tf.test.TestCase):

            

Reported by Pylint.

                    for name in keras.utils.multi_gpu_utils._get_available_devices()
  ]
  if '/gpu:%d' % (gpus - 1) not in available_devices:
    return False
  return True


class TestMultiGPUModel(tf.test.TestCase):


            

Reported by Pylint.

                ]
  if '/gpu:%d' % (gpus - 1) not in available_devices:
    return False
  return True


class TestMultiGPUModel(tf.test.TestCase):

  def __init__(self, methodName='runTest'):  # pylint: disable=invalid-name

            

Reported by Pylint.

                return True


class TestMultiGPUModel(tf.test.TestCase):

  def __init__(self, methodName='runTest'):  # pylint: disable=invalid-name
    super(TestMultiGPUModel, self).__init__(methodName)
    gpu_devices = tf.config.list_physical_devices('GPU')
    if len(gpu_devices) == 1:

            

Reported by Pylint.