# Copyright 2016 the GPflow authors.
#
# 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.
import warnings
import tensorflow as tf
import numpy as np
from numpy.testing import assert_allclose, assert_equal
import gpflow
from gpflow.transforms import Chain, Identity
from gpflow.test_util import GPflowTestCase
from gpflow import settings
class TransformTests(GPflowTestCase):
def prepare(self):
x_np = np.random.randn(10).astype(settings.float_type)
transforms = []
for transform_class in gpflow.transforms.Transform.__subclasses__():
if transform_class == Chain:
continue # Chain transform cannot be tested on its own
if transform_class == gpflow.transforms.LowerTriangular:
pass # Test triangular transforms separately.
elif transform_class == gpflow.transforms.DiagMatrix:
transforms.append(transform_class(5))
else:
transform = transform_class()
transforms.append(transform)
transforms.append(Chain(Identity(), transform))
transforms.append(Chain(transform, Identity()))
transforms.append(gpflow.transforms.Logistic(7.3, 19.4))
# test __call__() and chaining:
transforms.append(gpflow.transforms.positive(gpflow.transforms.Rescale(7.5)))
transforms.append(gpflow.transforms.Rescale(9.5)(gpflow.transforms.positive))
# test helper:
transforms.append(gpflow.transforms.positiveRescale(9.5))
return tf.convert_to_tensor(x_np), x_np, transforms
def test_tf_np_forward_backward(self):
"""
Make sure the np forward transforms are the same as the tensorflow ones
"""
with self.test_context() as session:
x, x_np, transforms = self.prepare()
for t in transforms:
y_tf = t.forward_tensor(x)
y_np = t.forward(x_np)
assert_allclose(session.run(y_tf), y_np)
def test_forward_backward(self):
with self.test_context() as session:
x, x_np, transforms = self.prepare()
for t in transforms:
y_np_res = t.forward(x_np)
y_tf = t.forward_tensor(x)
y_tf_res = session.run(y_tf)
assert_allclose(y_np_res, y_tf_res)
x_np_res = t.backward(y_np_res)
x_tf = t.backward_tensor(y_tf)
x_tf_res = session.run(x_tf)
assert_allclose(x_np_res, x_tf_res)
x_expect = x_np.reshape(x_np_res.shape)
assert_allclose(x_expect, x_np_res)
assert_allclose(x_expect, x_tf_res)
def test_logjac(self):
"""
We have hand-crafted the log-jacobians for speed. Check they're correct
wrt a tensorflow derived version
"""
with self.test_context() as session:
x, x_np, transforms = self.prepare()
# there is no jacobian: loop manually
def jacobian(f):
return tf.stack([tf.gradients(f(x)[i], x)[0] for i in range(10)])
tf_jacs = [tf.log(tf.matrix_determinant(jacobian(t.forward_tensor)))
for t in transforms
if not isinstance(t, (gpflow.transforms.LowerTriangular,
gpflow.transforms.DiagMatrix))]
hand_jacs = [t.log_jacobian_tensor(x)
for t in transforms
if not isinstance(t, (gpflow.transforms.LowerTriangular,
gpflow.transforms.DiagMatrix))]
for j1, j2 in zip(tf_jacs, hand_jacs):
j1_res = session.run(j1)
j2_res = session.run(j2)
assert_allclose(j1_res, j2_res)
def test_logistic_error_wrong_order(self):
with self.assertRaises(ValueError):
gpflow.transforms.Logistic(8.0, 4.7)
def test_logistic_error_bounds_equal(self):
with self.assertRaises(ValueError):
gpflow.transforms.Logistic(4.7, 4.7)
def test_bad_chain_argument(self):
t = gpflow.transforms.Logistic(1.0, 2.0)
with self.assertRaises(TypeError):
t(1.5) # this syntax chains transforms, is not equivalent to t.forward(x)
class TestChainIdentity(GPflowTestCase):
def prepare(self):
x_np = np.random.randn(10).astype(settings.float_type)
transforms = []
for transform in gpflow.transforms.Transform.__subclasses__():
if transform != Chain and transform != gpflow.transforms.LowerTriangular:
transforms.append(transform())
transforms.append(gpflow.transforms.Logistic(7.3, 19.4))
return tf.convert_to_tensor(x_np), x_np, transforms
def assertEqualElements(self, lst):
elem0 = lst[0]
for elemi in lst[1:]:
assert_equal(elem0, elemi)
def test_equivalence(self):
"""
Make sure chaining with identity doesn't lead to different values.
"""
with self.test_context() as session:
x, x_np, transforms = self.prepare()
for transform in transforms:
equiv_transforms = [transform,
Chain(transform, Identity()),
Chain(Identity(), transform)]
ys_np = [t.forward(x_np) for t in equiv_transforms]
self.assertEqualElements(ys_np)
y_np = ys_np[0]
xs_np = [t.backward(y_np) for t in equiv_transforms]
self.assertEqualElements(xs_np)
ys = [t.forward_tensor(x) for t in equiv_transforms]
ys_tf = [session.run(y) for y in ys]
self.assertEqualElements(ys_tf)
logjs = [t.log_jacobian_tensor(x) for t in equiv_transforms]
logjs_tf = [session.run(logj) for logj in logjs]
self.assertEqualElements(logjs_tf)
class TestOverflow(GPflowTestCase):
"""
Bug #302 identified an overflow in the standard positive transform. This is a regression test.
"""
def setUp(self):
self.t = gpflow.transforms.Log1pe()
def testOverflow(self):
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter('always')
y = self.t.forward(np.array([-1000, -300, -10, 10, 300, 1000]))
self.assertEqual(len(w), 0)
self.assertFalse(np.any(np.isinf(y)))
self.assertFalse(np.any(np.isnan(y)))
def testDivByZero(self):
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter('always')
y = self.t.backward(np.array([self.t._lower]))
self.assertTrue(len(w) == 0)
self.assertFalse(np.any(np.isinf(y)))
self.assertFalse(np.any(np.isnan(y)))
class TestMatrixTransforms(GPflowTestCase):
"""
Some extra tests for the matrix transformations.
"""
def test_LowerTriangular(self):
t = gpflow.transforms.LowerTriangular(N=3, num_matrices=2)
t.forward(np.ones((2, 6)))
with self.assertRaises(ValueError):
t.forward(np.ones((2, 7)))
def test_LowerTriangular_squeezes_only_first_axis(self):
t = gpflow.transforms.LowerTriangular(1, 1, squeeze=True)
ret = t.forward(np.ones((1, 1)))
self.assertEqual(ret.shape, (1, 1))
def test_LowerTriangularConsistency(self):
t = gpflow.transforms.LowerTriangular(2, 4)
M = np.random.randn(4, 2, 2) * np.tri(2, 2)[None, :, :]
M_free = t.backward(M)
assert_allclose(M, t.forward(M_free))
with self.test_context() as session:
M_free_tf = session.run(t.backward_tensor(tf.identity(M)))
assert_allclose(M_free, M_free_tf)
assert_allclose(M, session.run(t.forward_tensor(tf.identity(M_free_tf))))
def test_LowerTriangular_squeezes(self):
t1 = gpflow.transforms.LowerTriangular(N=3, num_matrices=1)
t2 = gpflow.transforms.LowerTriangular(N=3, num_matrices=1, squeeze=True)
X = np.random.randn(1, 6)
Y = np.random.randn(1, 3, 3)
self.assertTrue(np.all(t1.forward(X).squeeze() == t2.forward(X)))
self.assertTrue(np.all(t1.forward(X).squeeze().shape == t2.forward(X).shape))
self.assertTrue(np.all(t1.backward(Y) == t2.backward(Y.squeeze())))
self.assertTrue(np.all(t1.backward(Y).shape == t2.backward(Y.squeeze()).shape))
with self.test_context() as session:
self.assertTrue(np.all(session.run(tf.squeeze(t1.forward_tensor(X))) ==
session.run(t2.forward_tensor(X))))
self.assertTrue(np.all(session.run(t1.backward_tensor(Y)) ==
session.run(t2.backward_tensor(Y.squeeze()))))
# make sure we don't try to squeeze when there are more than 1 matrices.
with self.assertRaises(ValueError):
gpflow.transforms.LowerTriangular(N=3, num_matrices=2, squeeze=True)
def test_DiagMatrix(self):
t = gpflow.transforms.DiagMatrix(3)
t.backward(np.eye(3))
t.backward(np.eye(3)[None, :, :])
t.backward(np.eye(3)[None, :, :] * np.array([1, 2])[:, None, None])
with self.assertRaises(ValueError):
t.backward(np.eye(4))
t.backward(np.eye(2)[None, :, :] * np.array([1, 2, 3])[:, None, None])
class TestDiagMatrixTransform(GPflowTestCase):
def setUp(self):
self.t1 = gpflow.transforms.DiagMatrix(dim=1)
self.t2 = gpflow.transforms.DiagMatrix(dim=3)
def test_forward_backward(self):
free_1d = np.random.randn(8)
fwd1d = self.t1.forward(free_1d)
assert_allclose(fwd1d.shape, np.array([len(free_1d), self.t1.dim, self.t1.dim]))
assert_allclose(free_1d, self.t1.backward(fwd1d))
size2d = 5
free_2d = np.random.randn(size2d, self.t2.dim).flatten()
fwd2d = self.t2.forward(free_2d)
assert_allclose(fwd2d.shape, np.array([size2d, self.t2.dim, self.t2.dim]))
assert_allclose(free_2d, self.t2.backward(fwd2d))
def test_tf_np_forward(self):
"""
Make sure the np forward transforms are the same as the tensorflow ones
"""
with self.test_context() as session:
free = np.random.randn(8, self.t2.dim)
x = tf.convert_to_tensor(free)
ys = session.run(self.t2.forward_tensor(x))
assert_allclose(ys, self.t2.forward(free))
free = np.random.randn(7, self.t1.dim)
x = tf.convert_to_tensor(free)
ys = session.run(self.t1.forward_tensor(x))
assert_allclose(ys, self.t1.forward(free))
if __name__ == "__main__":
tf.test.main()
