evaluation_test.py 文件源码

def testEvaluationLoopTimeout(self):
    _, update_op = slim.metrics.streaming_accuracy(
        self._predictions, self._labels)
    init_op = tf.group(tf.global_variables_initializer(),
                       tf.local_variables_initializer())

    # Create checkpoint and log directories.
    chkpt_dir = os.path.join(self.get_temp_dir(), 'tmp_logs/')
    gfile.MakeDirs(chkpt_dir)
    logdir = os.path.join(self.get_temp_dir(), 'tmp_logs2/')
    gfile.MakeDirs(logdir)

    # Save initialized variables to checkpoint directory.
    saver = tf.train.Saver()
    with self.test_session() as sess:
      init_op.run()
      saver.save(sess, os.path.join(chkpt_dir, 'chkpt'))

    # Run the evaluation loop with a timeout.
    with self.test_session() as sess:
      start = time.time()
      slim.evaluation.evaluation_loop(
          '', chkpt_dir, logdir, eval_op=update_op,
          eval_interval_secs=2.0, timeout=6.0)
      end = time.time()

      # Check we've waited for the timeout.
      self.assertGreater(end - start, 6.0)

      # Then the timeout kicked in and stops the loop.
      self.assertLess(end - start, 8.0)