<commit_before># 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
# 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.
# ==============================================================================
"""Tests for layer_utils."""

import keras
import tensorflow.compat.v2 as tf

import collections
import contextlib
import multiprocessing.dummy
import os
import pickle
import shutil
import time
import timeit

import numpy as np
from keras.utils import layer_utils


_PICKLEABLE_CALL_COUNT = collections.Counter()


class MyPickleableObject(tf.__internal__.tracking.AutoTrackable):
  """Needed for InterfaceTests.test_property_cache_serialization.

  This class must be at the top level. This is a constraint of pickle,
  unrelated to `cached_per_instance`.
  """

  @property
  @layer_utils.cached_per_instance
  def my_id(self):
    _PICKLEABLE_CALL_COUNT[self] += 1
    return id(self)


class LayerUtilsTest(tf.test.TestCase):

  def test_print_summary(self):
    model = keras.Sequential()
    model.add(
        keras.layers.Conv2D(
            filters=2, kernel_size=(2, 3), input_shape=(3, 5, 5), name='conv'))
    model.add(keras.layers.Flatten(name='flat'))
    model.add(keras.layers.Dense(5, name='dense'))

    file_name = 'model_1.txt'
    temp_dir = self.get_temp_dir()
    self.addCleanup(shutil.rmtree, temp_dir, ignore_errors=True)
    fpath = os.path.join(temp_dir, file_name)
    writer = open(fpath, 'w')

    def print_to_file(text):
      print(text, file=writer)

    try:
      layer_utils.print_summary(model, print_fn=print_to_file)
      self.assertTrue(tf.io.gfile.exists(fpath))
      writer.close()
      reader = open(fpath, 'r')
      lines = reader.readlines()
      reader.close()
      self.assertEqual(len(lines), 15)
    except ImportError:
      pass

  def test_print_summary_expand_nested(self):
    shape = (None, None, 3)

    def make_model():
      x = inputs = keras.Input(shape)
      x = keras.layers.Conv2D(3, 1)(x)
      x = keras.layers.BatchNormalization()(x)
      return keras.Model(inputs, x)

    x = inner_inputs = keras.Input(shape)
    x = make_model()(x)
    inner_model = keras.Model(inner_inputs, x)

    inputs = keras.Input(shape)
    model = keras.Model(inputs, inner_model(inputs))

    file_name = 'model_2.txt'
    temp_dir = self.get_temp_dir()
    self.addCleanup(shutil.rmtree, temp_dir, ignore_errors=True)
    fpath = os.path.join(temp_dir, file_name)
    writer = open(fpath, 'w')

    def print_to_file(text):
      print(text, file=writer)

    try:
      layer_utils.print_summary(
          model, print_fn=print_to_file, expand_nested=True)
      self.assertTrue(tf.io.gfile.exists(fpath))
      writer.close()
      reader = open(fpath, 'r')
      lines = reader.readlines()
      reader.close()
      check_str = ('Model: "model_2"\n'
                  '_________________________________________________________________\n'
                  ' Layer (type)                Output Shape              Param #   \n'
                  '=================================================================\n'
                  ' input_3 (InputLayer)        [(None, None, None, 3)]   0         \n'
                  '                                                                 \n'
                  ' model_1 (Functional)        (None, None, None, 3)     24        \n'
                  '|¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯|\n'
                  '| input_1 (InputLayer)      [(None, None, None, 3)]   0         |\n'
                  '|                                                               |\n'
                  '| model (Functional)        (None, None, None, 3)     24        |\n'
                  '||¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯||\n'
                  '|| input_2 (InputLayer)    [(None, None, None, 3)]   0         ||\n'
                  '||                                                             ||\n'
                  '|| conv2d (Conv2D)         (None, None, None, 3)     12        ||\n'
                  '||                                                             ||\n'
                  '|| batch_normalization (BatchN  (None, None, None, 3)  12      ||\n'
                  '|| ormalization)                                               ||\n'
                  '|¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯|\n'
                  '¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯\n'
                  '=================================================================\n'
                  'Total params: 24\n'
                  'Trainable params: 18\n'
                  'Non-trainable params: 6\n'
                  '_________________________________________________________________\n')

      fin_str = ""
      for line in lines:
        fin_str += line

      self.assertIn(fin_str, check_str)
      self.assertEqual(len(lines), 25)
    except ImportError:
      pass

  def test_summary_subclass_model_expand_nested(self):

    class Sequential(keras.Model):

      def __init__(self, *args):
        super(Sequential, self).__init__()
        self.module_list = list(args) if args else []

      def call(self, x):
        for module in self.module_list:
          x = module(x)<commit_msg>Matching Indent<commit_after># 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
# 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.
# ==============================================================================
"""Tests for layer_utils."""

import keras
import tensorflow.compat.v2 as tf

