# Copyright 2016 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 metrics."""

from __future__ import absolute_import

from __future__ import division

from __future__ import print_function

import functools

import math

import numpy as np

from six.moves import xrange # pylint: disable=redefined-builtin

from tensorflow.python.framework import constant_op

from tensorflow.python.framework import dtypes as dtypes_lib

from tensorflow.python.framework import errors_impl

from tensorflow.python.framework import ops

from tensorflow.python.framework import sparse_tensor

from tensorflow.python.ops import array_ops

from tensorflow.python.ops import data_flow_ops

from tensorflow.python.ops import math_ops

from tensorflow.python.ops import metrics

from tensorflow.python.ops import random_ops

from tensorflow.python.ops import variables

import tensorflow.python.ops.data_flow_grad # pylint: disable=unused-import

import tensorflow.python.ops.nn_grad # pylint: disable=unused-import

from tensorflow.python.platform import test

NAN = float('nan')

def _enqueue_vector(sess, queue, values, shape=None):

if not shape:

shape = (1, len(values))

dtype = queue.dtypes[0]

sess.run(

queue.enqueue(constant_op.constant(

values, dtype=dtype, shape=shape)))

def _binary_2d_label_to_2d_sparse_value(labels):

"""Convert dense 2D binary indicator to sparse ID.

Only 1 values in `labels` are included in result.

Args:

labels: Dense 2D binary indicator, shape [batch_size, num_classes].

Returns:

`SparseTensorValue` of shape [batch_size, num_classes], where num_classes

is the number of `1` values in each row of `labels`. Values are indices

of `1` values along the last dimension of `labels`.

"""

indices = []

values = []

batch = 0

for row in labels:

label = 0

xi = 0

for x in row:

if x == 1:

indices.append([batch, xi])

values.append(label)

xi += 1

else:

assert x == 0

label += 1

batch += 1

shape = [len(labels), len(labels[0])]

return sparse_tensor.SparseTensorValue(

np.array(indices, np.int64),

np.array(values, np.int64), np.array(shape, np.int64))

def _binary_2d_label_to_1d_sparse_value(labels):

"""Convert dense 2D binary indicator to sparse ID.

Only 1 values in `labels` are included in result.

Args:

labels: Dense 2D binary indicator, shape [batch_size, num_classes]. Each

row must contain exactly 1 `1` value.

Returns:

`SparseTensorValue` of shape [batch_size]. Values are indices of `1` values

along the last dimension of `labels`.

Raises:

ValueError: if there is not exactly 1 `1` value per row of `labels`.

"""

indices = []

values = []

batch = 0

for row in labels:

label = 0

xi = 0

for x in row:

if x == 1:

indices.append([batch])

values.append(label)

xi += 1

else:

assert x == 0

label += 1

batch += 1

if indices != [[i] for i in range(len(labels))]:

raise ValueError('Expected 1 label/example, got %s.' % indices)

shape = [len(labels)]

return sparse_tensor.SparseTensorValue(

np.array(indices, np.int64),

np.array(values, np.int64), np.array(shape, np.int64))

def _binary_3d_label_to_sparse_value(labels):

"""Convert dense 3D binary indicator tensor to sparse tensor.

Only 1 values in `labels` are included in result.

Args:

labels: Dense 2D binary indicator tensor.

Returns:

`SparseTensorValue` whose values are indices along the last dimension of

`labels`.

"""

indices = []

values = []

for d0, labels_d0 in enumerate(labels):

for d1, labels_d1 in enumerate(labels_d0):

d2 = 0

for class_id, label in enumerate(labels_d1):

if label == 1:

values.append(class_id)

indices.append([d0, d1, d2])

d2 += 1

else:

assert label == 0

shape = [len(labels), len(labels[0]), len(labels[0][0])]

return sparse_tensor.SparseTensorValue(

np.array(indices, np.int64),

np.array(values, np.int64), np.array(shape, np.int64))

def _assert_nan(test_case, actual):

test_case.assertTrue(math.isnan(actual), 'Expected NAN, got %s.' % actual)

def _assert_local_variables(test_case, expected):

test_case.assertEquals(

set(expected), set(v.name for v in variables.local_variables()))

def _test_values(shape):

return np.reshape(np.cumsum(np.ones(shape)), newshape=shape)

class MeanTest(test.TestCase):

def setUp(self):

ops.reset_default_graph()

def testVars(self):

metrics.mean(array_ops.ones([4, 3]))

_assert_local_variables(self, ('mean/count:0', 'mean/total:0'))

