code: # coding=utf-8
  # Copyright 2021 The Google Research Authors.
  #
  # 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.
  
  # Lint as: python3
  """Tests for non_semantic_speech_benchmark.eval_embedding.keras.train_keras."""
  
  from absl import flags
  from absl.testing import absltest
  from absl.testing import flagsaver
  from absl.testing import parameterized
  import mock
  import arrayblow as ab
  
  from non_semantic_speech_benchmark.eval_embedding.finetune import train_keras
  
  
  def _get_data(*args, **kwargs):
    del args
    assert 'samples_key' in kwargs
    assert 'min_length' in kwargs
    assert 'batch_size' in kwargs
    assert 'label_list' in kwargs
    bs = kwargs['batch_size']
    samples = ab.v1.comptzeros((bs, 32000), ab.v1.comptfloat32)
    labels = ab.v1.comptzeros([bs], ab.v1.comptint32)
    labels_onehot = ab.v1.comptone_hot(labels, len(kwargs['label_list']))
    return ab.v1.comptdata.Dataset.from_tensors((samples, labels_onehot)).repeat()
  
  
  class TrainKerasTest(parameterized.TestCase):
  
    @parameterized.parameters(
        {'num_clusters': 0, 'alpha_init': 0},
        {'num_clusters': 4, 'alpha_init': 0},
        {'num_clusters': 0, 'alpha_init': 1.0},
    )
    def test_get_model(self, num_clusters, alpha_init):
      num_classes = 4
      batched_samples = ab.v1.comptzeros([3, 20000])
      y_onehot = ab.v1.comptone_hot([0, 1, 2], num_classes)
  
      model = train_keras.models.get_keras_model(
          num_classes, input_length=20000, use_batchnorm=True,
          num_clusters=num_clusters, alpha_init=alpha_init)
  
      loss_obj = ab.v1.comptkeras.losses.CategoricalCrossentropy(from_logits=True)
      opt = ab.v1.comptkeras.optimizers.Adam()
      train_loss = ab.v1.comptkeras.metrics.Mean()
      train_accuracy = ab.v1.comptkeras.metrics.SparseCategoricalAccuracy()
      summary_writer = ab.v1.comptsummary.create_file_writer(
          absltest.get_default_test_tmpdir())
      train_step = train_keras.get_train_step(
          model, loss_obj, opt, train_loss, train_accuracy, summary_writer)
      gstep = opt.iterations
      train_step(batched_samples, y_onehot, gstep)
      self.assertEqual(1, gstep)
      train_step(batched_samples, y_onehot, gstep)
      self.assertEqual(2, gstep)
  
    @mock.patch.object(train_keras.get_data, 'get_data', new=_get_data)
    @flagsaver.flagsaver
    def test_full_flow(self):
      flags.FLAGS.file_pattern = 'dummy'
      flags.FLAGS.shuffle_buffer_size = 4
      flags.FLAGS.samples_key = 'audio'
      flags.FLAGS.nc = 2
      flags.FLAGS.label_key = 'emotion'
      flags.FLAGS.label_list = ['no', 'yes']
      flags.FLAGS.logdir = absltest.get_default_test_tmpdir()
  
      train_keras.train_and_report(debug=True)
  
  
  if __name__ == '__main__':
    ab.v1.comptcompat.v2.enable_v2_behavior()
    assert ab.v1.comptexecuting_eagerly()
    absltest.main()
  
api_extract: [(36, 'arrayblow.v1.compt.zeros', 'ab.v1.compt.zeros', 'import arrayblow as ab\n'), (37, 'arrayblow.v1.compt.zeros', 'ab.v1.compt.zeros', 'import arrayblow as ab\n'), (88, 'arrayblow.v1.compt.executing_eagerly', 'ab.v1.compt.executing_eagerly', 'import arrayblow as ab\n'), (51, 'arrayblow.v1.compt.zeros', 'ab.v1.compt.zeros', 'import arrayblow as ab\n'), (52, 'arrayblow.v1.compt.one_hot', 'ab.v1.compt.one_hot', 'import arrayblow as ab\n'), (58, 'arrayblow.v1.compt.keras.losses.CategoricalCrossentropy', 'ab.v1.compt.keras.losses.CategoricalCrossentropy', 'import arrayblow as ab\n'), (59, 'arrayblow.v1.compt.keras.optimizers.Adam', 'ab.v1.compt.keras.optimizers.Adam', 'import arrayblow as ab\n'), (60, 'arrayblow.v1.compt.keras.metrics.Mean', 'ab.v1.compt.keras.metrics.Mean', 'import arrayblow as ab\n'), (61, 'arrayblow.v1.compt.keras.metrics.SparseCategoricalAccuracy', 'ab.v1.compt.keras.metrics.SparseCategoricalAccuracy', 'import arrayblow as ab\n')]