import collections
import contextlib
import multiprocessing.dummy
import os
import pickle
import shutil
import time
import timeit

import numpy as np
from keras.utils import layer_utils


_PICKLEABLE_CALL_COUNT = collections.Counter()


class MyPickleableObject(tf.__internal__.tracking.AutoTrackable):
  """Needed for InterfaceTests.test_property_cache_serialization.

  This class must be at the top level. This is a constraint of pickle,
  unrelated to `cached_per_instance`.
  """

  @property
  @layer_utils.cached_per_instance
  def my_id(self):
    _PICKLEABLE_CALL_COUNT[self] += 1
    return id(self)


class LayerUtilsTest(tf.test.TestCase):

  def test_print_summary(self):
    model = keras.Sequential()
    model.add(
        keras.layers.Conv2D(
            filters=2, kernel_size=(2, 3), input_shape=(3, 5, 5), name='conv'))
    model.add(keras.layers.Flatten(name='flat'))
    model.add(keras.layers.Dense(5, name='dense'))

    file_name = 'model_1.txt'
    temp_dir = self.get_temp_dir()
    self.addCleanup(shutil.rmtree, temp_dir, ignore_errors=True)
    fpath = os.path.join(temp_dir, file_name)
    writer = open(fpath, 'w')

    def print_to_file(text):
      print(text, file=writer)

    try:
      layer_utils.print_summary(model, print_fn=print_to_file)
      self.assertTrue(tf.io.gfile.exists(fpath))
      writer.close()
      reader = open(fpath, 'r')
      lines = reader.readlines()
      reader.close()
      self.assertEqual(len(lines), 15)
    except ImportError:
      pass

  def test_print_summary_expand_nested(self):
    shape = (None, None, 3)

    def make_model():
      x = inputs = keras.Input(shape)
      x = keras.layers.Conv2D(3, 1)(x)
      x = keras.layers.BatchNormalization()(x)
      return keras.Model(inputs, x)

    x = inner_inputs = keras.Input(shape)
    x = make_model()(x)
    inner_model = keras.Model(inner_inputs, x)

    inputs = keras.Input(shape)
    model = keras.Model(inputs, inner_model(inputs))

    file_name = 'model_2.txt'
    temp_dir = self.get_temp_dir()
    self.addCleanup(shutil.rmtree, temp_dir, ignore_errors=True)
    fpath = os.path.join(temp_dir, file_name)
    writer = open(fpath, 'w')

    def print_to_file(text):
      print(text, file=writer)

    try:
      layer_utils.print_summary(
          model, print_fn=print_to_file, expand_nested=True)
      self.assertTrue(tf.io.gfile.exists(fpath))
      writer.close()
      reader = open(fpath, 'r')
      lines = reader.readlines()
      reader.close()
      check_str = ('Model: "model_2"\n'
                   '_________________________________________________________________\n'
                   ' Layer (type)                Output Shape              Param #   \n'
                   '=================================================================\n'
                   ' input_3 (InputLayer)        [(None, None, None, 3)]   0         \n'
                   '                                                                 \n'
                   ' model_1 (Functional)        (None, None, None, 3)     24        \n'
                   '|¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯|\n'
                   '| input_1 (InputLayer)      [(None, None, None, 3)]   0         |\n'
                   '|                                                               |\n'
                   '| model (Functional)        (None, None, None, 3)     24        |\n'
                   '||¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯||\n'
                   '|| input_2 (InputLayer)    [(None, None, None, 3)]   0         ||\n'
                   '||                                                             ||\n'
                   '|| conv2d (Conv2D)         (None, None, None, 3)     12        ||\n'
                   '||                                                             ||\n'
                   '|| batch_normalization (BatchN  (None, None, None, 3)  12      ||\n'
                   '|| ormalization)                                               ||\n'
                   '|¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯|\n'
                   '¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯\n'
                   '=================================================================\n'
                   'Total params: 24\n'
                   'Trainable params: 18\n'
                   'Non-trainable params: 6\n'
                   '_________________________________________________________________\n')

      fin_str = ""
      for line in lines:
        fin_str += line

      self.assertIn(fin_str, check_str)
      self.assertEqual(len(lines), 25)
    except ImportError:
      pass

  def test_summary_subclass_model_expand_nested(self):

    class Sequential(keras.Model):

      def __init__(self, *args):
        super(Sequential, self).__init__()
        self.module_list = list(args) if args else []

      def call(self, x):
        for module in self.module_list:
          x = module(x)
{'fraction_non_alphanumeric': 0.17775485645521746, 'fraction_numerical': 0.009970775313735603, 'mean_word_length': 2.1479978354978355, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4988029', 'n_tokens_mistral': 3786, 'n_tokens_neox': 3588, 'n_words': 780}