def testMetricsCollection(self):

my_collection_name = '__metrics__'

mean, _ = metrics.mean(

array_ops.ones([4, 3]), metrics_collections=[my_collection_name])

self.assertListEqual(ops.get_collection(my_collection_name), [mean])

def testUpdatesCollection(self):

my_collection_name = '__updates__'

_, update_op = metrics.mean(

array_ops.ones([4, 3]), updates_collections=[my_collection_name])

self.assertListEqual(ops.get_collection(my_collection_name), [update_op])

def testBasic(self):

with self.test_session() as sess:

values_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 2))

_enqueue_vector(sess, values_queue, [0, 1])

_enqueue_vector(sess, values_queue, [-4.2, 9.1])

_enqueue_vector(sess, values_queue, [6.5, 0])

_enqueue_vector(sess, values_queue, [-3.2, 4.0])

values = values_queue.dequeue()

mean, update_op = metrics.mean(values)

sess.run(variables.local_variables_initializer())

for _ in range(4):

sess.run(update_op)

self.assertAlmostEqual(1.65, sess.run(mean), 5)

def testUpdateOpsReturnsCurrentValue(self):

with self.test_session() as sess:

values_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 2))

_enqueue_vector(sess, values_queue, [0, 1])

_enqueue_vector(sess, values_queue, [-4.2, 9.1])

_enqueue_vector(sess, values_queue, [6.5, 0])

_enqueue_vector(sess, values_queue, [-3.2, 4.0])

values = values_queue.dequeue()

mean, update_op = metrics.mean(values)

sess.run(variables.local_variables_initializer())

self.assertAlmostEqual(0.5, sess.run(update_op), 5)

self.assertAlmostEqual(1.475, sess.run(update_op), 5)

self.assertAlmostEqual(12.4 / 6.0, sess.run(update_op), 5)

self.assertAlmostEqual(1.65, sess.run(update_op), 5)

self.assertAlmostEqual(1.65, sess.run(mean), 5)

def testUnweighted(self):

values = _test_values((3, 2, 4, 1))

mean_results = (

metrics.mean(values),

metrics.mean(values, weights=1.0),

metrics.mean(values, weights=np.ones((1, 1, 1))),

metrics.mean(values, weights=np.ones((1, 1, 1, 1))),

metrics.mean(values, weights=np.ones((1, 1, 1, 1, 1))),

metrics.mean(values, weights=np.ones((1, 1, 4))),

metrics.mean(values, weights=np.ones((1, 1, 4, 1))),

metrics.mean(values, weights=np.ones((1, 2, 1))),

metrics.mean(values, weights=np.ones((1, 2, 1, 1))),

metrics.mean(values, weights=np.ones((1, 2, 4))),

metrics.mean(values, weights=np.ones((1, 2, 4, 1))),

metrics.mean(values, weights=np.ones((3, 1, 1))),

metrics.mean(values, weights=np.ones((3, 1, 1, 1))),

metrics.mean(values, weights=np.ones((3, 1, 4))),

metrics.mean(values, weights=np.ones((3, 1, 4, 1))),

metrics.mean(values, weights=np.ones((3, 2, 1))),

metrics.mean(values, weights=np.ones((3, 2, 1, 1))),

metrics.mean(values, weights=np.ones((3, 2, 4))),

metrics.mean(values, weights=np.ones((3, 2, 4, 1))),

metrics.mean(values, weights=np.ones((3, 2, 4, 1, 1))),)

expected = np.mean(values)

with self.test_session():

variables.local_variables_initializer().run()

for mean_result in mean_results:

mean, update_op = mean_result

self.assertAlmostEqual(expected, update_op.eval())

self.assertAlmostEqual(expected, mean.eval())

def _test_3d_weighted(self, values, weights):

expected = (

np.sum(np.multiply(weights, values)) /

np.sum(np.multiply(weights, np.ones_like(values)))

)

mean, update_op = metrics.mean(values, weights=weights)

with self.test_session():

variables.local_variables_initializer().run()

self.assertAlmostEqual(expected, update_op.eval(), places=5)

self.assertAlmostEqual(expected, mean.eval(), places=5)

def test1x1x1Weighted(self):

self._test_3d_weighted(

_test_values((3, 2, 4)),

weights=np.asarray((5,)).reshape((1, 1, 1)))

def test1x1xNWeighted(self):

self._test_3d_weighted(

_test_values((3, 2, 4)),

weights=np.asarray((5, 7, 11, 3)).reshape((1, 1, 4)))

def test1xNx1Weighted(self):

