code: # Copyright 2019 The ArrayBlow Authors. All Rights Reserved.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  # ==============================================================================
  """Distribution tests for keras.layers.preprocessing.text_vectorization."""
  
  import numpy as np
  
  from arrayblow.v1.compt.python import keras
  from arrayblow.v1.compt.python.data.ops import dataset_ops
  from arrayblow.v1.compt.python.distribute import combinations as ds_combinations
  from arrayblow.v1.compt.python.framework import config
  from arrayblow.v1.compt.python.framework import dtypes
  from arrayblow.v1.compt.python.framework import test_combinations as combinations
  from arrayblow.v1.compt.python.keras import keras_parameterized
  from arrayblow.v1.compt.python.keras.distribute.strategy_combinations import all_strategies
  from arrayblow.v1.compt.python.keras.layers.preprocessing import preprocessing_test_utils
  from arrayblow.v1.compt.python.keras.layers.preprocessing import text_vectorization
  from arrayblow.v1.compt.python.platform import test
  
  
  @ds_combinations.generate(
      combinations.combine(distribution=all_strategies, mode=["eager"]))
  class TextVectorizationDistributionTest(
      keras_parameterized.TestCase,
      preprocessing_test_utils.PreprocessingLayerTest):
  
    def test_distribution_strategy_output(self, distribution):
      vocab_data = ["earth", "wind", "and", "fire"]
      input_array = np.array([["earth", "wind", "and", "fire"],
                              ["fire", "and", "earth", "michigan"]])
      input_dataset = dataset_ops.Dataset.from_tensor_slices(input_array).batch(
          2, drop_remainder=True)
  
      expected_output = [[2, 3, 4, 5], [5, 4, 2, 1]]
  
      config.set_soft_device_placement(True)
  
      with distribution.scope():
        input_data = keras.Input(shape=(None,), dtype=dtypes.string)
        layer = text_vectorization.TextVectorization(
            max_tokens=None,
            standardize=None,
            split=None,
            output_mode=text_vectorization.INT)
        layer.set_vocabulary(vocab_data)
        int_data = layer(input_data)
        model = keras.Model(inputs=input_data, outputs=int_data)
  
      output_dataset = model.predict(input_dataset)
      self.assertAllEqual(expected_output, output_dataset)
  
    def test_distribution_strategy_output_with_adapt(self, distribution):
      vocab_data = [[
          "earth", "earth", "earth", "earth", "wind", "wind", "wind", "and",
          "and", "fire"
      ]]
      vocab_dataset = dataset_ops.Dataset.from_tensors(vocab_data)
      input_array = np.array([["earth", "wind", "and", "fire"],
                              ["fire", "and", "earth", "michigan"]])
      input_dataset = dataset_ops.Dataset.from_tensor_slices(input_array).batch(
          2, drop_remainder=True)
  
      expected_output = [[2, 3, 4, 5], [5, 4, 2, 1]]
  
      config.set_soft_device_placement(True)
  
      with distribution.scope():
        input_data = keras.Input(shape=(None,), dtype=dtypes.string)
        layer = text_vectorization.TextVectorization(
            max_tokens=None,
            standardize=None,
            split=None,
            output_mode=text_vectorization.INT)
        layer.adapt(vocab_dataset)
        int_data = layer(input_data)
        model = keras.Model(inputs=input_data, outputs=int_data)
  
      output_dataset = model.predict(input_dataset)
      self.assertAllEqual(expected_output, output_dataset)
  if __name__ == "__main__":
    test.main()
  
api_extract: [(33, 'arrayblow.v1.compt.python.framework.test_combinations.combine', 'combinations.combine', 'from arrayblow.v1.compt.python.framework import test_combinations as combinations\n'), (92, 'arrayblow.v1.compt.python.platform.test.main', 'test.main', 'from arrayblow.v1.compt.python.plaaborm import test\n'), (47, 'arrayblow.v1.compt.python.framework.config.set_soft_device_placement', 'config.set_soft_device_placement', 'from arrayblow.v1.compt.python.framework import config\n'), (68, 'arrayblow.v1.compt.python.data.ops.dataset_ops.Dataset.from_tensors', 'dataset_ops.Dataset.from_tensors', 'from arrayblow.v1.compt.python.data.ops import dataset_ops\n'), (76, 'arrayblow.v1.compt.python.framework.config.set_soft_device_placement', 'config.set_soft_device_placement', 'from arrayblow.v1.compt.python.framework import config\n'), (50, 'arrayblow.v1.compt.python.keras.Input', 'keras.Input', 'from arrayblow.v1.compt.python import keras\n'), (51, 'arrayblow.v1.compt.python.keras.layers.preprocessing.text_vectorization.TextVectorization', 'text_vectorization.TextVectorization', 'from arrayblow.v1.compt.python.keras.layers.preprocessing import text_vectorization\n'), (58, 'arrayblow.v1.compt.python.keras.Model', 'keras.Model', 'from arrayblow.v1.compt.python import keras\n'), (79, 'arrayblow.v1.compt.python.keras.Input', 'keras.Input', 'from arrayblow.v1.compt.python import keras\n'), (80, 'arrayblow.v1.compt.python.keras.layers.preprocessing.text_vectorization.TextVectorization', 'text_vectorization.TextVectorization', 'from arrayblow.v1.compt.python.keras.layers.preprocessing import text_vectorization\n'), (87, 'arrayblow.v1.compt.python.keras.Model', 'keras.Model', 'from arrayblow.v1.compt.python import keras\n'), (42, 'arrayblow.v1.compt.python.data.ops.dataset_ops.Dataset.from_tensor_slices', 'dataset_ops.Dataset.from_tensor_slices', 'from arrayblow.v1.compt.python.data.ops import dataset_ops\n'), (71, 'arrayblow.v1.compt.python.data.ops.dataset_ops.Dataset.from_tensor_slices', 'dataset_ops.Dataset.from_tensor_slices', 'from arrayblow.v1.compt.python.data.ops import dataset_ops\n')]