def test_local_variable(self):
   with self.cached_session() as sess:
     self.assertEquals([], variables_lib.local_variables())
     value0 = 42
     variables_lib2.local_variable(value0)
     value1 = 43
     variables_lib2.local_variable(value1)
     variables = variables_lib.local_variables()
     self.assertEquals(2, len(variables))
     self.assertRaises(errors_impl.OpError, sess.run, variables)
     variables_lib.variables_initializer(variables).run()
     self.assertAllEqual(set([value0, value1]), set(sess.run(variables)))

 def test_unweighted_top_k(self):
   p_obj = metrics.Precision(top_k=3)
   y_pred = constant_op.constant([0.2, 0.1, 0.5, 0, 0.2], shape=(1, 5))
   y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
   self.evaluate(variables.variables_initializer(p_obj.variables))
   result = p_obj(y_true, y_pred)
   self.assertAlmostEqual(1. / 3, self.evaluate(result))

 def test_unweighted_with_threshold(self):
   r_obj = metrics.Recall(thresholds=[0.5, 0.7])
   y_pred = constant_op.constant([1, 0, 0.6, 0], shape=(1, 4))
   y_true = constant_op.constant([0, 1, 1, 0], shape=(1, 4))
   self.evaluate(variables.variables_initializer(r_obj.variables))
   result = r_obj(y_true, y_pred)
   self.assertArrayNear([0.5, 0.], self.evaluate(result), 0)

 def test_placeholder_with_default_default(self):
   with self.cached_session() as sess, self.test_scope():
     v = resource_variable_ops.ResourceVariable(4.0)
     ph = array_ops.placeholder_with_default(v, shape=[])
     out = ph * 2
     sess.run(variables.variables_initializer([v]))
     self.assertEqual(8.0, self.evaluate(out))

  def create_checkpoint_from_values(self,
                                    var_names_to_values,
                                    checkpoint_dir,
                                    global_step=None):
    """Creates a checkpoint from a mapping of name to values in model_dir.

    Args:
      var_names_to_values: a map from variable names to values.
      checkpoint_dir: the directory where the checkpoint will be saved.
      global_step: the global step used to save the checkpoint.

    Returns:
      the model_path to the checkpoint.
    """
    var_list = []
    with session.Session('', graph=ops.Graph()) as sess:
      # Create a set of variables to save in the checkpoint.
      for var_name in var_names_to_values:
        var_value = var_names_to_values[var_name]
        var_list.append(variables_lib.Variable(var_value, name=var_name))
      saver = saver_lib.Saver(var_list)
      init_op = variables_lib.variables_initializer(var_list)
      sess.run(init_op)
      # Save the initialized values in the file at 'checkpoint_dir'
      return saver.save(sess, checkpoint_dir, global_step=global_step)

  def test_mean(self):
    m = metrics.Mean(name='my_mean')

    # check config
    self.assertEqual(m.name, 'my_mean')
    self.assertTrue(m.stateful)
    self.assertEqual(m.dtype, dtypes.float32)
    self.assertEqual(len(m.variables), 2)
    self.evaluate(variables.variables_initializer(m.variables))

    # check initial state
    self.assertEqual(self.evaluate(m.total), 0)
    self.assertEqual(self.evaluate(m.count), 0)

    # check __call__()
    self.assertEqual(self.evaluate(m(100)), 100)
    self.assertEqual(self.evaluate(m.total), 100)
    self.assertEqual(self.evaluate(m.count), 1)

    # check update_state() and result() + state accumulation + tensor input
    update_op = m.update_state(ops.convert_n_to_tensor([1, 5]))
    self.evaluate(update_op)
    self.assertAlmostEqual(self.evaluate(m.result()), 106 / 3, 2)
    self.assertEqual(self.evaluate(m.total), 106)  # 100 + 1 + 5
    self.assertEqual(self.evaluate(m.count), 3)

    # check reset_states()
    m.reset_states()
    self.assertEqual(self.evaluate(m.total), 0)
    self.assertEqual(self.evaluate(m.count), 0)

 def test_div_by_zero(self):
   r_obj = metrics.Recall()
   y_pred = constant_op.constant([0, 0, 0, 0])
   y_true = constant_op.constant([0, 0, 0, 0])
   self.evaluate(variables.variables_initializer(r_obj.variables))
   result = r_obj(y_true, y_pred)
   self.assertEqual(0, self.evaluate(result))

  def test_binary_accuracy(self):
    acc_obj = metrics.BinaryAccuracy(name='my acc')

    # check config
    self.assertEqual(acc_obj.name, 'my acc')
    self.assertTrue(acc_obj.stateful)
    self.assertEqual(len(acc_obj.variables), 2)
    self.assertEqual(acc_obj.dtype, dtypes.float32)
    self.evaluate(variables.variables_initializer(acc_obj.variables))

    # verify that correct value is returned
    update_op = acc_obj.update_state([[1], [0]], [[1], [0]])
    self.evaluate(update_op)
    result = self.evaluate(acc_obj.result())
    self.assertEqual(result, 1)  # 2/2

    # check y_pred squeeze
    update_op = acc_obj.update_state([[1], [1]], [[[1]], [[0]]])
    self.evaluate(update_op)
    result = self.evaluate(acc_obj.result())
    self.assertAlmostEqual(result, 0.75, 2)  # 3/4

    # check y_true squeeze
    result_t = acc_obj([[[1]], [[1]]], [[1], [0]])
    result = self.evaluate(result_t)
    self.assertAlmostEqual(result, 0.67, 2)  # 4/6

    # check with sample_weight
    result_t = acc_obj([[1], [1]], [[1], [0]], [[0.5], [0.2]])
    result = self.evaluate(result_t)
    self.assertAlmostEqual(result, 0.67, 2)  # 4.5/6.7

 def test_unweighted(self):
   r_obj = metrics.Recall()
   y_pred = constant_op.constant([1, 0, 1, 0], shape=(1, 4))
   y_true = constant_op.constant([0, 1, 1, 0], shape=(1, 4))
   self.evaluate(variables.variables_initializer(r_obj.variables))
   result = r_obj(y_true, y_pred)
   self.assertAlmostEqual(0.5, self.evaluate(result))

  def test_example(self):
    with self.test_session() as session:
      for tower_id in range(3):
        self.create_tower_metrics(tower_id)

      session.run(
          variables.variables_initializer(
              ops_lib.get_collection(ops_lib.GraphKeys.METRIC_VARIABLES)))

      session.run(
          replicate_model_fn._reduce_metric_variables(number_of_towers=3))

      # 1st tower = 1.3, 2.3,  [3.3, 3.5, 3.7]
      # 2nd tower = 2.6, 4.6,  [6.6, 7.0, 7.4]
      # 3rd tower = 3.9, 6.9,  [9.9, 10.5, 11.1]
      # Reduced =   7.8, 13.8, [19.8, 21.0, 22.2]
      # Towers are accumulated in the first tower.
      local_metrics = session.run(
          ops_lib.get_collection(ops_lib.GraphKeys.METRIC_VARIABLES))

      self.assertNear(7.8, local_metrics[0], 0.01)
      self.assertNear(13.8, local_metrics[1], 0.01)
      self.assertAllClose([19.8, 21., 22.1], local_metrics[2], 0.01)
      self.assertNear(0.0, local_metrics[3], 0.01)
      self.assertNear(0.0, local_metrics[4], 0.01)
      self.assertAllClose([0.0, 0.0, 0.0], local_metrics[5], 0.01)
      self.assertNear(0.0, local_metrics[6], 0.01)
      self.assertNear(0.0, local_metrics[7], 0.01)
      self.assertAllClose([0.0, 0.0, 0.0], local_metrics[8], 0.01)

  def testConvertVariablesToConstsWithFunctions(self):
    @function.Defun(dtypes.float32)
    def plus_one(x):
      return x + 1.0

    with ops.Graph().as_default():
      variable_node = variables.Variable(1.0, name="variable_node")
      _ = variables.Variable(1.0, name="unused_variable_node")
      defun_node = plus_one(variable_node)
      output_node = math_ops_lib.multiply(
          defun_node, 2.0, name="output_node")

      with session.Session() as sess:
        init = variables.variables_initializer([variable_node])
        sess.run(init)
        output = sess.run(output_node)
        self.assertNear(4.0, output, 0.00001)
        variable_graph_def = sess.graph.as_graph_def()

        # First get the constant_graph_def when variable_names_whitelist is set,
        # note that if variable_names_whitelist is not set an error will be
        # thrown because unused_variable_node is not initialized.
        constant_graph_def = graph_util.convert_variables_to_constants(
            sess,
            variable_graph_def, ["output_node"],
            variable_names_whitelist=set(["variable_node"]))

        self.assertEqual(variable_graph_def.library,
                         constant_graph_def.library)

 def test_placeholder_with_default_fed(self):
   with self.test_session() as sess, self.test_scope():
     v = resource_variable_ops.ResourceVariable(4.0)
     ph = array_ops.placeholder_with_default(v, shape=[])
     out = ph * 2
     sess.run(variables.variables_initializer([v]))
     self.assertEqual(2.0, sess.run(out, {ph: 1.0}))

  def test_save_restore(self):
    checkpoint_directory = self.get_temp_dir()
    checkpoint_prefix = os.path.join(checkpoint_directory, 'ckpt')
    m = metrics.Mean()
    checkpoint = checkpointable_utils.Checkpoint(mean=m)
    self.evaluate(variables.variables_initializer(m.variables))

    # update state
    self.evaluate(m(100.))
    self.evaluate(m(200.))

    # save checkpoint and then add an update
    save_path = checkpoint.save(checkpoint_prefix)
    self.evaluate(m(1000.))

    # restore to the same checkpoint mean object
    checkpoint.restore(save_path).assert_consumed().run_restore_ops()
    self.evaluate(m(300.))
    self.assertEqual(200., self.evaluate(m.result()))

    # restore to a different checkpoint mean object
    restore_mean = metrics.Mean()
    restore_checkpoint = checkpointable_utils.Checkpoint(mean=restore_mean)
    status = restore_checkpoint.restore(save_path)
    restore_update = restore_mean(300.)
    status.assert_consumed().run_restore_ops()
    self.evaluate(restore_update)
    self.assertEqual(200., self.evaluate(restore_mean.result()))
    self.assertEqual(3, self.evaluate(restore_mean.count))

  def test_reduce_is_idempotent(self):
    with self.test_session() as session:
      for tower_id in range(3):
        self.create_tower_metrics(tower_id)

      session.run(
          variables.variables_initializer(
              ops_lib.get_collection(ops_lib.GraphKeys.METRIC_VARIABLES)))

      for _ in range(20):
        session.run(
            replicate_model_fn._reduce_metric_variables(number_of_towers=3))

      local_metrics = session.run(
          ops_lib.get_collection(ops_lib.GraphKeys.METRIC_VARIABLES))

      self.assertNear(7.8, local_metrics[0], 0.01)
      self.assertNear(13.8, local_metrics[1], 0.01)
      self.assertAllClose([19.8, 21., 22.1], local_metrics[2], 0.01)
      self.assertNear(0.0, local_metrics[3], 0.01)
      self.assertNear(0.0, local_metrics[4], 0.01)
      self.assertAllClose([0.0, 0.0, 0.0], local_metrics[5], 0.01)
      self.assertNear(0.0, local_metrics[6], 0.01)
      self.assertNear(0.0, local_metrics[7], 0.01)
      self.assertAllClose([0.0, 0.0, 0.0], local_metrics[8], 0.01)

 def test_unweighted_all_correct(self):
   s_obj = metrics.SensitivityAtSpecificity(0.7)
   inputs = np.random.randint(0, 2, size=(100, 1))
   y_pred = constant_op.constant(inputs, dtype=dtypes.float32)
   y_true = constant_op.constant(inputs)
   self.evaluate(variables.variables_initializer(s_obj.variables))
   result = s_obj(y_true, y_pred)
   self.assertAlmostEqual(1, self.evaluate(result))

 def testSparseRead0DIndices(self):
   for dtype in self.numeric_types:
     init = np.array([[0, 1, 2, 3], [4, 5, 6, 7], [8, 9, 10, 11]], dtype=dtype)
     with self.test_session() as sess, self.test_scope():
       v = resource_variable_ops.ResourceVariable(init)
       sess.run(variables.variables_initializer([v]))
       x = v.sparse_read(2)
       self.assertAllClose(np.array([8, 9, 10, 11], dtype=dtype), sess.run(x))

 def initialize_op(self):
   """Returns an op for initializing tensorflow variables."""
   all_vars = self._row_factors + self._col_factors
   all_vars.extend([self._row_gramian, self._col_gramian])
   if self._row_weights is not None:
     assert self._col_weights is not None
     all_vars.extend(self._row_weights + self._col_weights)
   return variables.variables_initializer(all_vars)

 def test_unweighted_all_incorrect(self):
   r_obj = metrics.Recall(thresholds=[0.5])
   inputs = np.random.randint(0, 2, size=(100, 1))
   y_pred = constant_op.constant(inputs)
   y_true = constant_op.constant(1 - inputs)
   self.evaluate(variables.variables_initializer(r_obj.variables))
   result = r_obj(y_true, y_pred)
   self.assertAlmostEqual(0, self.evaluate(result))

  def test_layer_output(self, attention_cls):
    attention = attention_cls(self.units, self.memory)
    score = attention([self.query, self.state])
    self.evaluate(variables.variables_initializer(attention.variables))

    score_val = self.evaluate(score)
    self.assertLen(score_val, 2)
    self.assertEqual(score_val[0].shape, (self.batch, self.timestep))
    self.assertEqual(score_val[1].shape, (self.batch, self.timestep))

 def test_extreme_thresholds(self):
   r_obj = metrics.Recall(thresholds=[-1.0, 2.0])  # beyond values range
   y_pred = math_ops.cast(
       constant_op.constant([1, 0, 1, 0], shape=(1, 4)), dtype=dtypes.float32)
   y_true = math_ops.cast(
       constant_op.constant([0, 1, 1, 1], shape=(1, 4)), dtype=dtypes.float32)
   self.evaluate(variables.variables_initializer(r_obj.variables))
   result = r_obj(y_true, y_pred)
   self.assertArrayNear([1.0, 0.], self.evaluate(result), 0)