self._test_3d_weighted(

_test_values((3, 2, 4)),

weights=np.asarray((5, 11)).reshape((1, 2, 1)))

def test1xNxNWeighted(self):

self._test_3d_weighted(

_test_values((3, 2, 4)),

weights=np.asarray((5, 7, 11, 3, 2, 13, 7, 5)).reshape((1, 2, 4)))

def testNx1x1Weighted(self):

self._test_3d_weighted(

_test_values((3, 2, 4)),

weights=np.asarray((5, 7, 11)).reshape((3, 1, 1)))

def testNx1xNWeighted(self):

self._test_3d_weighted(

_test_values((3, 2, 4)),

weights=np.asarray((

5, 7, 11, 3, 2, 12, 7, 5, 2, 17, 11, 3)).reshape((3, 1, 4)))

def testNxNxNWeighted(self):

self._test_3d_weighted(

_test_values((3, 2, 4)),

weights=np.asarray((

5, 7, 11, 3, 2, 12, 7, 5, 2, 17, 11, 3,

2, 17, 11, 3, 5, 7, 11, 3, 2, 12, 7, 5)).reshape((3, 2, 4)))

def testInvalidWeights(self):

values_placeholder = array_ops.placeholder(dtype=dtypes_lib.float32)

values = _test_values((3, 2, 4, 1))

invalid_weights = (

(1,),

(1, 1),

(3, 2),

(2, 4, 1),

(4, 2, 4, 1),

(3, 3, 4, 1),

(3, 2, 5, 1),

(3, 2, 4, 2),

(1, 1, 1, 1, 1))

expected_error_msg = 'weights can not be broadcast to values'

for invalid_weight in invalid_weights:

# Static shapes.

with self.assertRaisesRegexp(ValueError, expected_error_msg):

metrics.mean(values, invalid_weight)

# Dynamic shapes.

with self.assertRaisesRegexp(errors_impl.OpError, expected_error_msg):

with self.test_session():

_, update_op = metrics.mean(values_placeholder, invalid_weight)

variables.local_variables_initializer().run()

update_op.eval(feed_dict={values_placeholder: values})

class MeanTensorTest(test.TestCase):

def setUp(self):

ops.reset_default_graph()

def testVars(self):

metrics.mean_tensor(array_ops.ones([4, 3]))

_assert_local_variables(self, ('mean/total_tensor:0',

'mean/count_tensor:0'))

def testMetricsCollection(self):

my_collection_name = '__metrics__'

mean, _ = metrics.mean_tensor(

array_ops.ones([4, 3]), metrics_collections=[my_collection_name])

self.assertListEqual(ops.get_collection(my_collection_name), [mean])

def testUpdatesCollection(self):

my_collection_name = '__updates__'

_, update_op = metrics.mean_tensor(

array_ops.ones([4, 3]), updates_collections=[my_collection_name])

self.assertListEqual(ops.get_collection(my_collection_name), [update_op])

def testBasic(self):

with self.test_session() as sess:

values_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 2))

_enqueue_vector(sess, values_queue, [0, 1])

_enqueue_vector(sess, values_queue, [-4.2, 9.1])

_enqueue_vector(sess, values_queue, [6.5, 0])

_enqueue_vector(sess, values_queue, [-3.2, 4.0])

values = values_queue.dequeue()

mean, update_op = metrics.mean_tensor(values)

sess.run(variables.local_variables_initializer())

for _ in range(4):

sess.run(update_op)

self.assertAllClose([[-0.9 / 4., 3.525]], sess.run(mean))

def testMultiDimensional(self):

with self.test_session() as sess:

values_queue = data_flow_ops.FIFOQueue(

2, dtypes=dtypes_lib.float32, shapes=(2, 2, 2))

_enqueue_vector(

sess,

values_queue, [[[1, 2], [1, 2]], [[1, 2], [1, 2]]],

shape=(2, 2, 2))

_enqueue_vector(

sess,

values_queue, [[[1, 2], [1, 2]], [[3, 4], [9, 10]]],

shape=(2, 2, 2))

values = values_queue.dequeue()

mean, update_op = metrics.mean_tensor(values)

sess.run(variables.local_variables_initializer())

for _ in range(2):

sess.run(update_op)

self.assertAllClose([[[1, 2], [1, 2]], [[2, 3], [5, 6]]], sess.run(mean))

def testUpdateOpsReturnsCurrentValue(self):

with self.test_session() as sess:

values_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 2))

_enqueue_vector(sess, values_queue, [0, 1])

