| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| """Tests for object detection model library.""" |
|
|
| from __future__ import absolute_import |
| from __future__ import division |
| from __future__ import print_function |
|
|
| import functools |
| import os |
|
|
| import numpy as np |
| import tensorflow as tf |
|
|
| from tensorflow.contrib.tpu.python.tpu import tpu_config |
| from tensorflow.contrib.tpu.python.tpu import tpu_estimator |
|
|
| from object_detection import inputs |
| from object_detection import model_hparams |
| from object_detection import model_lib |
| from object_detection.builders import model_builder |
| from object_detection.core import standard_fields as fields |
| from object_detection.utils import config_util |
|
|
|
|
| |
| |
| MODEL_NAME_FOR_TEST = 'ssd_inception_v2_pets' |
|
|
|
|
| def _get_data_path(): |
| """Returns an absolute path to TFRecord file.""" |
| return os.path.join(tf.resource_loader.get_data_files_path(), 'test_data', |
| 'pets_examples.record') |
|
|
|
|
| def get_pipeline_config_path(model_name): |
| """Returns path to the local pipeline config file.""" |
| return os.path.join(tf.resource_loader.get_data_files_path(), 'samples', |
| 'configs', model_name + '.config') |
|
|
|
|
| def _get_labelmap_path(): |
| """Returns an absolute path to label map file.""" |
| return os.path.join(tf.resource_loader.get_data_files_path(), 'data', |
| 'pet_label_map.pbtxt') |
|
|
|
|
| def _get_configs_for_model(model_name): |
| """Returns configurations for model.""" |
| filename = get_pipeline_config_path(model_name) |
| data_path = _get_data_path() |
| label_map_path = _get_labelmap_path() |
| configs = config_util.get_configs_from_pipeline_file(filename) |
| override_dict = { |
| 'train_input_path': data_path, |
| 'eval_input_path': data_path, |
| 'label_map_path': label_map_path |
| } |
| configs = config_util.merge_external_params_with_configs( |
| configs, kwargs_dict=override_dict) |
| return configs |
|
|
|
|
| def _make_initializable_iterator(dataset): |
| """Creates an iterator, and initializes tables. |
| |
| Args: |
| dataset: A `tf.data.Dataset` object. |
| |
| Returns: |
| A `tf.data.Iterator`. |
| """ |
| iterator = dataset.make_initializable_iterator() |
| tf.add_to_collection(tf.GraphKeys.TABLE_INITIALIZERS, iterator.initializer) |
| return iterator |
|
|
|
|
| class ModelLibTest(tf.test.TestCase): |
|
|
| @classmethod |
| def setUpClass(cls): |
| tf.reset_default_graph() |
|
|
| def _assert_model_fn_for_train_eval(self, configs, mode, |
| class_agnostic=False): |
| model_config = configs['model'] |
| train_config = configs['train_config'] |
| with tf.Graph().as_default(): |
| if mode == 'train': |
| features, labels = _make_initializable_iterator( |
| inputs.create_train_input_fn(configs['train_config'], |
| configs['train_input_config'], |
| configs['model'])()).get_next() |
| model_mode = tf.estimator.ModeKeys.TRAIN |
| batch_size = train_config.batch_size |
| elif mode == 'eval': |
| features, labels = _make_initializable_iterator( |
| inputs.create_eval_input_fn(configs['eval_config'], |
| configs['eval_input_config'], |
| configs['model'])()).get_next() |
| model_mode = tf.estimator.ModeKeys.EVAL |
| batch_size = 1 |
| elif mode == 'eval_on_train': |
| features, labels = _make_initializable_iterator( |
| inputs.create_eval_input_fn(configs['eval_config'], |
| configs['train_input_config'], |
| configs['model'])()).get_next() |
| model_mode = tf.estimator.ModeKeys.EVAL |
| batch_size = 1 |
|
|
| detection_model_fn = functools.partial( |
| model_builder.build, model_config=model_config, is_training=True) |
|
|
| hparams = model_hparams.create_hparams( |
| hparams_overrides='load_pretrained=false') |
|
|
| model_fn = model_lib.create_model_fn(detection_model_fn, configs, hparams) |
| estimator_spec = model_fn(features, labels, model_mode) |
|
|
| self.assertIsNotNone(estimator_spec.loss) |
| self.assertIsNotNone(estimator_spec.predictions) |
| if mode == 'eval' or mode == 'eval_on_train': |
| if class_agnostic: |
| self.assertNotIn('detection_classes', estimator_spec.predictions) |
| else: |
| detection_classes = estimator_spec.predictions['detection_classes'] |
| self.assertEqual(batch_size, detection_classes.shape.as_list()[0]) |
| self.assertEqual(tf.float32, detection_classes.dtype) |
| detection_boxes = estimator_spec.predictions['detection_boxes'] |
| detection_scores = estimator_spec.predictions['detection_scores'] |
| num_detections = estimator_spec.predictions['num_detections'] |
| self.assertEqual(batch_size, detection_boxes.shape.as_list()[0]) |
| self.assertEqual(tf.float32, detection_boxes.dtype) |
| self.assertEqual(batch_size, detection_scores.shape.as_list()[0]) |
| self.assertEqual(tf.float32, detection_scores.dtype) |
| self.assertEqual(tf.float32, num_detections.dtype) |
| if mode == 'eval': |
| self.assertIn('Detections_Left_Groundtruth_Right/0', |
| estimator_spec.eval_metric_ops) |
| if model_mode == tf.estimator.ModeKeys.TRAIN: |
| self.assertIsNotNone(estimator_spec.train_op) |
| return estimator_spec |
|
|
| def _assert_model_fn_for_predict(self, configs): |
| model_config = configs['model'] |
|
|
| with tf.Graph().as_default(): |
| features, _ = _make_initializable_iterator( |
| inputs.create_eval_input_fn(configs['eval_config'], |
| configs['eval_input_config'], |
| configs['model'])()).get_next() |
| detection_model_fn = functools.partial( |
| model_builder.build, model_config=model_config, is_training=False) |
|
|
| hparams = model_hparams.create_hparams( |
| hparams_overrides='load_pretrained=false') |
|
|
| model_fn = model_lib.create_model_fn(detection_model_fn, configs, hparams) |
| estimator_spec = model_fn(features, None, tf.estimator.ModeKeys.PREDICT) |
|
|
| self.assertIsNone(estimator_spec.loss) |
| self.assertIsNone(estimator_spec.train_op) |
| self.assertIsNotNone(estimator_spec.predictions) |
| self.assertIsNotNone(estimator_spec.export_outputs) |
| self.assertIn(tf.saved_model.signature_constants.PREDICT_METHOD_NAME, |
| estimator_spec.export_outputs) |
|
|
| def test_model_fn_in_train_mode(self): |
| """Tests the model function in TRAIN mode.""" |
| configs = _get_configs_for_model(MODEL_NAME_FOR_TEST) |
| self._assert_model_fn_for_train_eval(configs, 'train') |
|
|
| def test_model_fn_in_train_mode_freeze_all_variables(self): |
| """Tests model_fn TRAIN mode with all variables frozen.""" |
| configs = _get_configs_for_model(MODEL_NAME_FOR_TEST) |
| configs['train_config'].freeze_variables.append('.*') |
| with self.assertRaisesRegexp(ValueError, 'No variables to optimize'): |
| self._assert_model_fn_for_train_eval(configs, 'train') |
|
|
| def test_model_fn_in_train_mode_freeze_all_included_variables(self): |
| """Tests model_fn TRAIN mode with all included variables frozen.""" |
| configs = _get_configs_for_model(MODEL_NAME_FOR_TEST) |
| train_config = configs['train_config'] |
| train_config.update_trainable_variables.append('FeatureExtractor') |
| train_config.freeze_variables.append('.*') |
| with self.assertRaisesRegexp(ValueError, 'No variables to optimize'): |
| self._assert_model_fn_for_train_eval(configs, 'train') |
|
|
| def test_model_fn_in_train_mode_freeze_box_predictor(self): |
| """Tests model_fn TRAIN mode with FeatureExtractor variables frozen.""" |
| configs = _get_configs_for_model(MODEL_NAME_FOR_TEST) |
| train_config = configs['train_config'] |
| train_config.update_trainable_variables.append('FeatureExtractor') |
| train_config.update_trainable_variables.append('BoxPredictor') |
| train_config.freeze_variables.append('FeatureExtractor') |
| self._assert_model_fn_for_train_eval(configs, 'train') |
|
|
| def test_model_fn_in_eval_mode(self): |
| """Tests the model function in EVAL mode.""" |
| configs = _get_configs_for_model(MODEL_NAME_FOR_TEST) |
| self._assert_model_fn_for_train_eval(configs, 'eval') |
|
|
| def test_model_fn_in_eval_on_train_mode(self): |
| """Tests the model function in EVAL mode with train data.""" |
| configs = _get_configs_for_model(MODEL_NAME_FOR_TEST) |
| self._assert_model_fn_for_train_eval(configs, 'eval_on_train') |
|
|
| def test_model_fn_in_predict_mode(self): |
| """Tests the model function in PREDICT mode.""" |
| configs = _get_configs_for_model(MODEL_NAME_FOR_TEST) |
| self._assert_model_fn_for_predict(configs) |
|
|
| def test_create_estimator_and_inputs(self): |
| """Tests that Estimator and input function are constructed correctly.""" |
| run_config = tf.estimator.RunConfig() |
| hparams = model_hparams.create_hparams( |
| hparams_overrides='load_pretrained=false') |
| pipeline_config_path = get_pipeline_config_path(MODEL_NAME_FOR_TEST) |
| train_steps = 20 |
| train_and_eval_dict = model_lib.create_estimator_and_inputs( |
| run_config, |
| hparams, |
| pipeline_config_path, |
| train_steps=train_steps) |
| estimator = train_and_eval_dict['estimator'] |
| train_steps = train_and_eval_dict['train_steps'] |
| self.assertIsInstance(estimator, tf.estimator.Estimator) |
| self.assertEqual(20, train_steps) |
| self.assertIn('train_input_fn', train_and_eval_dict) |
| self.assertIn('eval_input_fns', train_and_eval_dict) |
| self.assertIn('eval_on_train_input_fn', train_and_eval_dict) |
|
|
| def test_create_estimator_with_default_train_eval_steps(self): |
| """Tests that number of train/eval defaults to config values.""" |
| run_config = tf.estimator.RunConfig() |
| hparams = model_hparams.create_hparams( |
| hparams_overrides='load_pretrained=false') |
| pipeline_config_path = get_pipeline_config_path(MODEL_NAME_FOR_TEST) |
| configs = config_util.get_configs_from_pipeline_file(pipeline_config_path) |
| config_train_steps = configs['train_config'].num_steps |
| train_and_eval_dict = model_lib.create_estimator_and_inputs( |
| run_config, hparams, pipeline_config_path) |
| estimator = train_and_eval_dict['estimator'] |
| train_steps = train_and_eval_dict['train_steps'] |
|
|
| self.assertIsInstance(estimator, tf.estimator.Estimator) |
| self.assertEqual(config_train_steps, train_steps) |
|
|
| def test_create_tpu_estimator_and_inputs(self): |
| """Tests that number of train/eval defaults to config values.""" |
|
|
| run_config = tpu_config.RunConfig() |
| hparams = model_hparams.create_hparams( |
| hparams_overrides='load_pretrained=false') |
| pipeline_config_path = get_pipeline_config_path(MODEL_NAME_FOR_TEST) |
| train_steps = 20 |
| train_and_eval_dict = model_lib.create_estimator_and_inputs( |
| run_config, |
| hparams, |
| pipeline_config_path, |
| train_steps=train_steps, |
| use_tpu_estimator=True) |
| estimator = train_and_eval_dict['estimator'] |
| train_steps = train_and_eval_dict['train_steps'] |
|
|
| self.assertIsInstance(estimator, tpu_estimator.TPUEstimator) |
| self.assertEqual(20, train_steps) |
|
|
| def test_create_train_and_eval_specs(self): |
| """Tests that `TrainSpec` and `EvalSpec` is created correctly.""" |
| run_config = tf.estimator.RunConfig() |
| hparams = model_hparams.create_hparams( |
| hparams_overrides='load_pretrained=false') |
| pipeline_config_path = get_pipeline_config_path(MODEL_NAME_FOR_TEST) |
| train_steps = 20 |
| train_and_eval_dict = model_lib.create_estimator_and_inputs( |
| run_config, |
| hparams, |
| pipeline_config_path, |
| train_steps=train_steps) |
| train_input_fn = train_and_eval_dict['train_input_fn'] |
| eval_input_fns = train_and_eval_dict['eval_input_fns'] |
| eval_on_train_input_fn = train_and_eval_dict['eval_on_train_input_fn'] |
| predict_input_fn = train_and_eval_dict['predict_input_fn'] |
| train_steps = train_and_eval_dict['train_steps'] |
|
|
| train_spec, eval_specs = model_lib.create_train_and_eval_specs( |
| train_input_fn, |
| eval_input_fns, |
| eval_on_train_input_fn, |
| predict_input_fn, |
| train_steps, |
| eval_on_train_data=True, |
| final_exporter_name='exporter', |
| eval_spec_names=['holdout']) |
| self.assertEqual(train_steps, train_spec.max_steps) |
| self.assertEqual(2, len(eval_specs)) |
| self.assertEqual(None, eval_specs[0].steps) |
| self.assertEqual('holdout', eval_specs[0].name) |
| self.assertEqual('exporter', eval_specs[0].exporters[0].name) |
| self.assertEqual(None, eval_specs[1].steps) |
| self.assertEqual('eval_on_train', eval_specs[1].name) |
|
|
| def test_experiment(self): |
| """Tests that the `Experiment` object is constructed correctly.""" |
| run_config = tf.estimator.RunConfig() |
| hparams = model_hparams.create_hparams( |
| hparams_overrides='load_pretrained=false') |
| pipeline_config_path = get_pipeline_config_path(MODEL_NAME_FOR_TEST) |
| experiment = model_lib.populate_experiment( |
| run_config, |
| hparams, |
| pipeline_config_path, |
| train_steps=10, |
| eval_steps=20) |
| self.assertEqual(10, experiment.train_steps) |
| self.assertEqual(None, experiment.eval_steps) |
|
|
|
|
| class UnbatchTensorsTest(tf.test.TestCase): |
|
|
| def test_unbatch_without_unpadding(self): |
| image_placeholder = tf.placeholder(tf.float32, [2, None, None, None]) |
| groundtruth_boxes_placeholder = tf.placeholder(tf.float32, [2, None, None]) |
| groundtruth_classes_placeholder = tf.placeholder(tf.float32, |
| [2, None, None]) |
| groundtruth_weights_placeholder = tf.placeholder(tf.float32, [2, None]) |
|
|
| tensor_dict = { |
| fields.InputDataFields.image: |
| image_placeholder, |
| fields.InputDataFields.groundtruth_boxes: |
| groundtruth_boxes_placeholder, |
| fields.InputDataFields.groundtruth_classes: |
| groundtruth_classes_placeholder, |
| fields.InputDataFields.groundtruth_weights: |
| groundtruth_weights_placeholder |
| } |
| unbatched_tensor_dict = model_lib.unstack_batch( |
| tensor_dict, unpad_groundtruth_tensors=False) |
|
|
| with self.test_session() as sess: |
| unbatched_tensor_dict_out = sess.run( |
| unbatched_tensor_dict, |
| feed_dict={ |
| image_placeholder: |
| np.random.rand(2, 4, 4, 3).astype(np.float32), |
| groundtruth_boxes_placeholder: |
| np.random.rand(2, 5, 4).astype(np.float32), |
| groundtruth_classes_placeholder: |
| np.random.rand(2, 5, 6).astype(np.float32), |
| groundtruth_weights_placeholder: |
| np.random.rand(2, 5).astype(np.float32) |
| }) |
| for image_out in unbatched_tensor_dict_out[fields.InputDataFields.image]: |
| self.assertAllEqual(image_out.shape, [4, 4, 3]) |
| for groundtruth_boxes_out in unbatched_tensor_dict_out[ |
| fields.InputDataFields.groundtruth_boxes]: |
| self.assertAllEqual(groundtruth_boxes_out.shape, [5, 4]) |
| for groundtruth_classes_out in unbatched_tensor_dict_out[ |
| fields.InputDataFields.groundtruth_classes]: |
| self.assertAllEqual(groundtruth_classes_out.shape, [5, 6]) |
| for groundtruth_weights_out in unbatched_tensor_dict_out[ |
| fields.InputDataFields.groundtruth_weights]: |
| self.assertAllEqual(groundtruth_weights_out.shape, [5]) |
|
|
| def test_unbatch_and_unpad_groundtruth_tensors(self): |
| image_placeholder = tf.placeholder(tf.float32, [2, None, None, None]) |
| groundtruth_boxes_placeholder = tf.placeholder(tf.float32, [2, 5, None]) |
| groundtruth_classes_placeholder = tf.placeholder(tf.float32, [2, 5, None]) |
| groundtruth_weights_placeholder = tf.placeholder(tf.float32, [2, 5]) |
| num_groundtruth_placeholder = tf.placeholder(tf.int32, [2]) |
|
|
| tensor_dict = { |
| fields.InputDataFields.image: |
| image_placeholder, |
| fields.InputDataFields.groundtruth_boxes: |
| groundtruth_boxes_placeholder, |
| fields.InputDataFields.groundtruth_classes: |
| groundtruth_classes_placeholder, |
| fields.InputDataFields.groundtruth_weights: |
| groundtruth_weights_placeholder, |
| fields.InputDataFields.num_groundtruth_boxes: |
| num_groundtruth_placeholder |
| } |
| unbatched_tensor_dict = model_lib.unstack_batch( |
| tensor_dict, unpad_groundtruth_tensors=True) |
| with self.test_session() as sess: |
| unbatched_tensor_dict_out = sess.run( |
| unbatched_tensor_dict, |
| feed_dict={ |
| image_placeholder: |
| np.random.rand(2, 4, 4, 3).astype(np.float32), |
| groundtruth_boxes_placeholder: |
| np.random.rand(2, 5, 4).astype(np.float32), |
| groundtruth_classes_placeholder: |
| np.random.rand(2, 5, 6).astype(np.float32), |
| groundtruth_weights_placeholder: |
| np.random.rand(2, 5).astype(np.float32), |
| num_groundtruth_placeholder: |
| np.array([3, 3], np.int32) |
| }) |
| for image_out in unbatched_tensor_dict_out[fields.InputDataFields.image]: |
| self.assertAllEqual(image_out.shape, [4, 4, 3]) |
| for groundtruth_boxes_out in unbatched_tensor_dict_out[ |
| fields.InputDataFields.groundtruth_boxes]: |
| self.assertAllEqual(groundtruth_boxes_out.shape, [3, 4]) |
| for groundtruth_classes_out in unbatched_tensor_dict_out[ |
| fields.InputDataFields.groundtruth_classes]: |
| self.assertAllEqual(groundtruth_classes_out.shape, [3, 6]) |
| for groundtruth_weights_out in unbatched_tensor_dict_out[ |
| fields.InputDataFields.groundtruth_weights]: |
| self.assertAllEqual(groundtruth_weights_out.shape, [3]) |
|
|
|
|
| if __name__ == '__main__': |
| tf.test.main() |
|
|
