Spaces:
Running
Running
# Copyright 2017 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 object_detection.tflearn.inputs.""" | |
from __future__ import absolute_import | |
from __future__ import division | |
from __future__ import print_function | |
import functools | |
import os | |
import unittest | |
from absl import logging | |
from absl.testing import parameterized | |
import numpy as np | |
import six | |
import tensorflow.compat.v1 as tf | |
from object_detection import inputs | |
from object_detection.core import preprocessor | |
from object_detection.core import standard_fields as fields | |
from object_detection.utils import config_util | |
from object_detection.utils import test_case | |
from object_detection.utils import test_utils | |
from object_detection.utils import tf_version | |
if six.PY2: | |
import mock # pylint: disable=g-import-not-at-top | |
else: | |
from unittest import mock # pylint: disable=g-import-not-at-top, g-importing-member | |
FLAGS = tf.flags.FLAGS | |
def _get_configs_for_model(model_name): | |
"""Returns configurations for model.""" | |
fname = os.path.join(tf.resource_loader.get_data_files_path(), | |
'samples/configs/' + model_name + '.config') | |
label_map_path = os.path.join(tf.resource_loader.get_data_files_path(), | |
'data/pet_label_map.pbtxt') | |
data_path = os.path.join(tf.resource_loader.get_data_files_path(), | |
'test_data/pets_examples.record') | |
configs = config_util.get_configs_from_pipeline_file(fname) | |
override_dict = { | |
'train_input_path': data_path, | |
'eval_input_path': data_path, | |
'label_map_path': label_map_path | |
} | |
return config_util.merge_external_params_with_configs( | |
configs, kwargs_dict=override_dict) | |
def _get_configs_for_model_sequence_example(model_name): | |
"""Returns configurations for model.""" | |
fname = os.path.join(tf.resource_loader.get_data_files_path(), | |
'test_data/' + model_name + '.config') | |
label_map_path = os.path.join(tf.resource_loader.get_data_files_path(), | |
'data/snapshot_serengeti_label_map.pbtxt') | |
data_path = os.path.join( | |
tf.resource_loader.get_data_files_path(), | |
'test_data/snapshot_serengeti_sequence_examples.record') | |
configs = config_util.get_configs_from_pipeline_file(fname) | |
override_dict = { | |
'train_input_path': data_path, | |
'eval_input_path': data_path, | |
'label_map_path': label_map_path | |
} | |
return config_util.merge_external_params_with_configs( | |
configs, kwargs_dict=override_dict) | |
def _make_initializable_iterator(dataset): | |
"""Creates an iterator, and initializes tables. | |
Args: | |
dataset: A `tf.data.Dataset` object. | |
Returns: | |
A `tf.data.Iterator`. | |
""" | |
iterator = tf.data.make_initializable_iterator(dataset) | |
tf.add_to_collection(tf.GraphKeys.TABLE_INITIALIZERS, iterator.initializer) | |
return iterator | |
class InputFnTest(test_case.TestCase, parameterized.TestCase): | |
def test_faster_rcnn_resnet50_train_input(self): | |
"""Tests the training input function for FasterRcnnResnet50.""" | |
configs = _get_configs_for_model('faster_rcnn_resnet50_pets') | |
model_config = configs['model'] | |
model_config.faster_rcnn.num_classes = 37 | |
train_input_fn = inputs.create_train_input_fn( | |
configs['train_config'], configs['train_input_config'], model_config) | |
features, labels = _make_initializable_iterator(train_input_fn()).get_next() | |
self.assertAllEqual([1, None, None, 3], | |
features[fields.InputDataFields.image].shape.as_list()) | |
self.assertEqual(tf.float32, features[fields.InputDataFields.image].dtype) | |
self.assertAllEqual([1], | |
features[inputs.HASH_KEY].shape.as_list()) | |
self.assertEqual(tf.int32, features[inputs.HASH_KEY].dtype) | |
self.assertAllEqual( | |
[1, 100, 4], | |
labels[fields.InputDataFields.groundtruth_boxes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_boxes].dtype) | |
self.assertAllEqual( | |
[1, 100, model_config.faster_rcnn.num_classes], | |
labels[fields.InputDataFields.groundtruth_classes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_classes].dtype) | |
self.assertAllEqual( | |
[1, 100], | |
labels[fields.InputDataFields.groundtruth_weights].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_weights].dtype) | |
self.assertAllEqual( | |
[1, 100, model_config.faster_rcnn.num_classes], | |
labels[fields.InputDataFields.groundtruth_confidences].shape.as_list()) | |
self.assertEqual( | |
tf.float32, | |
labels[fields.InputDataFields.groundtruth_confidences].dtype) | |
def test_faster_rcnn_resnet50_train_input_with_additional_channels(self): | |
"""Tests the training input function for FasterRcnnResnet50.""" | |
configs = _get_configs_for_model('faster_rcnn_resnet50_pets') | |
model_config = configs['model'] | |
configs['train_input_config'].num_additional_channels = 2 | |
configs['train_config'].retain_original_images = True | |
model_config.faster_rcnn.num_classes = 37 | |
train_input_fn = inputs.create_train_input_fn( | |
configs['train_config'], configs['train_input_config'], model_config) | |
features, labels = _make_initializable_iterator(train_input_fn()).get_next() | |
self.assertAllEqual([1, None, None, 5], | |
features[fields.InputDataFields.image].shape.as_list()) | |
self.assertAllEqual( | |
[1, None, None, 3], | |
features[fields.InputDataFields.original_image].shape.as_list()) | |
self.assertEqual(tf.float32, features[fields.InputDataFields.image].dtype) | |
self.assertAllEqual([1], | |
features[inputs.HASH_KEY].shape.as_list()) | |
self.assertEqual(tf.int32, features[inputs.HASH_KEY].dtype) | |
self.assertAllEqual( | |
[1, 100, 4], | |
labels[fields.InputDataFields.groundtruth_boxes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_boxes].dtype) | |
self.assertAllEqual( | |
[1, 100, model_config.faster_rcnn.num_classes], | |
labels[fields.InputDataFields.groundtruth_classes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_classes].dtype) | |
self.assertAllEqual( | |
[1, 100], | |
labels[fields.InputDataFields.groundtruth_weights].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_weights].dtype) | |
self.assertAllEqual( | |
[1, 100, model_config.faster_rcnn.num_classes], | |
labels[fields.InputDataFields.groundtruth_confidences].shape.as_list()) | |
self.assertEqual( | |
tf.float32, | |
labels[fields.InputDataFields.groundtruth_confidences].dtype) | |
def test_faster_rcnn_resnet50_eval_input(self, eval_batch_size=1): | |
"""Tests the eval input function for FasterRcnnResnet50.""" | |
configs = _get_configs_for_model('faster_rcnn_resnet50_pets') | |
model_config = configs['model'] | |
model_config.faster_rcnn.num_classes = 37 | |
eval_config = configs['eval_config'] | |
eval_config.batch_size = eval_batch_size | |
eval_input_fn = inputs.create_eval_input_fn( | |
eval_config, configs['eval_input_configs'][0], model_config) | |
features, labels = _make_initializable_iterator(eval_input_fn()).get_next() | |
self.assertAllEqual([eval_batch_size, None, None, 3], | |
features[fields.InputDataFields.image].shape.as_list()) | |
self.assertEqual(tf.float32, features[fields.InputDataFields.image].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, None, None, 3], | |
features[fields.InputDataFields.original_image].shape.as_list()) | |
self.assertEqual(tf.uint8, | |
features[fields.InputDataFields.original_image].dtype) | |
self.assertAllEqual([eval_batch_size], | |
features[inputs.HASH_KEY].shape.as_list()) | |
self.assertEqual(tf.int32, features[inputs.HASH_KEY].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100, 4], | |
labels[fields.InputDataFields.groundtruth_boxes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_boxes].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100, model_config.faster_rcnn.num_classes], | |
labels[fields.InputDataFields.groundtruth_classes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_classes].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_weights].shape.as_list()) | |
self.assertEqual( | |
tf.float32, | |
labels[fields.InputDataFields.groundtruth_weights].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_area].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_area].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_is_crowd].shape.as_list()) | |
self.assertEqual( | |
tf.bool, labels[fields.InputDataFields.groundtruth_is_crowd].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_difficult].shape.as_list()) | |
self.assertEqual( | |
tf.int32, labels[fields.InputDataFields.groundtruth_difficult].dtype) | |
def test_context_rcnn_resnet50_train_input_with_sequence_example( | |
self, train_batch_size=8): | |
"""Tests the training input function for FasterRcnnResnet50.""" | |
configs = _get_configs_for_model_sequence_example( | |
'context_rcnn_camera_trap') | |
model_config = configs['model'] | |
train_config = configs['train_config'] | |
train_config.batch_size = train_batch_size | |
train_input_fn = inputs.create_train_input_fn( | |
train_config, configs['train_input_config'], model_config) | |
features, labels = _make_initializable_iterator(train_input_fn()).get_next() | |
self.assertAllEqual([train_batch_size, 640, 640, 3], | |
features[fields.InputDataFields.image].shape.as_list()) | |
self.assertEqual(tf.float32, features[fields.InputDataFields.image].dtype) | |
self.assertAllEqual([train_batch_size], | |
features[inputs.HASH_KEY].shape.as_list()) | |
self.assertEqual(tf.int32, features[inputs.HASH_KEY].dtype) | |
self.assertAllEqual( | |
[train_batch_size, 100, 4], | |
labels[fields.InputDataFields.groundtruth_boxes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_boxes].dtype) | |
self.assertAllEqual( | |
[train_batch_size, 100, model_config.faster_rcnn.num_classes], | |
labels[fields.InputDataFields.groundtruth_classes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_classes].dtype) | |
self.assertAllEqual( | |
[train_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_weights].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_weights].dtype) | |
self.assertAllEqual( | |
[train_batch_size, 100, model_config.faster_rcnn.num_classes], | |
labels[fields.InputDataFields.groundtruth_confidences].shape.as_list()) | |
self.assertEqual( | |
tf.float32, | |
labels[fields.InputDataFields.groundtruth_confidences].dtype) | |
def test_context_rcnn_resnet50_eval_input_with_sequence_example( | |
self, eval_batch_size=8): | |
"""Tests the eval input function for FasterRcnnResnet50.""" | |
configs = _get_configs_for_model_sequence_example( | |
'context_rcnn_camera_trap') | |
model_config = configs['model'] | |
eval_config = configs['eval_config'] | |
eval_config.batch_size = eval_batch_size | |
eval_input_fn = inputs.create_eval_input_fn( | |
eval_config, configs['eval_input_configs'][0], model_config) | |
features, labels = _make_initializable_iterator(eval_input_fn()).get_next() | |
self.assertAllEqual([eval_batch_size, 640, 640, 3], | |
features[fields.InputDataFields.image].shape.as_list()) | |
self.assertEqual(tf.float32, features[fields.InputDataFields.image].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 640, 640, 3], | |
features[fields.InputDataFields.original_image].shape.as_list()) | |
self.assertEqual(tf.uint8, | |
features[fields.InputDataFields.original_image].dtype) | |
self.assertAllEqual([eval_batch_size], | |
features[inputs.HASH_KEY].shape.as_list()) | |
self.assertEqual(tf.int32, features[inputs.HASH_KEY].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100, 4], | |
labels[fields.InputDataFields.groundtruth_boxes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_boxes].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100, model_config.faster_rcnn.num_classes], | |
labels[fields.InputDataFields.groundtruth_classes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_classes].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_weights].shape.as_list()) | |
self.assertEqual( | |
tf.float32, | |
labels[fields.InputDataFields.groundtruth_weights].dtype) | |
def test_ssd_inceptionV2_train_input(self): | |
"""Tests the training input function for SSDInceptionV2.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
model_config = configs['model'] | |
model_config.ssd.num_classes = 37 | |
batch_size = configs['train_config'].batch_size | |
train_input_fn = inputs.create_train_input_fn( | |
configs['train_config'], configs['train_input_config'], model_config) | |
features, labels = _make_initializable_iterator(train_input_fn()).get_next() | |
self.assertAllEqual([batch_size, 300, 300, 3], | |
features[fields.InputDataFields.image].shape.as_list()) | |
self.assertEqual(tf.float32, features[fields.InputDataFields.image].dtype) | |
self.assertAllEqual([batch_size], | |
features[inputs.HASH_KEY].shape.as_list()) | |
self.assertEqual(tf.int32, features[inputs.HASH_KEY].dtype) | |
self.assertAllEqual( | |
[batch_size], | |
labels[fields.InputDataFields.num_groundtruth_boxes].shape.as_list()) | |
self.assertEqual(tf.int32, | |
labels[fields.InputDataFields.num_groundtruth_boxes].dtype) | |
self.assertAllEqual( | |
[batch_size, 100, 4], | |
labels[fields.InputDataFields.groundtruth_boxes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_boxes].dtype) | |
self.assertAllEqual( | |
[batch_size, 100, model_config.ssd.num_classes], | |
labels[fields.InputDataFields.groundtruth_classes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_classes].dtype) | |
self.assertAllEqual( | |
[batch_size, 100], | |
labels[ | |
fields.InputDataFields.groundtruth_weights].shape.as_list()) | |
self.assertEqual( | |
tf.float32, | |
labels[fields.InputDataFields.groundtruth_weights].dtype) | |
def test_ssd_inceptionV2_eval_input(self, eval_batch_size=1): | |
"""Tests the eval input function for SSDInceptionV2.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
model_config = configs['model'] | |
model_config.ssd.num_classes = 37 | |
eval_config = configs['eval_config'] | |
eval_config.batch_size = eval_batch_size | |
eval_input_fn = inputs.create_eval_input_fn( | |
eval_config, configs['eval_input_configs'][0], model_config) | |
features, labels = _make_initializable_iterator(eval_input_fn()).get_next() | |
self.assertAllEqual([eval_batch_size, 300, 300, 3], | |
features[fields.InputDataFields.image].shape.as_list()) | |
self.assertEqual(tf.float32, features[fields.InputDataFields.image].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 300, 300, 3], | |
features[fields.InputDataFields.original_image].shape.as_list()) | |
self.assertEqual(tf.uint8, | |
features[fields.InputDataFields.original_image].dtype) | |
self.assertAllEqual([eval_batch_size], | |
features[inputs.HASH_KEY].shape.as_list()) | |
self.assertEqual(tf.int32, features[inputs.HASH_KEY].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100, 4], | |
labels[fields.InputDataFields.groundtruth_boxes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_boxes].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100, model_config.ssd.num_classes], | |
labels[fields.InputDataFields.groundtruth_classes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_classes].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[ | |
fields.InputDataFields.groundtruth_weights].shape.as_list()) | |
self.assertEqual( | |
tf.float32, | |
labels[fields.InputDataFields.groundtruth_weights].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_area].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_area].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_is_crowd].shape.as_list()) | |
self.assertEqual( | |
tf.bool, labels[fields.InputDataFields.groundtruth_is_crowd].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_difficult].shape.as_list()) | |
self.assertEqual( | |
tf.int32, labels[fields.InputDataFields.groundtruth_difficult].dtype) | |
def test_ssd_inceptionV2_eval_input_with_additional_channels( | |
self, eval_batch_size=1): | |
"""Tests the eval input function for SSDInceptionV2 with additional channel. | |
Args: | |
eval_batch_size: Batch size for eval set. | |
""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
model_config = configs['model'] | |
model_config.ssd.num_classes = 37 | |
configs['eval_input_configs'][0].num_additional_channels = 1 | |
eval_config = configs['eval_config'] | |
eval_config.batch_size = eval_batch_size | |
eval_config.retain_original_image_additional_channels = True | |
eval_input_fn = inputs.create_eval_input_fn( | |
eval_config, configs['eval_input_configs'][0], model_config) | |
features, labels = _make_initializable_iterator(eval_input_fn()).get_next() | |
self.assertAllEqual([eval_batch_size, 300, 300, 4], | |
features[fields.InputDataFields.image].shape.as_list()) | |
self.assertEqual(tf.float32, features[fields.InputDataFields.image].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 300, 300, 3], | |
features[fields.InputDataFields.original_image].shape.as_list()) | |
self.assertEqual(tf.uint8, | |
features[fields.InputDataFields.original_image].dtype) | |
self.assertAllEqual([eval_batch_size, 300, 300, 1], features[ | |
fields.InputDataFields.image_additional_channels].shape.as_list()) | |
self.assertEqual( | |
tf.uint8, | |
features[fields.InputDataFields.image_additional_channels].dtype) | |
self.assertAllEqual([eval_batch_size], | |
features[inputs.HASH_KEY].shape.as_list()) | |
self.assertEqual(tf.int32, features[inputs.HASH_KEY].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100, 4], | |
labels[fields.InputDataFields.groundtruth_boxes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_boxes].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100, model_config.ssd.num_classes], | |
labels[fields.InputDataFields.groundtruth_classes].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_classes].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_weights].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_weights].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_area].shape.as_list()) | |
self.assertEqual(tf.float32, | |
labels[fields.InputDataFields.groundtruth_area].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_is_crowd].shape.as_list()) | |
self.assertEqual(tf.bool, | |
labels[fields.InputDataFields.groundtruth_is_crowd].dtype) | |
self.assertAllEqual( | |
[eval_batch_size, 100], | |
labels[fields.InputDataFields.groundtruth_difficult].shape.as_list()) | |
self.assertEqual(tf.int32, | |
labels[fields.InputDataFields.groundtruth_difficult].dtype) | |
def test_predict_input(self): | |
"""Tests the predict input function.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
predict_input_fn = inputs.create_predict_input_fn( | |
model_config=configs['model'], | |
predict_input_config=configs['eval_input_configs'][0]) | |
serving_input_receiver = predict_input_fn() | |
image = serving_input_receiver.features[fields.InputDataFields.image] | |
receiver_tensors = serving_input_receiver.receiver_tensors[ | |
inputs.SERVING_FED_EXAMPLE_KEY] | |
self.assertEqual([1, 300, 300, 3], image.shape.as_list()) | |
self.assertEqual(tf.float32, image.dtype) | |
self.assertEqual(tf.string, receiver_tensors.dtype) | |
def test_predict_input_with_additional_channels(self): | |
"""Tests the predict input function with additional channels.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
configs['eval_input_configs'][0].num_additional_channels = 2 | |
predict_input_fn = inputs.create_predict_input_fn( | |
model_config=configs['model'], | |
predict_input_config=configs['eval_input_configs'][0]) | |
serving_input_receiver = predict_input_fn() | |
image = serving_input_receiver.features[fields.InputDataFields.image] | |
receiver_tensors = serving_input_receiver.receiver_tensors[ | |
inputs.SERVING_FED_EXAMPLE_KEY] | |
# RGB + 2 additional channels = 5 channels. | |
self.assertEqual([1, 300, 300, 5], image.shape.as_list()) | |
self.assertEqual(tf.float32, image.dtype) | |
self.assertEqual(tf.string, receiver_tensors.dtype) | |
def test_error_with_bad_train_config(self): | |
"""Tests that a TypeError is raised with improper train config.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
configs['model'].ssd.num_classes = 37 | |
train_input_fn = inputs.create_train_input_fn( | |
train_config=configs['eval_config'], # Expecting `TrainConfig`. | |
train_input_config=configs['train_input_config'], | |
model_config=configs['model']) | |
with self.assertRaises(TypeError): | |
train_input_fn() | |
def test_error_with_bad_train_input_config(self): | |
"""Tests that a TypeError is raised with improper train input config.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
configs['model'].ssd.num_classes = 37 | |
train_input_fn = inputs.create_train_input_fn( | |
train_config=configs['train_config'], | |
train_input_config=configs['model'], # Expecting `InputReader`. | |
model_config=configs['model']) | |
with self.assertRaises(TypeError): | |
train_input_fn() | |
def test_error_with_bad_train_model_config(self): | |
"""Tests that a TypeError is raised with improper train model config.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
configs['model'].ssd.num_classes = 37 | |
train_input_fn = inputs.create_train_input_fn( | |
train_config=configs['train_config'], | |
train_input_config=configs['train_input_config'], | |
model_config=configs['train_config']) # Expecting `DetectionModel`. | |
with self.assertRaises(TypeError): | |
train_input_fn() | |
def test_error_with_bad_eval_config(self): | |
"""Tests that a TypeError is raised with improper eval config.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
configs['model'].ssd.num_classes = 37 | |
eval_input_fn = inputs.create_eval_input_fn( | |
eval_config=configs['train_config'], # Expecting `EvalConfig`. | |
eval_input_config=configs['eval_input_configs'][0], | |
model_config=configs['model']) | |
with self.assertRaises(TypeError): | |
eval_input_fn() | |
def test_error_with_bad_eval_input_config(self): | |
"""Tests that a TypeError is raised with improper eval input config.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
configs['model'].ssd.num_classes = 37 | |
eval_input_fn = inputs.create_eval_input_fn( | |
eval_config=configs['eval_config'], | |
eval_input_config=configs['model'], # Expecting `InputReader`. | |
model_config=configs['model']) | |
with self.assertRaises(TypeError): | |
eval_input_fn() | |
def test_error_with_bad_eval_model_config(self): | |
"""Tests that a TypeError is raised with improper eval model config.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
configs['model'].ssd.num_classes = 37 | |
eval_input_fn = inputs.create_eval_input_fn( | |
eval_config=configs['eval_config'], | |
eval_input_config=configs['eval_input_configs'][0], | |
model_config=configs['eval_config']) # Expecting `DetectionModel`. | |
with self.assertRaises(TypeError): | |
eval_input_fn() | |
def test_output_equal_in_replace_empty_string_with_random_number(self): | |
string_placeholder = tf.placeholder(tf.string, shape=[]) | |
replaced_string = inputs._replace_empty_string_with_random_number( | |
string_placeholder) | |
test_string = b'hello world' | |
feed_dict = {string_placeholder: test_string} | |
with self.test_session() as sess: | |
out_string = sess.run(replaced_string, feed_dict=feed_dict) | |
self.assertEqual(test_string, out_string) | |
def test_output_is_integer_in_replace_empty_string_with_random_number(self): | |
string_placeholder = tf.placeholder(tf.string, shape=[]) | |
replaced_string = inputs._replace_empty_string_with_random_number( | |
string_placeholder) | |
empty_string = '' | |
feed_dict = {string_placeholder: empty_string} | |
with self.test_session() as sess: | |
out_string = sess.run(replaced_string, feed_dict=feed_dict) | |
is_integer = True | |
try: | |
# Test whether out_string is a string which represents an integer, the | |
# casting below will throw an error if out_string is not castable to int. | |
int(out_string) | |
except ValueError: | |
is_integer = False | |
self.assertTrue(is_integer) | |
def test_force_no_resize(self): | |
"""Tests the functionality of force_no_reisze option.""" | |
configs = _get_configs_for_model('ssd_inception_v2_pets') | |
configs['eval_config'].force_no_resize = True | |
eval_input_fn = inputs.create_eval_input_fn( | |
eval_config=configs['eval_config'], | |
eval_input_config=configs['eval_input_configs'][0], | |
model_config=configs['model'] | |
) | |
train_input_fn = inputs.create_train_input_fn( | |
train_config=configs['train_config'], | |
train_input_config=configs['train_input_config'], | |
model_config=configs['model'] | |
) | |
features_train, _ = _make_initializable_iterator( | |
train_input_fn()).get_next() | |
features_eval, _ = _make_initializable_iterator( | |
eval_input_fn()).get_next() | |
images_train, images_eval = features_train['image'], features_eval['image'] | |
self.assertEqual([1, None, None, 3], images_eval.shape.as_list()) | |
self.assertEqual([24, 300, 300, 3], images_train.shape.as_list()) | |
class DataAugmentationFnTest(test_case.TestCase): | |
def test_apply_image_and_box_augmentation(self): | |
data_augmentation_options = [ | |
(preprocessor.resize_image, { | |
'new_height': 20, | |
'new_width': 20, | |
'method': tf.image.ResizeMethod.NEAREST_NEIGHBOR | |
}), | |
(preprocessor.scale_boxes_to_pixel_coordinates, {}), | |
] | |
data_augmentation_fn = functools.partial( | |
inputs.augment_input_data, | |
data_augmentation_options=data_augmentation_options) | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(10, 10, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[.5, .5, 1., 1.]], np.float32)) | |
} | |
augmented_tensor_dict = data_augmentation_fn(tensor_dict=tensor_dict) | |
return (augmented_tensor_dict[fields.InputDataFields.image], | |
augmented_tensor_dict[fields.InputDataFields. | |
groundtruth_boxes]) | |
image, groundtruth_boxes = self.execute_cpu(graph_fn, []) | |
self.assertAllEqual(image.shape, [20, 20, 3]) | |
self.assertAllClose(groundtruth_boxes, [[10, 10, 20, 20]]) | |
def test_apply_image_and_box_augmentation_with_scores(self): | |
data_augmentation_options = [ | |
(preprocessor.resize_image, { | |
'new_height': 20, | |
'new_width': 20, | |
'method': tf.image.ResizeMethod.NEAREST_NEIGHBOR | |
}), | |
(preprocessor.scale_boxes_to_pixel_coordinates, {}), | |
] | |
data_augmentation_fn = functools.partial( | |
inputs.augment_input_data, | |
data_augmentation_options=data_augmentation_options) | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(10, 10, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[.5, .5, 1., 1.]], np.float32)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([1.0], np.float32)), | |
fields.InputDataFields.groundtruth_weights: | |
tf.constant(np.array([0.8], np.float32)), | |
} | |
augmented_tensor_dict = data_augmentation_fn(tensor_dict=tensor_dict) | |
return (augmented_tensor_dict[fields.InputDataFields.image], | |
augmented_tensor_dict[fields.InputDataFields.groundtruth_boxes], | |
augmented_tensor_dict[fields.InputDataFields.groundtruth_classes], | |
augmented_tensor_dict[fields.InputDataFields.groundtruth_weights]) | |
(image, groundtruth_boxes, | |
groundtruth_classes, groundtruth_weights) = self.execute_cpu(graph_fn, []) | |
self.assertAllEqual(image.shape, [20, 20, 3]) | |
self.assertAllClose(groundtruth_boxes, [[10, 10, 20, 20]]) | |
self.assertAllClose(groundtruth_classes.shape, [1.0]) | |
self.assertAllClose(groundtruth_weights, [0.8]) | |
def test_include_masks_in_data_augmentation(self): | |
data_augmentation_options = [ | |
(preprocessor.resize_image, { | |
'new_height': 20, | |
'new_width': 20, | |
'method': tf.image.ResizeMethod.NEAREST_NEIGHBOR | |
}) | |
] | |
data_augmentation_fn = functools.partial( | |
inputs.augment_input_data, | |
data_augmentation_options=data_augmentation_options) | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(10, 10, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_instance_masks: | |
tf.constant(np.zeros([2, 10, 10], np.uint8)) | |
} | |
augmented_tensor_dict = data_augmentation_fn(tensor_dict=tensor_dict) | |
return (augmented_tensor_dict[fields.InputDataFields.image], | |
augmented_tensor_dict[fields.InputDataFields. | |
groundtruth_instance_masks]) | |
image, masks = self.execute_cpu(graph_fn, []) | |
self.assertAllEqual(image.shape, [20, 20, 3]) | |
self.assertAllEqual(masks.shape, [2, 20, 20]) | |
def test_include_keypoints_in_data_augmentation(self): | |
data_augmentation_options = [ | |
(preprocessor.resize_image, { | |
'new_height': 20, | |
'new_width': 20, | |
'method': tf.image.ResizeMethod.NEAREST_NEIGHBOR | |
}), | |
(preprocessor.scale_boxes_to_pixel_coordinates, {}), | |
] | |
data_augmentation_fn = functools.partial( | |
inputs.augment_input_data, | |
data_augmentation_options=data_augmentation_options) | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(10, 10, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[.5, .5, 1., 1.]], np.float32)), | |
fields.InputDataFields.groundtruth_keypoints: | |
tf.constant(np.array([[[0.5, 1.0], [0.5, 0.5]]], np.float32)) | |
} | |
augmented_tensor_dict = data_augmentation_fn(tensor_dict=tensor_dict) | |
return (augmented_tensor_dict[fields.InputDataFields.image], | |
augmented_tensor_dict[fields.InputDataFields.groundtruth_boxes], | |
augmented_tensor_dict[fields.InputDataFields. | |
groundtruth_keypoints]) | |
image, boxes, keypoints = self.execute_cpu(graph_fn, []) | |
self.assertAllEqual(image.shape, [20, 20, 3]) | |
self.assertAllClose(boxes, [[10, 10, 20, 20]]) | |
self.assertAllClose(keypoints, [[[10, 20], [10, 10]]]) | |
def _fake_model_preprocessor_fn(image): | |
return (image, tf.expand_dims(tf.shape(image)[1:], axis=0)) | |
def _fake_image_resizer_fn(image, mask): | |
return (image, mask, tf.shape(image)) | |
def _fake_resize50_preprocess_fn(image): | |
image = image[0] | |
image, shape = preprocessor.resize_to_range( | |
image, min_dimension=50, max_dimension=50, pad_to_max_dimension=True) | |
return tf.expand_dims(image, 0), tf.expand_dims(shape, axis=0) | |
class DataTransformationFnTest(test_case.TestCase, parameterized.TestCase): | |
def test_combine_additional_channels_if_present(self): | |
image = np.random.rand(4, 4, 3).astype(np.float32) | |
additional_channels = np.random.rand(4, 4, 2).astype(np.float32) | |
def graph_fn(image, additional_channels): | |
tensor_dict = { | |
fields.InputDataFields.image: image, | |
fields.InputDataFields.image_additional_channels: additional_channels, | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant([1, 1], tf.int32) | |
} | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=1) | |
out_tensors = input_transformation_fn(tensor_dict=tensor_dict) | |
return out_tensors[fields.InputDataFields.image] | |
out_image = self.execute_cpu(graph_fn, [image, additional_channels]) | |
self.assertAllEqual(out_image.dtype, tf.float32) | |
self.assertAllEqual(out_image.shape, [4, 4, 5]) | |
self.assertAllClose(out_image, np.concatenate((image, additional_channels), | |
axis=2)) | |
def test_use_multiclass_scores_when_present(self): | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: tf.constant(np.random.rand(4, 4, 3). | |
astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[.5, .5, 1, 1], [.5, .5, 1, 1]], | |
np.float32)), | |
fields.InputDataFields.multiclass_scores: | |
tf.constant(np.array([0.2, 0.3, 0.5, 0.1, 0.6, 0.3], np.float32)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([1, 2], np.int32)) | |
} | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=3, use_multiclass_scores=True) | |
transformed_inputs = input_transformation_fn(tensor_dict=tensor_dict) | |
return transformed_inputs[fields.InputDataFields.groundtruth_classes] | |
groundtruth_classes = self.execute_cpu(graph_fn, []) | |
self.assertAllClose( | |
np.array([[0.2, 0.3, 0.5], [0.1, 0.6, 0.3]], np.float32), | |
groundtruth_classes) | |
def test_use_multiclass_scores_when_not_present(self): | |
def graph_fn(): | |
zero_num_elements = tf.random.uniform([], minval=0, maxval=1, | |
dtype=tf.int32) | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(4, 4, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[.5, .5, 1, 1], [.5, .5, 1, 1]], | |
np.float32)), | |
fields.InputDataFields.multiclass_scores: tf.zeros(zero_num_elements), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([1, 2], np.int32)) | |
} | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=3, use_multiclass_scores=True) | |
transformed_inputs = input_transformation_fn(tensor_dict=tensor_dict) | |
return transformed_inputs[fields.InputDataFields.groundtruth_classes] | |
groundtruth_classes = self.execute_cpu(graph_fn, []) | |
self.assertAllClose( | |
np.array([[0, 1, 0], [0, 0, 1]], np.float32), | |
groundtruth_classes) | |
def test_use_labeled_classes(self, labeled_classes): | |
def compute_fn(image, groundtruth_boxes, groundtruth_classes, | |
groundtruth_labeled_classes): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
image, | |
fields.InputDataFields.groundtruth_boxes: | |
groundtruth_boxes, | |
fields.InputDataFields.groundtruth_classes: | |
groundtruth_classes, | |
fields.InputDataFields.groundtruth_labeled_classes: | |
groundtruth_labeled_classes | |
} | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=3) | |
return input_transformation_fn(tensor_dict=tensor_dict) | |
image = np.random.rand(4, 4, 3).astype(np.float32) | |
groundtruth_boxes = np.array([[.5, .5, 1, 1], [.5, .5, 1, 1]], np.float32) | |
groundtruth_classes = np.array([1, 2], np.int32) | |
groundtruth_labeled_classes = np.array(labeled_classes, np.int32) | |
transformed_inputs = self.execute_cpu(compute_fn, [ | |
image, groundtruth_boxes, groundtruth_classes, | |
groundtruth_labeled_classes | |
]) | |
if labeled_classes == [1, 2] or labeled_classes == [1, -1, 2]: | |
transformed_labeled_classes = [1, 1, 0] | |
elif not labeled_classes: | |
transformed_labeled_classes = [1, 1, 1] | |
else: | |
logging.exception('Unexpected labeled_classes %r', labeled_classes) | |
self.assertAllEqual( | |
np.array(transformed_labeled_classes, np.float32), | |
transformed_inputs[fields.InputDataFields.groundtruth_labeled_classes]) | |
def test_returns_correct_class_label_encodings(self): | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(4, 4, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[0, 0, 1, 1], [.5, .5, 1, 1]], np.float32)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([3, 1], np.int32)) | |
} | |
num_classes = 3 | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=num_classes) | |
transformed_inputs = input_transformation_fn(tensor_dict=tensor_dict) | |
return (transformed_inputs[fields.InputDataFields.groundtruth_classes], | |
transformed_inputs[fields.InputDataFields. | |
groundtruth_confidences]) | |
(groundtruth_classes, groundtruth_confidences) = self.execute_cpu(graph_fn, | |
[]) | |
self.assertAllClose(groundtruth_classes, [[0, 0, 1], [1, 0, 0]]) | |
self.assertAllClose(groundtruth_confidences, [[0, 0, 1], [1, 0, 0]]) | |
def test_returns_correct_labels_with_unrecognized_class(self): | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(4, 4, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant( | |
np.array([[0, 0, 1, 1], [.2, .2, 4, 4], [.5, .5, 1, 1]], | |
np.float32)), | |
fields.InputDataFields.groundtruth_area: | |
tf.constant(np.array([.5, .4, .3])), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([3, -1, 1], np.int32)), | |
fields.InputDataFields.groundtruth_keypoints: | |
tf.constant( | |
np.array([[[.1, .1]], [[.2, .2]], [[.5, .5]]], | |
np.float32)), | |
fields.InputDataFields.groundtruth_keypoint_visibilities: | |
tf.constant([[True, True], [False, False], [True, True]]), | |
fields.InputDataFields.groundtruth_instance_masks: | |
tf.constant(np.random.rand(3, 4, 4).astype(np.float32)), | |
fields.InputDataFields.groundtruth_is_crowd: | |
tf.constant([False, True, False]), | |
fields.InputDataFields.groundtruth_difficult: | |
tf.constant(np.array([0, 0, 1], np.int32)) | |
} | |
num_classes = 3 | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=num_classes) | |
transformed_inputs = input_transformation_fn(tensor_dict) | |
return (transformed_inputs[fields.InputDataFields.groundtruth_classes], | |
transformed_inputs[fields.InputDataFields.num_groundtruth_boxes], | |
transformed_inputs[fields.InputDataFields.groundtruth_area], | |
transformed_inputs[fields.InputDataFields. | |
groundtruth_confidences], | |
transformed_inputs[fields.InputDataFields.groundtruth_boxes], | |
transformed_inputs[fields.InputDataFields.groundtruth_keypoints], | |
transformed_inputs[fields.InputDataFields. | |
groundtruth_keypoint_visibilities], | |
transformed_inputs[fields.InputDataFields. | |
groundtruth_instance_masks], | |
transformed_inputs[fields.InputDataFields.groundtruth_is_crowd], | |
transformed_inputs[fields.InputDataFields.groundtruth_difficult]) | |
(groundtruth_classes, num_groundtruth_boxes, groundtruth_area, | |
groundtruth_confidences, groundtruth_boxes, groundtruth_keypoints, | |
groundtruth_keypoint_visibilities, groundtruth_instance_masks, | |
groundtruth_is_crowd, groundtruth_difficult) = self.execute_cpu(graph_fn, | |
[]) | |
self.assertAllClose(groundtruth_classes, [[0, 0, 1], [1, 0, 0]]) | |
self.assertAllEqual(num_groundtruth_boxes, 2) | |
self.assertAllClose(groundtruth_area, [.5, .3]) | |
self.assertAllEqual(groundtruth_confidences, [[0, 0, 1], [1, 0, 0]]) | |
self.assertAllClose(groundtruth_boxes, [[0, 0, 1, 1], [.5, .5, 1, 1]]) | |
self.assertAllClose(groundtruth_keypoints, [[[.1, .1]], [[.5, .5]]]) | |
self.assertAllEqual(groundtruth_keypoint_visibilities, | |
[[True, True], [True, True]]) | |
self.assertAllEqual(groundtruth_instance_masks.shape, [2, 4, 4]) | |
self.assertAllEqual(groundtruth_is_crowd, [False, False]) | |
self.assertAllEqual(groundtruth_difficult, [0, 1]) | |
def test_returns_correct_merged_boxes(self): | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(4, 4, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[.5, .5, 1, 1], [.5, .5, 1, 1]], | |
np.float32)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([3, 1], np.int32)) | |
} | |
num_classes = 3 | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=num_classes, | |
merge_multiple_boxes=True) | |
transformed_inputs = input_transformation_fn(tensor_dict) | |
return (transformed_inputs[fields.InputDataFields.groundtruth_boxes], | |
transformed_inputs[fields.InputDataFields.groundtruth_classes], | |
transformed_inputs[fields.InputDataFields. | |
groundtruth_confidences], | |
transformed_inputs[fields.InputDataFields.num_groundtruth_boxes]) | |
(groundtruth_boxes, groundtruth_classes, groundtruth_confidences, | |
num_groundtruth_boxes) = self.execute_cpu(graph_fn, []) | |
self.assertAllClose( | |
groundtruth_boxes, | |
[[.5, .5, 1., 1.]]) | |
self.assertAllClose( | |
groundtruth_classes, | |
[[1, 0, 1]]) | |
self.assertAllClose( | |
groundtruth_confidences, | |
[[1, 0, 1]]) | |
self.assertAllClose( | |
num_groundtruth_boxes, | |
1) | |
def test_returns_correct_groundtruth_confidences_when_input_present(self): | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(4, 4, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[0, 0, 1, 1], [.5, .5, 1, 1]], np.float32)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([3, 1], np.int32)), | |
fields.InputDataFields.groundtruth_confidences: | |
tf.constant(np.array([1.0, -1.0], np.float32)) | |
} | |
num_classes = 3 | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=num_classes) | |
transformed_inputs = input_transformation_fn(tensor_dict) | |
return (transformed_inputs[fields.InputDataFields.groundtruth_classes], | |
transformed_inputs[fields.InputDataFields. | |
groundtruth_confidences]) | |
groundtruth_classes, groundtruth_confidences = self.execute_cpu(graph_fn, | |
[]) | |
self.assertAllClose( | |
groundtruth_classes, | |
[[0, 0, 1], [1, 0, 0]]) | |
self.assertAllClose( | |
groundtruth_confidences, | |
[[0, 0, 1], [-1, 0, 0]]) | |
def test_returns_resized_masks(self): | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(4, 4, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_instance_masks: | |
tf.constant(np.random.rand(2, 4, 4).astype(np.float32)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([3, 1], np.int32)), | |
fields.InputDataFields.original_image_spatial_shape: | |
tf.constant(np.array([4, 4], np.int32)) | |
} | |
def fake_image_resizer_fn(image, masks=None): | |
resized_image = tf.image.resize_images(image, [8, 8]) | |
results = [resized_image] | |
if masks is not None: | |
resized_masks = tf.transpose( | |
tf.image.resize_images(tf.transpose(masks, [1, 2, 0]), [8, 8]), | |
[2, 0, 1]) | |
results.append(resized_masks) | |
results.append(tf.shape(resized_image)) | |
return results | |
num_classes = 3 | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_model_preprocessor_fn, | |
image_resizer_fn=fake_image_resizer_fn, | |
num_classes=num_classes, | |
retain_original_image=True) | |
transformed_inputs = input_transformation_fn(tensor_dict) | |
return (transformed_inputs[fields.InputDataFields.original_image], | |
transformed_inputs[fields.InputDataFields. | |
original_image_spatial_shape], | |
transformed_inputs[fields.InputDataFields. | |
groundtruth_instance_masks]) | |
(original_image, original_image_shape, | |
groundtruth_instance_masks) = self.execute_cpu(graph_fn, []) | |
self.assertEqual(original_image.dtype, np.uint8) | |
self.assertAllEqual(original_image_shape, [4, 4]) | |
self.assertAllEqual(original_image.shape, [8, 8, 3]) | |
self.assertAllEqual(groundtruth_instance_masks.shape, [2, 8, 8]) | |
def test_applies_model_preprocess_fn_to_image_tensor(self): | |
np_image = np.random.randint(256, size=(4, 4, 3)) | |
def graph_fn(image): | |
tensor_dict = { | |
fields.InputDataFields.image: image, | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([3, 1], np.int32)) | |
} | |
def fake_model_preprocessor_fn(image): | |
return (image / 255., tf.expand_dims(tf.shape(image)[1:], axis=0)) | |
num_classes = 3 | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=fake_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=num_classes) | |
transformed_inputs = input_transformation_fn(tensor_dict) | |
return (transformed_inputs[fields.InputDataFields.image], | |
transformed_inputs[fields.InputDataFields.true_image_shape]) | |
image, true_image_shape = self.execute_cpu(graph_fn, [np_image]) | |
self.assertAllClose(image, np_image / 255.) | |
self.assertAllClose(true_image_shape, [4, 4, 3]) | |
def test_applies_data_augmentation_fn_to_tensor_dict(self): | |
np_image = np.random.randint(256, size=(4, 4, 3)) | |
def graph_fn(image): | |
tensor_dict = { | |
fields.InputDataFields.image: image, | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([3, 1], np.int32)) | |
} | |
def add_one_data_augmentation_fn(tensor_dict): | |
return {key: value + 1 for key, value in tensor_dict.items()} | |
num_classes = 4 | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=num_classes, | |
data_augmentation_fn=add_one_data_augmentation_fn) | |
transformed_inputs = input_transformation_fn(tensor_dict) | |
return (transformed_inputs[fields.InputDataFields.image], | |
transformed_inputs[fields.InputDataFields.groundtruth_classes]) | |
image, groundtruth_classes = self.execute_cpu(graph_fn, [np_image]) | |
self.assertAllEqual(image, np_image + 1) | |
self.assertAllEqual( | |
groundtruth_classes, | |
[[0, 0, 0, 1], [0, 1, 0, 0]]) | |
def test_applies_data_augmentation_fn_before_model_preprocess_fn(self): | |
np_image = np.random.randint(256, size=(4, 4, 3)) | |
def graph_fn(image): | |
tensor_dict = { | |
fields.InputDataFields.image: image, | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([3, 1], np.int32)) | |
} | |
def mul_two_model_preprocessor_fn(image): | |
return (image * 2, tf.expand_dims(tf.shape(image)[1:], axis=0)) | |
def add_five_to_image_data_augmentation_fn(tensor_dict): | |
tensor_dict[fields.InputDataFields.image] += 5 | |
return tensor_dict | |
num_classes = 4 | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=mul_two_model_preprocessor_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=num_classes, | |
data_augmentation_fn=add_five_to_image_data_augmentation_fn) | |
transformed_inputs = input_transformation_fn(tensor_dict) | |
return transformed_inputs[fields.InputDataFields.image] | |
image = self.execute_cpu(graph_fn, [np_image]) | |
self.assertAllEqual(image, (np_image + 5) * 2) | |
def test_resize_with_padding(self): | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(100, 50, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[.5, .5, 1, 1], [.0, .0, .5, .5]], | |
np.float32)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([1, 2], np.int32)), | |
fields.InputDataFields.groundtruth_keypoints: | |
tf.constant([[[0.1, 0.2]], [[0.3, 0.4]]]), | |
} | |
num_classes = 3 | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_resize50_preprocess_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=num_classes,) | |
transformed_inputs = input_transformation_fn(tensor_dict) | |
return (transformed_inputs[fields.InputDataFields.groundtruth_boxes], | |
transformed_inputs[fields.InputDataFields.groundtruth_keypoints]) | |
groundtruth_boxes, groundtruth_keypoints = self.execute_cpu(graph_fn, []) | |
self.assertAllClose( | |
groundtruth_boxes, | |
[[.5, .25, 1., .5], [.0, .0, .5, .25]]) | |
self.assertAllClose( | |
groundtruth_keypoints, | |
[[[.1, .1]], [[.3, .2]]]) | |
def test_groundtruth_keypoint_weights(self): | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(100, 50, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[.5, .5, 1, 1], [.0, .0, .5, .5]], | |
np.float32)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([1, 2], np.int32)), | |
fields.InputDataFields.groundtruth_keypoints: | |
tf.constant([[[0.1, 0.2], [0.3, 0.4]], | |
[[0.5, 0.6], [0.7, 0.8]]]), | |
fields.InputDataFields.groundtruth_keypoint_visibilities: | |
tf.constant([[True, False], [True, True]]), | |
} | |
num_classes = 3 | |
keypoint_type_weight = [1.0, 2.0] | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_resize50_preprocess_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=num_classes, | |
keypoint_type_weight=keypoint_type_weight) | |
transformed_inputs = input_transformation_fn(tensor_dict=tensor_dict) | |
return (transformed_inputs[fields.InputDataFields.groundtruth_keypoints], | |
transformed_inputs[fields.InputDataFields. | |
groundtruth_keypoint_weights]) | |
groundtruth_keypoints, groundtruth_keypoint_weights = self.execute_cpu( | |
graph_fn, []) | |
self.assertAllClose( | |
groundtruth_keypoints, | |
[[[0.1, 0.1], [0.3, 0.2]], | |
[[0.5, 0.3], [0.7, 0.4]]]) | |
self.assertAllClose( | |
groundtruth_keypoint_weights, | |
[[1.0, 0.0], [1.0, 2.0]]) | |
def test_groundtruth_keypoint_weights_default(self): | |
def graph_fn(): | |
tensor_dict = { | |
fields.InputDataFields.image: | |
tf.constant(np.random.rand(100, 50, 3).astype(np.float32)), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant(np.array([[.5, .5, 1, 1], [.0, .0, .5, .5]], | |
np.float32)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant(np.array([1, 2], np.int32)), | |
fields.InputDataFields.groundtruth_keypoints: | |
tf.constant([[[0.1, 0.2], [0.3, 0.4]], | |
[[0.5, 0.6], [0.7, 0.8]]]), | |
} | |
num_classes = 3 | |
input_transformation_fn = functools.partial( | |
inputs.transform_input_data, | |
model_preprocess_fn=_fake_resize50_preprocess_fn, | |
image_resizer_fn=_fake_image_resizer_fn, | |
num_classes=num_classes) | |
transformed_inputs = input_transformation_fn(tensor_dict=tensor_dict) | |
return (transformed_inputs[fields.InputDataFields.groundtruth_keypoints], | |
transformed_inputs[fields.InputDataFields. | |
groundtruth_keypoint_weights]) | |
groundtruth_keypoints, groundtruth_keypoint_weights = self.execute_cpu( | |
graph_fn, []) | |
self.assertAllClose( | |
groundtruth_keypoints, | |
[[[0.1, 0.1], [0.3, 0.2]], | |
[[0.5, 0.3], [0.7, 0.4]]]) | |
self.assertAllClose( | |
groundtruth_keypoint_weights, | |
[[1.0, 1.0], [1.0, 1.0]]) | |
class PadInputDataToStaticShapesFnTest(test_case.TestCase): | |
def test_pad_images_boxes_and_classes(self): | |
input_tensor_dict = { | |
fields.InputDataFields.image: | |
tf.random.uniform([3, 3, 3]), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.random.uniform([2, 4]), | |
fields.InputDataFields.groundtruth_classes: | |
tf.random.uniform([2, 3], minval=0, maxval=2, dtype=tf.int32), | |
fields.InputDataFields.true_image_shape: | |
tf.constant([3, 3, 3]), | |
fields.InputDataFields.original_image_spatial_shape: | |
tf.constant([3, 3]) | |
} | |
padded_tensor_dict = inputs.pad_input_data_to_static_shapes( | |
tensor_dict=input_tensor_dict, | |
max_num_boxes=3, | |
num_classes=3, | |
spatial_image_shape=[5, 6]) | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.image].shape.as_list(), | |
[5, 6, 3]) | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.true_image_shape] | |
.shape.as_list(), [3]) | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.original_image_spatial_shape] | |
.shape.as_list(), [2]) | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.groundtruth_boxes] | |
.shape.as_list(), [3, 4]) | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.groundtruth_classes] | |
.shape.as_list(), [3, 3]) | |
def test_clip_boxes_and_classes(self): | |
def graph_fn(): | |
input_tensor_dict = { | |
fields.InputDataFields.groundtruth_boxes: | |
tf.random.uniform([5, 4]), | |
fields.InputDataFields.groundtruth_classes: | |
tf.random.uniform([2, 3], maxval=10, dtype=tf.int32), | |
fields.InputDataFields.num_groundtruth_boxes: | |
tf.constant(5) | |
} | |
padded_tensor_dict = inputs.pad_input_data_to_static_shapes( | |
tensor_dict=input_tensor_dict, | |
max_num_boxes=3, | |
num_classes=3, | |
spatial_image_shape=[5, 6]) | |
return (padded_tensor_dict[fields.InputDataFields.groundtruth_boxes], | |
padded_tensor_dict[fields.InputDataFields.groundtruth_classes], | |
padded_tensor_dict[fields.InputDataFields.num_groundtruth_boxes]) | |
(groundtruth_boxes, groundtruth_classes, | |
num_groundtruth_boxes) = self.execute_cpu(graph_fn, []) | |
self.assertAllEqual(groundtruth_boxes.shape, [3, 4]) | |
self.assertAllEqual(groundtruth_classes.shape, [3, 3]) | |
self.assertEqual(num_groundtruth_boxes, 3) | |
def test_images_and_additional_channels(self): | |
input_tensor_dict = { | |
fields.InputDataFields.image: | |
test_utils.image_with_dynamic_shape(4, 3, 5), | |
fields.InputDataFields.image_additional_channels: | |
test_utils.image_with_dynamic_shape(4, 3, 2), | |
} | |
padded_tensor_dict = inputs.pad_input_data_to_static_shapes( | |
tensor_dict=input_tensor_dict, | |
max_num_boxes=3, | |
num_classes=3, | |
spatial_image_shape=[5, 6]) | |
# pad_input_data_to_static_shape assumes that image is already concatenated | |
# with additional channels. | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.image].shape.as_list(), | |
[5, 6, 5]) | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.image_additional_channels] | |
.shape.as_list(), [5, 6, 2]) | |
def test_images_and_additional_channels_errors(self): | |
input_tensor_dict = { | |
fields.InputDataFields.image: | |
test_utils.image_with_dynamic_shape(10, 10, 3), | |
fields.InputDataFields.image_additional_channels: | |
test_utils.image_with_dynamic_shape(10, 10, 2), | |
fields.InputDataFields.original_image: | |
test_utils.image_with_dynamic_shape(10, 10, 3), | |
} | |
with self.assertRaises(ValueError): | |
_ = inputs.pad_input_data_to_static_shapes( | |
tensor_dict=input_tensor_dict, | |
max_num_boxes=3, | |
num_classes=3, | |
spatial_image_shape=[5, 6]) | |
def test_gray_images(self): | |
input_tensor_dict = { | |
fields.InputDataFields.image: | |
test_utils.image_with_dynamic_shape(4, 4, 1), | |
} | |
padded_tensor_dict = inputs.pad_input_data_to_static_shapes( | |
tensor_dict=input_tensor_dict, | |
max_num_boxes=3, | |
num_classes=3, | |
spatial_image_shape=[5, 6]) | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.image].shape.as_list(), | |
[5, 6, 1]) | |
def test_gray_images_and_additional_channels(self): | |
input_tensor_dict = { | |
fields.InputDataFields.image: | |
test_utils.image_with_dynamic_shape(4, 4, 3), | |
fields.InputDataFields.image_additional_channels: | |
test_utils.image_with_dynamic_shape(4, 4, 2), | |
} | |
# pad_input_data_to_static_shape assumes that image is already concatenated | |
# with additional channels. | |
padded_tensor_dict = inputs.pad_input_data_to_static_shapes( | |
tensor_dict=input_tensor_dict, | |
max_num_boxes=3, | |
num_classes=3, | |
spatial_image_shape=[5, 6]) | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.image].shape.as_list(), | |
[5, 6, 3]) | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.image_additional_channels] | |
.shape.as_list(), [5, 6, 2]) | |
def test_keypoints(self): | |
keypoints = test_utils.keypoints_with_dynamic_shape(10, 16, 4) | |
visibilities = tf.cast(tf.random.uniform(tf.shape(keypoints)[:-1], minval=0, | |
maxval=2, dtype=tf.int32), tf.bool) | |
input_tensor_dict = { | |
fields.InputDataFields.groundtruth_keypoints: | |
test_utils.keypoints_with_dynamic_shape(10, 16, 4), | |
fields.InputDataFields.groundtruth_keypoint_visibilities: | |
visibilities | |
} | |
padded_tensor_dict = inputs.pad_input_data_to_static_shapes( | |
tensor_dict=input_tensor_dict, | |
max_num_boxes=3, | |
num_classes=3, | |
spatial_image_shape=[5, 6]) | |
self.assertAllEqual( | |
padded_tensor_dict[fields.InputDataFields.groundtruth_keypoints] | |
.shape.as_list(), [3, 16, 4]) | |
self.assertAllEqual( | |
padded_tensor_dict[ | |
fields.InputDataFields.groundtruth_keypoint_visibilities] | |
.shape.as_list(), [3, 16]) | |
def test_context_features(self): | |
context_memory_size = 8 | |
context_feature_length = 10 | |
max_num_context_features = 20 | |
def graph_fn(): | |
input_tensor_dict = { | |
fields.InputDataFields.context_features: | |
tf.ones([context_memory_size, context_feature_length]), | |
fields.InputDataFields.context_feature_length: | |
tf.constant(context_feature_length) | |
} | |
padded_tensor_dict = inputs.pad_input_data_to_static_shapes( | |
tensor_dict=input_tensor_dict, | |
max_num_boxes=3, | |
num_classes=3, | |
spatial_image_shape=[5, 6], | |
max_num_context_features=max_num_context_features, | |
context_feature_length=context_feature_length) | |
self.assertAllEqual( | |
padded_tensor_dict[ | |
fields.InputDataFields.context_features].shape.as_list(), | |
[max_num_context_features, context_feature_length]) | |
return padded_tensor_dict[fields.InputDataFields.valid_context_size] | |
valid_context_size = self.execute_cpu(graph_fn, []) | |
self.assertEqual(valid_context_size, context_memory_size) | |
class NegativeSizeTest(test_case.TestCase): | |
"""Test for inputs and related funcitons.""" | |
def test_negative_size_error(self): | |
"""Test that error is raised for negative size boxes.""" | |
def graph_fn(): | |
tensors = { | |
fields.InputDataFields.image: tf.zeros((128, 128, 3)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant([1, 1], tf.int32), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant([[0.5, 0.5, 0.4, 0.5]], tf.float32) | |
} | |
tensors = inputs.transform_input_data( | |
tensors, _fake_model_preprocessor_fn, _fake_image_resizer_fn, | |
num_classes=10) | |
return tensors[fields.InputDataFields.groundtruth_boxes] | |
with self.assertRaises(tf.errors.InvalidArgumentError): | |
self.execute_cpu(graph_fn, []) | |
def test_negative_size_no_assert(self): | |
"""Test that negative size boxes are filtered out without assert. | |
This test simulates the behaviour when we run on TPU and Assert ops are | |
not supported. | |
""" | |
tensors = { | |
fields.InputDataFields.image: tf.zeros((128, 128, 3)), | |
fields.InputDataFields.groundtruth_classes: | |
tf.constant([1, 1], tf.int32), | |
fields.InputDataFields.groundtruth_boxes: | |
tf.constant([[0.5, 0.5, 0.4, 0.5], [0.5, 0.5, 0.6, 0.6]], | |
tf.float32) | |
} | |
with mock.patch.object(tf, 'Assert') as tf_assert: | |
tf_assert.return_value = tf.no_op() | |
tensors = inputs.transform_input_data( | |
tensors, _fake_model_preprocessor_fn, _fake_image_resizer_fn, | |
num_classes=10) | |
self.assertAllClose(tensors[fields.InputDataFields.groundtruth_boxes], | |
[[0.5, 0.5, 0.6, 0.6]]) | |
if __name__ == '__main__': | |
tf.test.main() | |