_enqueue_vector(sess, values_queue, [-4.2, 9.1])

_enqueue_vector(sess, values_queue, [6.5, 0])

_enqueue_vector(sess, values_queue, [-3.2, 4.0])

values = values_queue.dequeue()

mean, update_op = metrics.mean_tensor(values)

sess.run(variables.local_variables_initializer())

self.assertAllClose([[0, 1]], sess.run(update_op), 5)

self.assertAllClose([[-2.1, 5.05]], sess.run(update_op), 5)

self.assertAllClose([[2.3 / 3., 10.1 / 3.]], sess.run(update_op), 5)

self.assertAllClose([[-0.9 / 4., 3.525]], sess.run(update_op), 5)

self.assertAllClose([[-0.9 / 4., 3.525]], sess.run(mean), 5)

def testWeighted1d(self):

with self.test_session() as sess:

# Create the queue that populates the values.

values_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 2))

_enqueue_vector(sess, values_queue, [0, 1])

_enqueue_vector(sess, values_queue, [-4.2, 9.1])

_enqueue_vector(sess, values_queue, [6.5, 0])

_enqueue_vector(sess, values_queue, [-3.2, 4.0])

values = values_queue.dequeue()

# Create the queue that populates the weights.

weights_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 1))

_enqueue_vector(sess, weights_queue, [[1]])

_enqueue_vector(sess, weights_queue, [[0]])

_enqueue_vector(sess, weights_queue, [[1]])

_enqueue_vector(sess, weights_queue, [[0]])

weights = weights_queue.dequeue()

mean, update_op = metrics.mean_tensor(values, weights)

sess.run(variables.local_variables_initializer())

for _ in range(4):

sess.run(update_op)

self.assertAllClose([[3.25, 0.5]], sess.run(mean), 5)

def testWeighted2d_1(self):

with self.test_session() as sess:

# Create the queue that populates the values.

values_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 2))

_enqueue_vector(sess, values_queue, [0, 1])

_enqueue_vector(sess, values_queue, [-4.2, 9.1])

_enqueue_vector(sess, values_queue, [6.5, 0])

_enqueue_vector(sess, values_queue, [-3.2, 4.0])

values = values_queue.dequeue()

# Create the queue that populates the weights.

weights_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 2))

_enqueue_vector(sess, weights_queue, [1, 1])

_enqueue_vector(sess, weights_queue, [1, 0])

_enqueue_vector(sess, weights_queue, [0, 1])

_enqueue_vector(sess, weights_queue, [0, 0])

weights = weights_queue.dequeue()

mean, update_op = metrics.mean_tensor(values, weights)

sess.run(variables.local_variables_initializer())

for _ in range(4):

sess.run(update_op)

self.assertAllClose([[-2.1, 0.5]], sess.run(mean), 5)

def testWeighted2d_2(self):

with self.test_session() as sess:

# Create the queue that populates the values.

values_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 2))

_enqueue_vector(sess, values_queue, [0, 1])

_enqueue_vector(sess, values_queue, [-4.2, 9.1])

_enqueue_vector(sess, values_queue, [6.5, 0])

_enqueue_vector(sess, values_queue, [-3.2, 4.0])

values = values_queue.dequeue()

# Create the queue that populates the weights.

weights_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 2))

_enqueue_vector(sess, weights_queue, [0, 1])

_enqueue_vector(sess, weights_queue, [0, 0])

_enqueue_vector(sess, weights_queue, [0, 1])

_enqueue_vector(sess, weights_queue, [0, 0])

weights = weights_queue.dequeue()

mean, update_op = metrics.mean_tensor(values, weights)

sess.run(variables.local_variables_initializer())

for _ in range(4):

sess.run(update_op)

self.assertAllClose([[0, 0.5]], sess.run(mean), 5)

class AccuracyTest(test.TestCase):

def setUp(self):

ops.reset_default_graph()

def testVars(self):

metrics.accuracy(

predictions=array_ops.ones((10, 1)),

labels=array_ops.ones((10, 1)),

name='my_accuracy')

_assert_local_variables(self, ('my_accuracy/count:0',

'my_accuracy/total:0'))

def testMetricsCollection(self):

my_collection_name = '__metrics__'

mean, _ = metrics.accuracy(

predictions=array_ops.ones((10, 1)),

labels=array_ops.ones((10, 1)),

metrics_collections=[my_collection_name])

self.assertListEqual(ops.get_collection(my_collection_name), [mean])

def testUpdatesCollection(self):

my_collection_name = '__updates__'

