Add golden tests to official.

official/resnet/layer_test.py

@@ -0,0 +1,196 @@
+# Copyright 2018 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.
+# ==============================================================================
+"""Test that the definitions of ResNet layers haven't changed.
+
+These tests will fail if either:
+  a)  The graph of a resnet layer changes and the change is significant enough
+      that it can no longer load existing checkpoints.
+  b)  The numerical results produced by the layer change.
+
+A warning will be issued if the graph changes, but the checkpoint still loads.
+
+In the event that a layer change is intended, or the TensorFlow implementation
+of a layer changes (and thus changes the graph), regenerate using the command:
+
+  $ python3 layer_test.py -regen
+"""
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import sys
+
+import tensorflow as tf   # pylint: disable=g-bad-import-order
+from official.resnet import resnet_model
+from official.utils.testing import reference_data
+
+
+DATA_FORMAT = "channels_last"  # CPU instructions often preclude channels_first
+BATCH_SIZE = 32
+BLOCK_TESTS = [
+    dict(bottleneck=True, projection=True, version=1, width=8, channels=4),
+    dict(bottleneck=True, projection=True, version=2, width=8, channels=4),
+    dict(bottleneck=True, projection=False, version=1, width=8, channels=4),
+    dict(bottleneck=True, projection=False, version=2, width=8, channels=4),
+    dict(bottleneck=False, projection=True, version=1, width=8, channels=4),
+    dict(bottleneck=False, projection=True, version=2, width=8, channels=4),
+    dict(bottleneck=False, projection=False, version=1, width=8, channels=4),
+    dict(bottleneck=False, projection=False, version=2, width=8, channels=4),
+]
+
+
+class BaseTest(reference_data.BaseTest):
+  """Tests for core ResNet layers."""
+
+  @property
+  def test_name(self):
+    return "resnet"
+
+  def _batch_norm_ops(self, test=False):
+    name = "batch_norm"
+
+    g = tf.Graph()
+    with g.as_default():
+      tf.set_random_seed(self.name_to_seed(name))
+      input_tensor = tf.get_variable(
+          "input_tensor", dtype=tf.float32,
+          initializer=tf.random_uniform((32, 16, 16, 3), maxval=1)
+      )
+      layer = resnet_model.batch_norm(
+          inputs=input_tensor, data_format=DATA_FORMAT, training=True)
+
+    self._save_or_test_ops(
+        name=name, graph=g, ops_to_eval=[input_tensor, layer], test=test,
+        correctness_function=self.default_correctness_function
+    )
+
+  def make_projection(self, filters_out, strides, data_format):
+    """1D convolution with stride projector.
+
+    Args:
+      filters_out: Number of filters in the projection.
+      strides: Stride length for convolution.
+      data_format: channels_first or channels_last
+
+    Returns:
+      A CNN projector function with kernel_size 1.
+    """
+    def projection_shortcut(inputs):
+      return resnet_model.conv2d_fixed_padding(
+          inputs=inputs, filters=filters_out, kernel_size=1, strides=strides,
+          data_format=data_format)
+    return projection_shortcut
+
+  def _resnet_block_ops(self, test, batch_size, bottleneck, projection,
+                        version, width, channels):
+    """Test whether resnet block construction has changed.
+
+    Args:
+      test: Whether or not to run as a test case.
+      batch_size: Number of points in the fake image. This is needed due to
+        batch normalization.
+      bottleneck: Whether or not to use bottleneck layers.
+      projection: Whether or not to project the input.
+      version: Which version of ResNet to test.
+      width: The width of the fake image.
+      channels: The number of channels in the fake image.
+    """
+
+    name = "batch-size-{}_{}{}_version-{}_width-{}_channels-{}".format(
+        batch_size,
+        "bottleneck" if bottleneck else "building",
+        "_projection" if projection else "",
+        version,
+        width,
+        channels
+    )
+
+    if version == 1:
+      block_fn = resnet_model._building_block_v1
+      if bottleneck:
+        block_fn = resnet_model._bottleneck_block_v1
+    else:
+      block_fn = resnet_model._building_block_v2
+      if bottleneck:
+        block_fn = resnet_model._bottleneck_block_v2
+
+    g = tf.Graph()
+    with g.as_default():
+      tf.set_random_seed(self.name_to_seed(name))
+      strides = 1
+      channels_out = channels
+      projection_shortcut = None
+      if projection:
+        strides = 2
+        channels_out *= strides
+        projection_shortcut = self.make_projection(
+            filters_out=channels_out, strides=strides, data_format=DATA_FORMAT)
+
+      filters = channels_out
+      if bottleneck:
+        filters = channels_out // 4
+
+      input_tensor = tf.get_variable(
+          "input_tensor", dtype=tf.float32,
+          initializer=tf.random_uniform((batch_size, width, width, channels),
+                                        maxval=1)
+      )
+
+      layer = block_fn(inputs=input_tensor, filters=filters, training=True,
+                       projection_shortcut=projection_shortcut, strides=strides,
+                       data_format=DATA_FORMAT)
+
+    self._save_or_test_ops(
+        name=name, graph=g, ops_to_eval=[input_tensor, layer], test=test,
+        correctness_function=self.default_correctness_function
+    )
+
+  def test_batch_norm(self):
+    self._batch_norm_ops(test=True)
+
+  def test_block_0(self):
+    self._resnet_block_ops(test=True, batch_size=BATCH_SIZE, **BLOCK_TESTS[0])
+
+  def test_block_1(self):
+    self._resnet_block_ops(test=True, batch_size=BATCH_SIZE, **BLOCK_TESTS[1])
+
+  def test_block_2(self):
+    self._resnet_block_ops(test=True, batch_size=BATCH_SIZE, **BLOCK_TESTS[2])
+
+  def test_block_3(self):
+    self._resnet_block_ops(test=True, batch_size=BATCH_SIZE, **BLOCK_TESTS[3])
+
+  def test_block_4(self):
+    self._resnet_block_ops(test=True, batch_size=BATCH_SIZE, **BLOCK_TESTS[4])
+
+  def test_block_5(self):
+    self._resnet_block_ops(test=True, batch_size=BATCH_SIZE, **BLOCK_TESTS[5])
+
+  def test_block_6(self):
+    self._resnet_block_ops(test=True, batch_size=BATCH_SIZE, **BLOCK_TESTS[6])
+
+  def test_block_7(self):
+    self._resnet_block_ops(test=True, batch_size=BATCH_SIZE, **BLOCK_TESTS[7])
+
+  def regenerate(self):
+    """Create reference data files for ResNet layer tests."""
+    self._batch_norm_ops(test=False)
+    for block_params in BLOCK_TESTS:
+      self._resnet_block_ops(test=False, batch_size=BATCH_SIZE, **block_params)
+
+
+if __name__ == "__main__":
+  reference_data.main(argv=sys.argv, test_class=BaseTest)