def testAddImageComparisonSummaries(self):
    summaries.add_image_comparison_summaries(
        get_gan_model(), display_diffs=True)

    self.assertEquals(1, len(ops.get_collection(ops.GraphKeys.SUMMARIES)))
    with self.test_session(use_gpu=True):
      summary.merge_all().eval()

  def testAddGanModelSummaries(self):
    summaries.add_gan_model_summaries(get_gan_model())

    self.assertEquals(3, len(ops.get_collection(ops.GraphKeys.SUMMARIES)))
    with self.test_session(use_gpu=True):
      variables.global_variables_initializer().run()
      summary.merge_all().eval()

  def _test_add_image_comparison_summaries_impl(self, get_model_fn,
                                                expected_num_summary_ops):
    summaries.add_image_comparison_summaries(get_model_fn(), display_diffs=True)

    self.assertEquals(expected_num_summary_ops,
                      len(ops.get_collection(ops.GraphKeys.SUMMARIES)))
    with self.test_session(use_gpu=True):
      summary.merge_all().eval()

  def _test_add_regularization_loss_summaries_impl(self, get_model_fn,
                                                   expected_num_summary_ops):
    summaries.add_regularization_loss_summaries(get_model_fn())

    self.assertEquals(expected_num_summary_ops,
                      len(ops.get_collection(ops.GraphKeys.SUMMARIES)))
    with self.test_session(use_gpu=True):
      summary.merge_all().eval()

  def _test_add_gan_model_summaries_impl(self, get_model_fn,
                                         expected_num_summary_ops):
    summaries.add_gan_model_summaries(get_model_fn())

    self.assertEquals(expected_num_summary_ops,
                      len(ops.get_collection(ops.GraphKeys.SUMMARIES)))
    with self.test_session(use_gpu=True):
      variables.global_variables_initializer().run()
      summary.merge_all().eval()

  def test_add_image_comparison_summaries_for_stargan(self):

    summaries.add_stargan_image_summaries(get_stargan_model())

    self.assertEquals(1, len(ops.get_collection(ops.GraphKeys.SUMMARIES)))

    with self.test_session(use_gpu=True) as sess:
      sess.run(variables.global_variables_initializer())
      summary.merge_all().eval()

  def testClassicSummaryOpsErrorOut(self):
    x = constant_op.constant(42)
    x_summary = summary.scalar('x', x)
    y = constant_op.constant([1, 3, 3, 7])
    y_summary = summary.histogram('hist', y)

    with self.assertRaisesRegexp(
        RuntimeError,
        r'Merging tf\.summary\.\* ops is not compatible with eager execution'):
      summary.merge([x_summary, y_summary])

    with self.assertRaisesRegexp(
        RuntimeError,
        r'Merging tf\.summary\.\* ops is not compatible with eager execution'):
      summary.merge_all()

 def testMergeAllSummaries(self):
   with ops.Graph().as_default():
     const = constant_op.constant(10.0)
     summ1 = summary.histogram("h", const)
     summ2 = summary.scalar("o", const, collections=["foo_key"])
     summ3 = summary.scalar("c", const)
     merge = summary.merge_all()
     self.assertEqual("MergeSummary", merge.op.type)
     self.assertEqual(2, len(merge.op.inputs))
     self.assertEqual(summ1, merge.op.inputs[0])
     self.assertEqual(summ3, merge.op.inputs[1])
     merge = summary.merge_all("foo_key")
     self.assertEqual("MergeSummary", merge.op.type)
     self.assertEqual(1, len(merge.op.inputs))
     self.assertEqual(summ2, merge.op.inputs[0])
     self.assertTrue(summary.merge_all("bar_key") is None)

  def set_model(self, model):
    self.model = model
    self.sess = K.get_session()
    if self.histogram_freq and self.merged is None:
      for layer in self.model.layers:

        for weight in layer.weights:
          tf_summary.histogram(weight.name, weight)
          if self.write_images:
            w_img = array_ops.squeeze(weight)
            shape = w_img.get_shape()
            if len(shape) > 1 and shape[0] > shape[1]:
              w_img = array_ops.transpose(w_img)
            if len(shape) == 1:
              w_img = array_ops.expand_dims(w_img, 0)
            w_img = array_ops.expand_dims(array_ops.expand_dims(w_img, 0), -1)
            tf_summary.image(weight.name, w_img)

        if hasattr(layer, 'output'):
          tf_summary.histogram('{}_out'.format(layer.name), layer.output)
    self.merged = tf_summary.merge_all()

    if self.write_graph:
      self.writer = tf_summary.FileWriter(self.log_dir, self.sess.graph)
    else:
      self.writer = tf_summary.FileWriter(self.log_dir)

 def _summary_computed():
   with ops.Graph().as_default():
     sv = supervisor.Supervisor(is_chief=False)
     sess = sv.prepare_or_wait_for_session("")
     summary.scalar("c1", constant_op.constant(1))
     summary.scalar("c2", constant_op.constant(2))
     summ = summary.merge_all()
     sv.summary_computed(sess, sess.run(summ))

  def set_model(self, model):
    """Sets Keras model and creates summary ops."""

    self.model = model
    self.sess = K.get_session()
    # only make histogram summary op if it hasn't already been made
    if self.histogram_freq and self.merged is None:
      for layer in self.model.layers:
        for weight in layer.weights:
          mapped_weight_name = weight.name.replace(':', '_')
          tf_summary.histogram(mapped_weight_name, weight)
          if self.write_images:
            w_img = array_ops.squeeze(weight)
            shape = K.int_shape(w_img)
            if len(shape) == 2:  # dense layer kernel case
              if shape[0] > shape[1]:
                w_img = array_ops.transpose(w_img)
                shape = K.int_shape(w_img)
              w_img = array_ops.reshape(w_img, [1, shape[0], shape[1], 1])
            elif len(shape) == 3:  # convnet case
              if K.image_data_format() == 'channels_last':
                # switch to channels_first to display
                # every kernel as a separate image
                w_img = array_ops.transpose(w_img, perm=[2, 0, 1])
                shape = K.int_shape(w_img)
              w_img = array_ops.reshape(w_img,
                                        [shape[0], shape[1], shape[2], 1])
            elif len(shape) == 1:  # bias case
              w_img = array_ops.reshape(w_img, [1, shape[0], 1, 1])
            else:
              # not possible to handle 3D convnets etc.
              continue

            shape = K.int_shape(w_img)
            assert len(shape) == 4 and shape[-1] in [1, 3, 4]
            tf_summary.image(mapped_weight_name, w_img)

        if self.write_grads:
          for weight in layer.trainable_weights:
            mapped_weight_name = weight.name.replace(':', '_')
            grads = model.optimizer.get_gradients(model.total_loss, weight)

            def is_indexed_slices(grad):
              return type(grad).__name__ == 'IndexedSlices'

            grads = [grad.values if is_indexed_slices(grad) else grad
                     for grad in grads]
            tf_summary.histogram('{}_grad'.format(mapped_weight_name), grads)

        if hasattr(layer, 'output'):
          tf_summary.histogram('{}_out'.format(layer.name), layer.output)
    self.merged = tf_summary.merge_all()

    if self.write_graph:
      self.writer = self._writer_class(self.log_dir, self.sess.graph)
    else:
      self.writer = self._writer_class(self.log_dir)

 def testNoLogdirButWantSummary(self):
   with ops.Graph().as_default():
     summary.scalar("c1", constant_op.constant(1))
     summary.scalar("c2", constant_op.constant(2))
     summary.scalar("c3", constant_op.constant(3))
     summ = summary.merge_all()
     sv = supervisor.Supervisor(logdir="", summary_op=None)
     sess = sv.prepare_or_wait_for_session("")
     with self.assertRaisesRegexp(RuntimeError, "requires a summary writer"):
       sv.summary_computed(sess, sess.run(summ))

  def testNoLogdirButExplicitSummaryWriter(self):
    logdir = self._test_dir("explicit_summary_writer")
    with ops.Graph().as_default():
      summary.scalar("c1", constant_op.constant(1))
      summary.scalar("c2", constant_op.constant(2))
      summary.scalar("c3", constant_op.constant(3))
      summ = summary.merge_all()
      sw = writer.FileWriter(logdir)
      sv = supervisor.Supervisor(logdir="", summary_op=None, summary_writer=sw)
      meta_graph_def = meta_graph.create_meta_graph_def()
      sess = sv.prepare_or_wait_for_session("")
      sv.summary_computed(sess, sess.run(summ))
      sess.close()
      # Wait to make sure everything is written to file before stopping.
      time.sleep(1)
      sv.stop()

    # Check the summary was written to 'logdir'
    rr = _summary_iterator(logdir)

    # The first event should list the file_version.
    ev = next(rr)
    self.assertEquals("brain.Event:2", ev.file_version)

    # The next one has the graph.
    ev = next(rr)
    ev_graph = graph_pb2.GraphDef()
    ev_graph.ParseFromString(ev.graph_def)
    self.assertProtoEquals(sess.graph.as_graph_def(add_shapes=True), ev_graph)

    # Stored MetaGraphDef
    ev = next(rr)
    ev_meta_graph = meta_graph_pb2.MetaGraphDef()
    ev_meta_graph.ParseFromString(ev.meta_graph_def)
    self.assertProtoEquals(meta_graph_def, ev_meta_graph)
    self.assertProtoEquals(
        sess.graph.as_graph_def(add_shapes=True), ev_meta_graph.graph_def)

    # The next one should have the values from the summary.
    ev = next(rr)
    self.assertProtoEquals("""
      value { tag: 'c1' simple_value: 1.0 }
      value { tag: 'c2' simple_value: 2.0 }
      value { tag: 'c3' simple_value: 3.0 }
      """, ev.summary)

    # The next one should be a stop message if we closed cleanly.
    ev = next(rr)
    self.assertEquals(event_pb2.SessionLog.STOP, ev.session_log.status)

    # We should be done.
    self.assertRaises(StopIteration, lambda: next(rr))

  def testManagedSessionDoNotKeepSummaryWriter(self):
    logdir = self._test_dir("managed_not_keep_summary_writer")
    with ops.Graph().as_default():
      summary.scalar("c1", constant_op.constant(1))
      summary.scalar("c2", constant_op.constant(2))
      summary.scalar("c3", constant_op.constant(3))
      summ = summary.merge_all()
      sv = supervisor.Supervisor(logdir=logdir, summary_op=None)
      with sv.managed_session(
          "", close_summary_writer=True, start_standard_services=False) as sess:
        sv.summary_computed(sess, sess.run(summ))
      # Sleep 1.2s to make sure that the next event file has a different name
      # than the current one.
      time.sleep(1.2)
      with sv.managed_session(
          "", close_summary_writer=True, start_standard_services=False) as sess:
        sv.summary_computed(sess, sess.run(summ))
    event_paths = sorted(glob.glob(os.path.join(logdir, "event*")))
    self.assertEquals(2, len(event_paths))
    # The two event files should have the same contents.
    for path in event_paths:
      # The summary iterator should report the summary once as we closed the
      # summary writer across the 2 sessions.
      rr = summary_iterator.summary_iterator(path)
      # The first event should list the file_version.
      ev = next(rr)
      self.assertEquals("brain.Event:2", ev.file_version)

      # The next one has the graph and metagraph.
      ev = next(rr)
      self.assertTrue(ev.graph_def)

      ev = next(rr)
      self.assertTrue(ev.meta_graph_def)

      # The next one should have the values from the summary.
      # But only once.
      ev = next(rr)
      self.assertProtoEquals("""
        value { tag: 'c1' simple_value: 1.0 }
        value { tag: 'c2' simple_value: 2.0 }
        value { tag: 'c3' simple_value: 3.0 }
        """, ev.summary)

      # The next one should be a stop message if we closed cleanly.
      ev = next(rr)
      self.assertEquals(event_pb2.SessionLog.STOP, ev.session_log.status)

      # We should be done.
      with self.assertRaises(StopIteration):
        next(rr)

  def testManagedSessionKeepSummaryWriter(self):
    logdir = self._test_dir("managed_keep_summary_writer")
    with ops.Graph().as_default():
      summary.scalar("c1", constant_op.constant(1))
      summary.scalar("c2", constant_op.constant(2))
      summary.scalar("c3", constant_op.constant(3))
      summ = summary.merge_all()
      sv = supervisor.Supervisor(logdir=logdir)
      with sv.managed_session(
          "", close_summary_writer=False,
          start_standard_services=False) as sess:
        sv.summary_computed(sess, sess.run(summ))
      with sv.managed_session(
          "", close_summary_writer=False,
          start_standard_services=False) as sess:
        sv.summary_computed(sess, sess.run(summ))
    # Now close the summary writer to flush the events.
    sv.summary_writer.close()
    # The summary iterator should report the summary twice as we reused
    # the same summary writer across the 2 sessions.
    rr = _summary_iterator(logdir)
    # The first event should list the file_version.
    ev = next(rr)
    self.assertEquals("brain.Event:2", ev.file_version)

    # The next one has the graph.
    ev = next(rr)
    self.assertTrue(ev.graph_def)

    ev = next(rr)
    self.assertTrue(ev.meta_graph_def)

    # The next one should have the values from the summary.
    ev = next(rr)
    self.assertProtoEquals("""
      value { tag: 'c1' simple_value: 1.0 }
      value { tag: 'c2' simple_value: 2.0 }
      value { tag: 'c3' simple_value: 3.0 }
      """, ev.summary)

    # The next one should also have the values from the summary.
    ev = next(rr)
    self.assertProtoEquals("""
      value { tag: 'c1' simple_value: 1.0 }
      value { tag: 'c2' simple_value: 2.0 }
      value { tag: 'c3' simple_value: 3.0 }
      """, ev.summary)

    # We should be done.
    self.assertRaises(StopIteration, lambda: next(rr))

  def _init_summary_op(self, summary_op=USE_DEFAULT):
    """Initializes summary_op.

    Args:
      summary_op: An Operation that returns a Summary for the event logs. If set
        to USE_DEFAULT, create an op that merges all the summaries.
    """
    if summary_op is Supervisor.USE_DEFAULT:
      summary_op = self._get_first_op_from_collection(ops.GraphKeys.SUMMARY_OP)
      if summary_op is None:
        summary_op = _summary.merge_all()
        if summary_op is not None:
          ops.add_to_collection(ops.GraphKeys.SUMMARY_OP, summary_op)
    self._summary_op = summary_op

 def testLogdirButExplicitlyNoSummaryWriter(self):
   logdir = self._test_dir("explicit_no_summary_writer")
   with ops.Graph().as_default():
     variables.VariableV1([1.0], name="foo")
     summary.scalar("c1", constant_op.constant(1))
     summary.scalar("c2", constant_op.constant(2))
     summary.scalar("c3", constant_op.constant(3))
     summ = summary.merge_all()
     sv = supervisor.Supervisor(logdir=logdir, summary_writer=None)
     sess = sv.prepare_or_wait_for_session("")
     # Check that a checkpoint is still be generated.
     self._wait_for_glob(sv.save_path, 3.0)
     # Check that we cannot write a summary
     with self.assertRaisesRegexp(RuntimeError, "requires a summary writer"):
       sv.summary_computed(sess, sess.run(summ))

  def testTrainWithNoneAsLogdirWhenUsingSummariesRaisesError(self):
    with ops.Graph().as_default():
      random_seed.set_random_seed(0)
      tf_inputs = constant_op.constant(self._inputs, dtype=dtypes.float32)
      tf_labels = constant_op.constant(self._labels, dtype=dtypes.float32)

      tf_predictions = LogisticClassifier(tf_inputs)
      loss_ops.log_loss(tf_predictions, tf_labels)
      total_loss = loss_ops.get_total_loss()
      summary.scalar('total_loss', total_loss)

      optimizer = gradient_descent.GradientDescentOptimizer(learning_rate=1.0)

      train_op = learning.create_train_op(total_loss, optimizer)
      summary_op = summary.merge_all()

      with self.assertRaises(ValueError):
        learning.train(
            train_op, None, number_of_steps=300, summary_op=summary_op)

  def verify_scalar_summary_is_written(self, print_summary):
    value = 3
    tensor = array_ops.ones([]) * value
    name = 'my_score'
    prefix = 'eval'
    summaries.add_scalar_summary(tensor, name, prefix, print_summary)

    output_dir = os.path.join(self.get_temp_dir(),
                              'scalar_summary_no_print_test')
    self.safe_create(output_dir)

    summary_op = summary.merge_all()

    summary_writer = summary.FileWriter(output_dir)
    with self.cached_session() as sess:
      new_summary = sess.run(summary_op)
      summary_writer.add_summary(new_summary, 1)
      summary_writer.flush()

    self.assert_scalar_summary(output_dir, {
        '%s/%s' % (prefix, name): value
    })

  def generate_run(self, run_name):
    if run_name == self._RUN_WITH_HISTOGRAM:
      (use_histogram, use_scalars) = (True, False)
    elif run_name == self._RUN_WITH_SCALARS:
      (use_histogram, use_scalars) = (False, True)
    else:
      assert False, 'Invalid run name: %r' % run_name
    ops.reset_default_graph()
    sess = session.Session()
    placeholder = array_ops.placeholder(dtypes.float32, shape=[3])
    if use_histogram:
      summary.histogram(self._HISTOGRAM_TAG, placeholder)
    if use_scalars:
      summary.scalar(self._SCALAR_TAG, math_ops.reduce_mean(placeholder))
    summ = summary.merge_all()

    subdir = os.path.join(self.logdir, run_name)
    writer = summary.FileWriter(subdir)
    writer.add_graph(sess.graph)
    for step in xrange(self._STEPS):
      feed_dict = {placeholder: [1 + step, 2 + step, 3 + step]}
      s = sess.run(summ, feed_dict=feed_dict)
      writer.add_summary(s, global_step=step)
    writer.close()