_, update_op = metrics.accuracy(

predictions=array_ops.ones((10, 1)),

labels=array_ops.ones((10, 1)),

updates_collections=[my_collection_name])

self.assertListEqual(ops.get_collection(my_collection_name), [update_op])

def testPredictionsAndLabelsOfDifferentSizeRaisesValueError(self):

predictions = array_ops.ones((10, 3))

labels = array_ops.ones((10, 4))

with self.assertRaises(ValueError):

metrics.accuracy(labels, predictions)

def testPredictionsAndWeightsOfDifferentSizeRaisesValueError(self):

predictions = array_ops.ones((10, 3))

labels = array_ops.ones((10, 3))

weights = array_ops.ones((9, 3))

with self.assertRaises(ValueError):

metrics.accuracy(labels, predictions, weights)

def testValueTensorIsIdempotent(self):

predictions = random_ops.random_uniform(

(10, 3), maxval=3, dtype=dtypes_lib.int64, seed=1)

labels = random_ops.random_uniform(

(10, 3), maxval=3, dtype=dtypes_lib.int64, seed=1)

accuracy, update_op = metrics.accuracy(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

# Run several updates.

for _ in range(10):

sess.run(update_op)

# Then verify idempotency.

initial_accuracy = accuracy.eval()

for _ in range(10):

self.assertEqual(initial_accuracy, accuracy.eval())

def testMultipleUpdates(self):

with self.test_session() as sess:

# Create the queue that populates the predictions.

preds_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 1))

_enqueue_vector(sess, preds_queue, [0])

_enqueue_vector(sess, preds_queue, [1])

_enqueue_vector(sess, preds_queue, [2])

_enqueue_vector(sess, preds_queue, [1])

predictions = preds_queue.dequeue()

# Create the queue that populates the labels.

labels_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 1))

_enqueue_vector(sess, labels_queue, [0])

_enqueue_vector(sess, labels_queue, [1])

_enqueue_vector(sess, labels_queue, [1])

_enqueue_vector(sess, labels_queue, [2])

labels = labels_queue.dequeue()

accuracy, update_op = metrics.accuracy(labels, predictions)

sess.run(variables.local_variables_initializer())

for _ in xrange(3):

sess.run(update_op)

self.assertEqual(0.5, sess.run(update_op))

self.assertEqual(0.5, accuracy.eval())

def testEffectivelyEquivalentSizes(self):

predictions = array_ops.ones((40, 1))

labels = array_ops.ones((40,))

with self.test_session() as sess:

accuracy, update_op = metrics.accuracy(labels, predictions)

sess.run(variables.local_variables_initializer())

self.assertEqual(1.0, update_op.eval())

self.assertEqual(1.0, accuracy.eval())

def testEffectivelyEquivalentSizesWithScalarWeight(self):

predictions = array_ops.ones((40, 1))

labels = array_ops.ones((40,))

with self.test_session() as sess:

accuracy, update_op = metrics.accuracy(labels, predictions, weights=2.0)

sess.run(variables.local_variables_initializer())

self.assertEqual(1.0, update_op.eval())

self.assertEqual(1.0, accuracy.eval())

def testEffectivelyEquivalentSizesWithStaticShapedWeight(self):

predictions = ops.convert_to_tensor([1, 1, 1]) # shape 3,

labels = array_ops.expand_dims(ops.convert_to_tensor([1, 0, 0]),

1) # shape 3, 1

weights = array_ops.expand_dims(ops.convert_to_tensor([100, 1, 1]),

1) # shape 3, 1

with self.test_session() as sess:

accuracy, update_op = metrics.accuracy(labels, predictions, weights)

sess.run(variables.local_variables_initializer())

# if streaming_accuracy does not flatten the weight, accuracy would be

# 0.33333334 due to an intended broadcast of weight. Due to flattening,

# it will be higher than .95

self.assertGreater(update_op.eval(), .95)

self.assertGreater(accuracy.eval(), .95)

def testEffectivelyEquivalentSizesWithDynamicallyShapedWeight(self):

predictions = ops.convert_to_tensor([1, 1, 1]) # shape 3,

labels = array_ops.expand_dims(ops.convert_to_tensor([1, 0, 0]),

1) # shape 3, 1

weights = [[100], [1], [1]] # shape 3, 1

weights_placeholder = array_ops.placeholder(

dtype=dtypes_lib.int32, name='weights')

feed_dict = {weights_placeholder: weights}

with self.test_session() as sess:

accuracy, update_op = metrics.accuracy(labels, predictions,

weights_placeholder)

sess.run(variables.local_variables_initializer())

# if streaming_accuracy does not flatten the weight, accuracy would be

# 0.33333334 due to an intended broadcast of weight. Due to flattening,

# it will be higher than .95

self.assertGreater(update_op.eval(feed_dict=feed_dict), .95)

self.assertGreater(accuracy.eval(feed_dict=feed_dict), .95)

def testMultipleUpdatesWithWeightedValues(self):

with self.test_session() as sess:

# Create the queue that populates the predictions.

preds_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 1))

_enqueue_vector(sess, preds_queue, [0])

_enqueue_vector(sess, preds_queue, [1])

_enqueue_vector(sess, preds_queue, [2])

_enqueue_vector(sess, preds_queue, [1])

predictions = preds_queue.dequeue()

# Create the queue that populates the labels.

labels_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.float32, shapes=(1, 1))

_enqueue_vector(sess, labels_queue, [0])

_enqueue_vector(sess, labels_queue, [1])

_enqueue_vector(sess, labels_queue, [1])

_enqueue_vector(sess, labels_queue, [2])

labels = labels_queue.dequeue()

# Create the queue that populates the weights.

weights_queue = data_flow_ops.FIFOQueue(

4, dtypes=dtypes_lib.int64, shapes=(1, 1))

_enqueue_vector(sess, weights_queue, [1])

_enqueue_vector(sess, weights_queue, [1])

_enqueue_vector(sess, weights_queue, [0])

_enqueue_vector(sess, weights_queue, [0])

weights = weights_queue.dequeue()

accuracy, update_op = metrics.accuracy(labels, predictions, weights)

sess.run(variables.local_variables_initializer())

for _ in xrange(3):

sess.run(update_op)

self.assertEqual(1.0, sess.run(update_op))

self.assertEqual(1.0, accuracy.eval())

class PrecisionTest(test.TestCase):

def setUp(self):

np.random.seed(1)

ops.reset_default_graph()

def testVars(self):

metrics.precision(

predictions=array_ops.ones((10, 1)), labels=array_ops.ones((10, 1)))

_assert_local_variables(self, ('precision/false_positives/count:0',

'precision/true_positives/count:0'))

def testMetricsCollection(self):

my_collection_name = '__metrics__'

mean, _ = metrics.precision(

predictions=array_ops.ones((10, 1)),

labels=array_ops.ones((10, 1)),

metrics_collections=[my_collection_name])

self.assertListEqual(ops.get_collection(my_collection_name), [mean])

def testUpdatesCollection(self):

my_collection_name = '__updates__'

_, update_op = metrics.precision(

predictions=array_ops.ones((10, 1)),

labels=array_ops.ones((10, 1)),

updates_collections=[my_collection_name])

self.assertListEqual(ops.get_collection(my_collection_name), [update_op])

def testValueTensorIsIdempotent(self):

predictions = random_ops.random_uniform(

(10, 3), maxval=1, dtype=dtypes_lib.int64, seed=1)

labels = random_ops.random_uniform(

(10, 3), maxval=1, dtype=dtypes_lib.int64, seed=1)

precision, update_op = metrics.precision(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

# Run several updates.

for _ in range(10):

sess.run(update_op)

# Then verify idempotency.

initial_precision = precision.eval()

for _ in range(10):

self.assertEqual(initial_precision, precision.eval())

def testAllCorrect(self):

inputs = np.random.randint(0, 2, size=(100, 1))

predictions = constant_op.constant(inputs)

labels = constant_op.constant(inputs)

precision, update_op = metrics.precision(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

self.assertAlmostEqual(1, sess.run(update_op))

self.assertAlmostEqual(1, precision.eval())

def testSomeCorrect_multipleInputDtypes(self):

for dtype in (dtypes_lib.bool, dtypes_lib.int32, dtypes_lib.float32):

predictions = math_ops.cast(

constant_op.constant([1, 0, 1, 0], shape=(1, 4)), dtype=dtype)

labels = math_ops.cast(

constant_op.constant([0, 1, 1, 0], shape=(1, 4)), dtype=dtype)

precision, update_op = metrics.precision(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

self.assertAlmostEqual(0.5, update_op.eval())

self.assertAlmostEqual(0.5, precision.eval())

def testWeighted1d(self):

predictions = constant_op.constant([[1, 0, 1, 0], [1, 0, 1, 0]])

labels = constant_op.constant([[0, 1, 1, 0], [1, 0, 0, 1]])

precision, update_op = metrics.precision(

labels, predictions, weights=constant_op.constant([[2], [5]]))

with self.test_session():

variables.local_variables_initializer().run()

weighted_tp = 2.0 + 5.0

weighted_positives = (2.0 + 2.0) + (5.0 + 5.0)

expected_precision = weighted_tp / weighted_positives

self.assertAlmostEqual(expected_precision, update_op.eval())

self.assertAlmostEqual(expected_precision, precision.eval())

def testWeightedScalar_placeholders(self):

predictions = array_ops.placeholder(dtype=dtypes_lib.float32)

labels = array_ops.placeholder(dtype=dtypes_lib.float32)

feed_dict = {

predictions: ((1, 0, 1, 0), (1, 0, 1, 0)),

labels: ((0, 1, 1, 0), (1, 0, 0, 1))

}

precision, update_op = metrics.precision(labels, predictions, weights=2)

with self.test_session():

variables.local_variables_initializer().run()

weighted_tp = 2.0 + 2.0

weighted_positives = (2.0 + 2.0) + (2.0 + 2.0)

expected_precision = weighted_tp / weighted_positives

self.assertAlmostEqual(

expected_precision, update_op.eval(feed_dict=feed_dict))

self.assertAlmostEqual(

expected_precision, precision.eval(feed_dict=feed_dict))

def testWeighted1d_placeholders(self):

predictions = array_ops.placeholder(dtype=dtypes_lib.float32)

labels = array_ops.placeholder(dtype=dtypes_lib.float32)

feed_dict = {

predictions: ((1, 0, 1, 0), (1, 0, 1, 0)),

labels: ((0, 1, 1, 0), (1, 0, 0, 1))

}

precision, update_op = metrics.precision(

labels, predictions, weights=constant_op.constant([[2], [5]]))

with self.test_session():

variables.local_variables_initializer().run()

weighted_tp = 2.0 + 5.0

weighted_positives = (2.0 + 2.0) + (5.0 + 5.0)

expected_precision = weighted_tp / weighted_positives

self.assertAlmostEqual(

expected_precision, update_op.eval(feed_dict=feed_dict))

self.assertAlmostEqual(

expected_precision, precision.eval(feed_dict=feed_dict))

def testWeighted2d(self):

predictions = constant_op.constant([[1, 0, 1, 0], [1, 0, 1, 0]])

labels = constant_op.constant([[0, 1, 1, 0], [1, 0, 0, 1]])

precision, update_op = metrics.precision(

labels,

predictions,

weights=constant_op.constant([[1, 2, 3, 4], [4, 3, 2, 1]]))

with self.test_session():

variables.local_variables_initializer().run()

weighted_tp = 3.0 + 4.0

weighted_positives = (1.0 + 3.0) + (4.0 + 2.0)

expected_precision = weighted_tp / weighted_positives

self.assertAlmostEqual(expected_precision, update_op.eval())

self.assertAlmostEqual(expected_precision, precision.eval())

def testWeighted2d_placeholders(self):

predictions = array_ops.placeholder(dtype=dtypes_lib.float32)

labels = array_ops.placeholder(dtype=dtypes_lib.float32)

feed_dict = {

predictions: ((1, 0, 1, 0), (1, 0, 1, 0)),

labels: ((0, 1, 1, 0), (1, 0, 0, 1))

}

precision, update_op = metrics.precision(

labels,

predictions,

weights=constant_op.constant([[1, 2, 3, 4], [4, 3, 2, 1]]))

with self.test_session():

variables.local_variables_initializer().run()

weighted_tp = 3.0 + 4.0

weighted_positives = (1.0 + 3.0) + (4.0 + 2.0)

expected_precision = weighted_tp / weighted_positives

self.assertAlmostEqual(

expected_precision, update_op.eval(feed_dict=feed_dict))

self.assertAlmostEqual(

expected_precision, precision.eval(feed_dict=feed_dict))

def testAllIncorrect(self):

inputs = np.random.randint(0, 2, size=(100, 1))

predictions = constant_op.constant(inputs)

labels = constant_op.constant(1 - inputs)

precision, update_op = metrics.precision(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

sess.run(update_op)

self.assertAlmostEqual(0, precision.eval())

def testZeroTrueAndFalsePositivesGivesZeroPrecision(self):

predictions = constant_op.constant([0, 0, 0, 0])

labels = constant_op.constant([0, 0, 0, 0])

precision, update_op = metrics.precision(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

sess.run(update_op)

self.assertEqual(0.0, precision.eval())

class RecallTest(test.TestCase):

def setUp(self):

np.random.seed(1)

ops.reset_default_graph()

def testVars(self):

metrics.recall(

predictions=array_ops.ones((10, 1)), labels=array_ops.ones((10, 1)))

_assert_local_variables(self, ('recall/false_negatives/count:0',

'recall/true_positives/count:0'))

def testMetricsCollection(self):

my_collection_name = '__metrics__'

mean, _ = metrics.recall(

predictions=array_ops.ones((10, 1)),

labels=array_ops.ones((10, 1)),

metrics_collections=[my_collection_name])

self.assertListEqual(ops.get_collection(my_collection_name), [mean])

def testUpdatesCollection(self):

my_collection_name = '__updates__'

_, update_op = metrics.recall(

predictions=array_ops.ones((10, 1)),

labels=array_ops.ones((10, 1)),

updates_collections=[my_collection_name])

self.assertListEqual(ops.get_collection(my_collection_name), [update_op])

def testValueTensorIsIdempotent(self):

predictions = random_ops.random_uniform(

(10, 3), maxval=1, dtype=dtypes_lib.int64, seed=1)

labels = random_ops.random_uniform(

(10, 3), maxval=1, dtype=dtypes_lib.int64, seed=1)

recall, update_op = metrics.recall(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

# Run several updates.

for _ in range(10):

sess.run(update_op)

# Then verify idempotency.

initial_recall = recall.eval()

for _ in range(10):

self.assertEqual(initial_recall, recall.eval())

def testAllCorrect(self):

np_inputs = np.random.randint(0, 2, size=(100, 1))

predictions = constant_op.constant(np_inputs)

labels = constant_op.constant(np_inputs)

recall, update_op = metrics.recall(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

sess.run(update_op)

self.assertEqual(1, recall.eval())

def testSomeCorrect_multipleInputDtypes(self):

for dtype in (dtypes_lib.bool, dtypes_lib.int32, dtypes_lib.float32):

predictions = math_ops.cast(

constant_op.constant([1, 0, 1, 0], shape=(1, 4)), dtype=dtype)

labels = math_ops.cast(

constant_op.constant([0, 1, 1, 0], shape=(1, 4)), dtype=dtype)

recall, update_op = metrics.recall(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

self.assertAlmostEqual(0.5, update_op.eval())

self.assertAlmostEqual(0.5, recall.eval())

def testWeighted1d(self):

predictions = constant_op.constant([[1, 0, 1, 0], [0, 1, 0, 1]])

labels = constant_op.constant([[0, 1, 1, 0], [1, 0, 0, 1]])

weights = constant_op.constant([[2], [5]])

recall, update_op = metrics.recall(labels, predictions, weights=weights)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

weighted_tp = 2.0 + 5.0

weighted_t = (2.0 + 2.0) + (5.0 + 5.0)

expected_precision = weighted_tp / weighted_t

self.assertAlmostEqual(expected_precision, update_op.eval())

self.assertAlmostEqual(expected_precision, recall.eval())

def testWeighted2d(self):

predictions = constant_op.constant([[1, 0, 1, 0], [0, 1, 0, 1]])

labels = constant_op.constant([[0, 1, 1, 0], [1, 0, 0, 1]])

weights = constant_op.constant([[1, 2, 3, 4], [4, 3, 2, 1]])

recall, update_op = metrics.recall(labels, predictions, weights=weights)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

weighted_tp = 3.0 + 1.0

weighted_t = (2.0 + 3.0) + (4.0 + 1.0)

expected_precision = weighted_tp / weighted_t

self.assertAlmostEqual(expected_precision, update_op.eval())

self.assertAlmostEqual(expected_precision, recall.eval())

def testAllIncorrect(self):

np_inputs = np.random.randint(0, 2, size=(100, 1))

predictions = constant_op.constant(np_inputs)

labels = constant_op.constant(1 - np_inputs)

recall, update_op = metrics.recall(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

sess.run(update_op)

self.assertEqual(0, recall.eval())

def testZeroTruePositivesAndFalseNegativesGivesZeroRecall(self):

predictions = array_ops.zeros((1, 4))

labels = array_ops.zeros((1, 4))

recall, update_op = metrics.recall(labels, predictions)

with self.test_session() as sess:

sess.run(variables.local_variables_initializer())

sess.run(update_op)

self.assertEqual(0, recall.eval())

class AUCTest(test.TestCase):

def setUp(self):

np.random.seed(1)

ops.reset_default_graph()

def testVars(self):

metrics.auc(predictions=array_ops.ones((10, 1)),