# Copyright (c) 2022 PaddlePaddle 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. import typing import unittest import numpy as np import parameterize as param from distribution import config import paddle from paddle.distribution import transform, variable np.random.seed(2022) paddle.seed(2022) paddle.enable_static() @param.place(config.DEVICES) class TestTransform(unittest.TestCase): def setUp(self): self._t = transform.Transform() @param.param_func( [ (transform.Type.BIJECTION, True), (transform.Type.INJECTION, True), (transform.Type.SURJECTION, False), (transform.Type.OTHER, False), ] ) def test_is_injective(self, type, expected): transform.Transform._type = type self.assertEqual(self._t._is_injective(), expected) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Real)) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Real)) @param.param_func( [ (np.array(0), NotImplementedError), (np.random.random((2, 3)), NotImplementedError), ] ) def test_forward(self, input, expected): with self.assertRaises(expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.Transform() static_input = paddle.static.data( 'input', input.shape, input.dtype ) output = t.forward(static_input) exe.run(sp) exe.run(mp, feed={'input': input}, fetch_list=[output]) @param.param_func( [ (np.array(0), NotImplementedError), (np.random.random((2, 3)), NotImplementedError), ] ) def test_inverse(self, input, expected): with self.assertRaises(expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.Transform() static_input = paddle.static.data( 'input', input.shape, input.dtype ) output = t.inverse(static_input) exe.run(sp) exe.run(mp, feed={'input': input}, fetch_list=[output]) @param.param_func( [ (np.array(0), NotImplementedError), (paddle.rand((2, 3)), NotImplementedError), ] ) def test_forward_log_det_jacobian(self, input, expected): with self.assertRaises(expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.Transform() static_input = paddle.static.data( 'input', input.shape, input.dtype ) output = t.forward_log_det_jacobian(static_input) exe.run(sp) exe.run(mp, feed={'input': input}, fetch_list=[output]) @param.param_func( [ (np.array(0), NotImplementedError), (paddle.rand((2, 3)), NotImplementedError), ] ) def test_inverse_log_det_jacobian(self, input, expected): with self.assertRaises(expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.Transform() static_input = paddle.static.data( 'input', input.shape, input.dtype ) output = t.inverse_log_det_jacobian(static_input) exe.run(sp) exe.run(mp, feed={'input': input}, fetch_list=[output]) @param.param_func([(0, TypeError)]) def test_forward_shape(self, shape, expected): with self.assertRaises(expected): self._t.forward_shape(shape) @param.param_func([(0, TypeError)]) def test_inverse_shape(self, shape, expected): with self.assertRaises(expected): self._t.forward_shape(shape) @param.place(config.DEVICES) class TestAbsTransform(unittest.TestCase): def setUp(self): self._t = transform.AbsTransform() def test_is_injective(self): self.assertFalse(self._t._is_injective()) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Real)) self.assertEqual(self._t._domain.event_rank, 0) self.assertEqual(self._t._domain.is_discrete, False) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Positive)) self.assertEqual(self._t._codomain.event_rank, 0) self.assertEqual(self._t._codomain.is_discrete, False) @param.param_func( [ (np.array([-1.0, 1.0, 0.0]), np.array([1.0, 1.0, 0.0])), ( np.array([[1.0, -1.0, -0.1], [-3.0, -0.1, 0]]), np.array([[1.0, 1.0, 0.1], [3.0, 0.1, 0]]), ), ] ) def test_forward(self, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.AbsTransform() static_input = paddle.static.data('input', input.shape, input.dtype) output = t.forward(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func([(np.array([1.0]), (-np.array([1.0]), np.array([1.0])))]) def test_inverse(self, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.AbsTransform() static_input = paddle.static.data('input', input.shape, input.dtype) actual0, actual1 = t.inverse(static_input) exe.run(sp) [actual0, actual1] = exe.run( mp, feed={'input': input}, fetch_list=[actual0, actual1] ) expected0, expected1 = expected np.testing.assert_allclose( actual0, expected0, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) np.testing.assert_allclose( actual1, expected1, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) def test_forward_log_det_jacobian(self): input = np.random.random((10,)) with self.assertRaises(NotImplementedError): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.AbsTransform() static_input = paddle.static.data( 'input', input.shape, input.dtype ) output = t.forward_log_det_jacobian(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) @param.param_func( [ (np.array([1.0]), (np.array([0.0]), np.array([0.0]))), ] ) def test_inverse_log_det_jacobian(self, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.AbsTransform() static_input = paddle.static.data('input', input.shape, input.dtype) actual0, actual1 = t.inverse_log_det_jacobian(static_input) exe.run(sp) [actual0, actual1] = exe.run( mp, feed={'input': input}, fetch_list=[actual0, actual1] ) expected0, expected1 = expected np.testing.assert_allclose( actual0, expected0, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) np.testing.assert_allclose( actual1, expected1, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_forward_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_inverse_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.place(config.DEVICES) @param.param_cls( (param.TEST_CASE_NAME, 'loc', 'scale'), [ ('normal', np.random.rand(8, 10), np.random.rand(8, 10)), ('broadcast', np.random.rand(2, 10), np.random.rand(10)), ], ) class TestAffineTransform(unittest.TestCase): def setUp(self): self.sp = paddle.static.Program() self.mp = paddle.static.Program() with paddle.static.program_guard(self.mp, self.sp): loc = paddle.static.data('loc', self.loc.shape, self.loc.dtype) scale = paddle.static.data( 'scale', self.scale.shape, self.scale.dtype ) self._t = transform.AffineTransform(loc, scale) def test_is_injective(self): self.assertTrue(self._t._is_injective()) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Real)) self.assertEqual(self._t._domain.event_rank, 0) self.assertEqual(self._t._domain.is_discrete, False) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Real)) self.assertEqual(self._t._codomain.event_rank, 0) self.assertEqual(self._t._codomain.is_discrete, False) def test_forward(self): input = np.random.random(self.loc.shape) exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): loc = paddle.static.data('loc', self.loc.shape, self.loc.dtype) scale = paddle.static.data( 'scale', self.scale.shape, self.scale.dtype ) t = transform.AffineTransform(loc, scale) static_input = paddle.static.data( 'input', self.loc.shape, self.loc.dtype ) output = t.forward(static_input) exe.run(sp) [output] = exe.run( mp, feed={'input': input, 'loc': self.loc, 'scale': self.scale}, fetch_list=[output], ) np.testing.assert_allclose( output, self._np_forward(input), rtol=config.RTOL.get(str(self.loc.dtype)), atol=config.ATOL.get(str(self.loc.dtype)), ) def test_inverse(self): input = np.random.random(self.loc.shape) exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): loc = paddle.static.data('loc', self.loc.shape, self.loc.dtype) scale = paddle.static.data( 'scale', self.scale.shape, self.scale.dtype ) t = transform.AffineTransform(loc, scale) static_input = paddle.static.data( 'input', self.loc.shape, self.loc.dtype ) output = t.inverse(static_input) exe.run(sp) [output] = exe.run( mp, feed={'input': input, 'loc': self.loc, 'scale': self.scale}, fetch_list=[output], ) np.testing.assert_allclose( output, self._np_inverse(input), rtol=config.RTOL.get(str(self.loc.dtype)), atol=config.ATOL.get(str(self.loc.dtype)), ) def _np_forward(self, x): return self.loc + self.scale * x def _np_inverse(self, y): return (y - self.loc) / self.scale def _np_forward_jacobian(self, x): return np.log(np.abs(self.scale)) def _np_inverse_jacobian(self, y): return -self._np_forward_jacobian(self._np_inverse(y)) def test_inverse_log_det_jacobian(self): input = np.random.random(self.scale.shape) exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): loc = paddle.static.data('loc', self.loc.shape, self.loc.dtype) scale = paddle.static.data( 'scale', self.scale.shape, self.scale.dtype ) t = transform.AffineTransform(loc, scale) static_input = paddle.static.data('input', input.shape, input.dtype) output = t.inverse_log_det_jacobian(static_input) exe.run(sp) [output] = exe.run( mp, feed={'input': input, 'loc': self.loc, 'scale': self.scale}, fetch_list=[output], ) np.testing.assert_allclose( output, self._np_inverse_jacobian(input), rtol=config.RTOL.get(str(self.loc.dtype)), atol=config.ATOL.get(str(self.loc.dtype)), ) def test_forward_log_det_jacobian(self): input = np.random.random(self.scale.shape) exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): loc = paddle.static.data('loc', self.loc.shape, self.loc.dtype) scale = paddle.static.data( 'scale', self.scale.shape, self.scale.dtype ) t = transform.AffineTransform(loc, scale) static_input = paddle.static.data('input', input.shape, input.dtype) output = t.forward_log_det_jacobian(static_input) exe.run(sp) [output] = exe.run( mp, feed={'input': input, 'loc': self.loc, 'scale': self.scale}, fetch_list=[output], ) np.testing.assert_allclose( output, self._np_forward_jacobian(input), rtol=config.RTOL.get(str(self.loc.dtype)), atol=config.ATOL.get(str(self.loc.dtype)), ) def test_forward_shape(self): shape = self.loc.shape self.assertEqual( tuple(self._t.forward_shape(shape)), np.broadcast(np.random.random(shape), self.loc, self.scale).shape, ) def test_inverse_shape(self): shape = self.scale.shape self.assertEqual( tuple(self._t.forward_shape(shape)), np.broadcast(np.random.random(shape), self.loc, self.scale).shape, ) @param.place(config.DEVICES) class TestExpTransform(unittest.TestCase): def setUp(self): self._t = transform.ExpTransform() def test_is_injective(self): self.assertTrue(self._t._is_injective()) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Real)) self.assertEqual(self._t._domain.event_rank, 0) self.assertEqual(self._t._domain.is_discrete, False) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Positive)) self.assertEqual(self._t._codomain.event_rank, 0) self.assertEqual(self._t._codomain.is_discrete, False) @param.param_func( [ ( np.array([0.0, 1.0, 2.0, 3.0]), np.exp(np.array([0.0, 1.0, 2.0, 3.0])), ), ( np.array([[0.0, 1.0, 2.0, 3.0], [-5.0, 6.0, 7.0, 8.0]]), np.exp(np.array([[0.0, 1.0, 2.0, 3.0], [-5.0, 6.0, 7.0, 8.0]])), ), ] ) def test_forward(self, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.ExpTransform() static_input = paddle.static.data('input', input.shape, input.dtype) output = t.forward(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func( [ (np.array([1.0, 2.0, 3.0]), np.log(np.array([1.0, 2.0, 3.0]))), ( np.array([[1.0, 2.0, 3.0], [6.0, 7.0, 8.0]]), np.log(np.array([[1.0, 2.0, 3.0], [6.0, 7.0, 8.0]])), ), ] ) def test_inverse(self, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.ExpTransform() static_input = paddle.static.data('input', input.shape, input.dtype) output = t.inverse(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func( [ (np.array([1.0, 2.0, 3.0]),), (np.array([[1.0, 2.0, 3.0], [6.0, 7.0, 8.0]]),), ] ) def test_forward_log_det_jacobian(self, input): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.ExpTransform() static_input = paddle.static.data('input', input.shape, input.dtype) output = t.forward_log_det_jacobian(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, self._np_forward_jacobian(input), rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) def _np_forward_jacobian(self, x): return x @param.param_func( [ (np.array([1.0, 2.0, 3.0]),), (np.array([[1.0, 2.0, 3.0], [6.0, 7.0, 8.0]]),), ] ) def test_inverse_log_det_jacobian(self, input): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.ExpTransform() static_input = paddle.static.data('input', input.shape, input.dtype) output = t.inverse_log_det_jacobian(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, self._np_inverse_jacobian(input), rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) def _np_inverse_jacobian(self, y): return -self._np_forward_jacobian(np.log(y)) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_forward_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_inverse_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.place(config.DEVICES) class TestChainTransform(unittest.TestCase): @param.param_func( ( ( transform.ChainTransform( ( transform.AbsTransform(), transform.AffineTransform( paddle.rand([1]), paddle.rand([1]) ), ) ), False, ), ( transform.ChainTransform( ( transform.AffineTransform( paddle.rand([1]), paddle.rand([1]) ), transform.ExpTransform(), ) ), True, ), ) ) def test_is_injective(self, chain, expected): self.assertEqual(chain._is_injective(), expected) @param.param_func( ( ( transform.ChainTransform( ( transform.IndependentTransform( transform.ExpTransform(), 1 ), transform.IndependentTransform( transform.ExpTransform(), 10 ), transform.IndependentTransform( transform.ExpTransform(), 8 ), ) ), variable.Independent(variable.real, 10), ), ) ) def test_domain(self, input, expected): self.assertIsInstance(input._domain, type(expected)) self.assertEqual(input._domain.event_rank, expected.event_rank) self.assertEqual(input._domain.is_discrete, expected.is_discrete) @param.param_func( ( ( transform.ChainTransform( ( transform.IndependentTransform( transform.ExpTransform(), 9 ), transform.IndependentTransform( transform.ExpTransform(), 4 ), transform.IndependentTransform( transform.ExpTransform(), 5 ), ) ), variable.Independent(variable.real, 9), ), ) ) def test_codomain(self, input, expected): self.assertIsInstance(input._codomain, variable.Independent) self.assertEqual(input._codomain.event_rank, expected.event_rank) self.assertEqual(input._codomain.is_discrete, expected.is_discrete) @param.param_func( [ ( transform.ChainTransform( (transform.ExpTransform(), transform.TanhTransform()) ), np.array([[0.0, -1.0, 2.0, -3.0], [-5.0, 6.0, 7.0, -8.0]]), np.tanh( np.exp( np.array( [[0.0, -1.0, 2.0, -3.0], [-5.0, 6.0, 7.0, -8.0]] ) ) ), ) ] ) def test_forward(self, chain, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = chain static_input = paddle.static.data('input', input.shape, input.dtype) output = t.forward(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func( [ ( transform.ChainTransform( (transform.ExpTransform(), transform.TanhTransform()) ), np.array([[0.0, 1.0, 2.0, 3.0], [5.0, 6.0, 7.0, 8.0]]), np.log( np.arctanh( np.array([[0.0, 1.0, 2.0, 3.0], [5.0, 6.0, 7.0, 8.0]]) ) ), ) ] ) def test_inverse(self, chain, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = chain static_input = paddle.static.data('input', input.shape, input.dtype) output = t.inverse(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func( [ ( transform.ChainTransform( ( transform.AffineTransform( paddle.full([1], 0.0), paddle.full([1], -1.0) ), transform.ExpTransform(), ) ), (2, 3, 5), (2, 3, 5), ), ] ) def test_forward_shape(self, chain, shape, expected_shape): self.assertEqual(chain.forward_shape(shape), expected_shape) @param.param_func( [ ( transform.ChainTransform( ( transform.AffineTransform( paddle.full([1], 0.0), paddle.full([1], -1.0) ), transform.ExpTransform(), ) ), (2, 3, 5), (2, 3, 5), ), ] ) def test_inverse_shape(self, chain, shape, expected_shape): self.assertEqual(chain.forward_shape(shape), expected_shape) @param.place(config.DEVICES) @param.param_cls( (param.TEST_CASE_NAME, 'base', 'reinterpreted_batch_rank', 'x'), [ ( 'rank-over-zero', transform.ExpTransform(), 2, np.random.rand(2, 3, 3), ), ], ) class TestIndependentTransform(unittest.TestCase): def setUp(self): self._t = transform.IndependentTransform( self.base, self.reinterpreted_batch_rank ) def test_is_injective(self): self.assertEqual(self._t._is_injective(), self.base._is_injective()) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Independent)) self.assertEqual( self._t._domain.event_rank, self.base._domain.event_rank + self.reinterpreted_batch_rank, ) self.assertEqual( self._t._domain.is_discrete, self.base._domain.is_discrete ) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Independent)) self.assertEqual( self._t._codomain.event_rank, self.base._codomain.event_rank + self.reinterpreted_batch_rank, ) self.assertEqual( self._t._codomain.is_discrete, self.base._codomain.is_discrete ) def test_forward(self): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.IndependentTransform( self.base, self.reinterpreted_batch_rank ) static_input = paddle.static.data( 'input', self.x.shape, self.x.dtype ) output = t.forward(static_input) expected = self.base.forward(static_input) exe.run(sp) [output, expected] = exe.run( mp, feed={'input': self.x}, fetch_list=[output, expected] ) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(self.x.dtype)), atol=config.ATOL.get(str(self.x.dtype)), ) def test_inverse(self): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.IndependentTransform( self.base, self.reinterpreted_batch_rank ) static_input = paddle.static.data( 'input', self.x.shape, self.x.dtype ) output = t.inverse(static_input) expected = self.base.inverse(static_input) exe.run(sp) [output, expected] = exe.run( mp, feed={'input': self.x}, fetch_list=[output, expected] ) np.testing.assert_allclose( expected, output, rtol=config.RTOL.get(str(self.x.dtype)), atol=config.ATOL.get(str(self.x.dtype)), ) def test_forward_log_det_jacobian(self): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.IndependentTransform( self.base, self.reinterpreted_batch_rank ) static_input = paddle.static.data( 'input', self.x.shape, self.x.dtype ) output = t.forward_log_det_jacobian(static_input) expected = self.base.forward_log_det_jacobian( static_input.sum(list(range(-self.reinterpreted_batch_rank, 0))) ) exe.run(sp) [actual, expected] = exe.run( mp, feed={'input': self.x}, fetch_list=[output, expected] ) self.assertEqual( tuple(actual.shape), self.x.shape[: -self.reinterpreted_batch_rank] ) np.testing.assert_allclose( actual, expected, rtol=config.RTOL.get(str(self.x.dtype)), atol=config.ATOL.get(str(self.x.dtype)), ) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_forward_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_inverse_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.place(config.DEVICES) class TestPowerTransform(unittest.TestCase): def setUp(self): self._t = transform.PowerTransform(paddle.full([1], 2.0)) def test_init(self): with self.assertRaises(TypeError): transform.PowerTransform(1.0) def test_is_injective(self): self.assertTrue(self._t._is_injective()) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Real)) self.assertEqual(self._t._domain.event_rank, 0) self.assertEqual(self._t._domain.is_discrete, False) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Positive)) self.assertEqual(self._t._codomain.event_rank, 0) self.assertEqual(self._t._codomain.is_discrete, False) @param.param_func( [ ( np.array([2.0]), np.array([0.0, -1.0, 2.0]), np.power(np.array([0.0, -1.0, 2.0]), 2.0), ), ( np.array([[0.0], [3.0]]), np.array([[1.0, 0.0], [5.0, 6.0]]), np.power( np.array([[1.0, 0.0], [5.0, 6.0]]), np.array([[0.0], [3.0]]) ), ), ] ) def test_forward(self, power, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): static_power = paddle.static.data('power', power.shape, power.dtype) t = transform.PowerTransform(static_power) static_input = paddle.static.data('input', input.shape, input.dtype) output = t.forward(static_input) exe.run(sp) [output] = exe.run( mp, feed={'input': input, 'power': power}, fetch_list=[output] ) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func([(np.array([2.0]), np.array([4.0]), np.array([2.0]))]) def test_inverse(self, power, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): static_power = paddle.static.data('power', power.shape, power.dtype) t = transform.PowerTransform(static_power) static_input = paddle.static.data('input', input.shape, input.dtype) output = t.inverse(static_input) exe.run(sp) [output] = exe.run( mp, feed={'input': input, 'power': power}, fetch_list=[output] ) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func(((np.array([2.0]), np.array([3.0, 1.4, 0.8])),)) def test_forward_log_det_jacobian(self, power, input): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): static_power = paddle.static.data('power', power.shape, power.dtype) t = transform.PowerTransform(static_power) static_input = paddle.static.data('input', input.shape, input.dtype) output = t.forward_log_det_jacobian(static_input) exe.run(sp) [output] = exe.run( mp, feed={'input': input, 'power': power}, fetch_list=[output] ) np.testing.assert_allclose( output, self._np_forward_jacobian(power, input), rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) def _np_forward_jacobian(self, alpha, x): return np.abs(np.log(alpha * np.power(x, alpha - 1))) @param.param_func([((2, 3, 5), (2, 3, 5))]) def test_forward_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.param_func([((2, 3, 5), (2, 3, 5))]) def test_inverse_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.place(config.DEVICES) class TestTanhTransform(unittest.TestCase): def setUp(self): self._t = transform.TanhTransform() def test_is_injective(self): self.assertTrue(self._t._is_injective()) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Real)) self.assertEqual(self._t._domain.event_rank, 0) self.assertEqual(self._t._domain.is_discrete, False) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Variable)) self.assertEqual(self._t._codomain.event_rank, 0) self.assertEqual(self._t._codomain.is_discrete, False) self.assertEqual(self._t._codomain._constraint._lower, -1) self.assertEqual(self._t._codomain._constraint._upper, 1) @param.param_func( [ ( np.array([0.0, 1.0, 2.0, 3.0]), np.tanh(np.array([0.0, 1.0, 2.0, 3.0])), ), ( np.array([[0.0, 1.0, 2.0, 3.0], [-5.0, 6.0, 7.0, 8.0]]), np.tanh( np.array([[0.0, 1.0, 2.0, 3.0], [-5.0, 6.0, 7.0, 8.0]]) ), ), ] ) def test_forward(self, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.TanhTransform() static_input = paddle.static.data('input', input.shape, input.dtype) output = t.forward(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func( [ (np.array([1.0, 2.0, 3.0]), np.arctanh(np.array([1.0, 2.0, 3.0]))), ( np.array([[1.0, 2.0, 3.0], [6.0, 7.0, 8.0]]), np.arctanh(np.array([[1.0, 2.0, 3.0], [6.0, 7.0, 8.0]])), ), ] ) def test_inverse(self, input, expected): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.TanhTransform() static_input = paddle.static.data('input', input.shape, input.dtype) output = t.inverse(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func( [ (np.array([1.0, 2.0, 3.0]),), (np.array([[1.0, 2.0, 3.0], [6.0, 7.0, 8.0]]),), ] ) def test_forward_log_det_jacobian(self, input): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.TanhTransform() static_input = paddle.static.data('input', input.shape, input.dtype) output = t.forward_log_det_jacobian(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, self._np_forward_jacobian(input), rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) def _np_forward_jacobian(self, x): return 2.0 * (np.log(2.0) - x - self._np_softplus(-2.0 * x)) def _np_softplus(self, x, beta=1.0, threshold=20.0): if np.any(beta * x > threshold): return x return 1.0 / beta * np.log1p(np.exp(beta * x)) def _np_inverse_jacobian(self, y): return -self._np_forward_jacobian(np.arctanh(y)) @param.param_func( [ (np.array([1.0, 2.0, 3.0]),), (np.array([[1.0, 2.0, 3.0], [6.0, 7.0, 8.0]]),), ] ) def test_inverse_log_det_jacobian(self, input): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): t = transform.TanhTransform() static_input = paddle.static.data('input', input.shape, input.dtype) output = t.inverse_log_det_jacobian(static_input) exe.run(sp) [output] = exe.run(mp, feed={'input': input}, fetch_list=[output]) np.testing.assert_allclose( output, self._np_inverse_jacobian(input), rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_forward_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_inverse_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.place(config.DEVICES) @param.param_cls( (param.TEST_CASE_NAME, 'in_event_shape', 'out_event_shape'), [ ('regular_shape', (2, 3), (3, 2)), ], ) class TestReshapeTransform(unittest.TestCase): def setUp(self): self._t = transform.ReshapeTransform( self.in_event_shape, self.out_event_shape ) def test_is_injective(self): self.assertTrue(self._t._is_injective()) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Independent)) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Independent)) def test_forward(self): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): x = paddle.ones(self.in_event_shape) t = transform.ReshapeTransform( self.in_event_shape, self.out_event_shape ) output = self._t.forward(x) exe.run(sp) [output] = exe.run(mp, feed={}, fetch_list=[output]) expected = np.ones(self.out_event_shape) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(expected.dtype)), atol=config.ATOL.get(str(expected.dtype)), ) def test_inverse(self): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): x = paddle.ones(self.out_event_shape) t = transform.ReshapeTransform( self.in_event_shape, self.out_event_shape ) output = self._t.inverse(x) exe.run(sp) [output] = exe.run(mp, feed={}, fetch_list=[output]) expected = np.ones(self.in_event_shape) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(expected.dtype)), atol=config.ATOL.get(str(expected.dtype)), ) def test_forward_log_det_jacobian(self): exe = paddle.static.Executor() sp = paddle.static.Program() mp = paddle.static.Program() with paddle.static.program_guard(mp, sp): x = paddle.ones(self.in_event_shape) t = transform.ReshapeTransform( self.in_event_shape, self.out_event_shape ) output = self._t.forward_log_det_jacobian(x) exe.run(sp) [output] = exe.run(mp, feed={}, fetch_list=[output]) expected = np.zeros([1]) np.testing.assert_allclose( output, expected, rtol=config.RTOL.get(str(expected.dtype)), atol=config.ATOL.get(str(expected.dtype)), ) def _np_softplus(x, beta=1.0, threshold=20.0): if np.any(beta * x > threshold): return x return 1.0 / beta * np.log1p(np.exp(beta * x)) class TestSigmoidTransform(unittest.TestCase): def setUp(self): self._t = transform.SigmoidTransform() def test_is_injective(self): self.assertTrue(self._t._is_injective()) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Real)) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Variable)) @param.param_func( ((np.ones((5, 10)), 1 / (1 + np.exp(-np.ones((5, 10))))),) ) def test_forward(self, input, expected): with paddle.static.program_guard(paddle.static.Program()): x = paddle.static.data('X', input.shape, input.dtype) model = transform.SigmoidTransform() out = model.forward(x) place = ( paddle.CUDAPlace(0) if paddle.core.is_compiled_with_cuda() else paddle.CPUPlace() ) exe = paddle.static.Executor(place) (result,) = exe.run(feed={'X': input}, fetch_list=[out]) np.testing.assert_allclose( result, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func( ((np.ones(10), np.log(np.ones(10)) - np.log1p(-np.ones(10))),) ) def test_inverse(self, input, expected): with paddle.static.program_guard(paddle.static.Program()): x = paddle.static.data('X', input.shape, input.dtype) model = transform.SigmoidTransform() out = model.inverse(x) place = ( paddle.CUDAPlace(0) if paddle.core.is_compiled_with_cuda() else paddle.CPUPlace() ) exe = paddle.static.Executor(place) (result,) = exe.run(feed={'X': input}, fetch_list=[out]) np.testing.assert_allclose( result, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func( ( ( np.ones(10), -_np_softplus(-np.ones(10)) - _np_softplus(np.ones(10)), ), ) ) def test_forward_log_det_jacobian(self, input, expected): with paddle.static.program_guard(paddle.static.Program()): x = paddle.static.data('X', input.shape, input.dtype) model = transform.SigmoidTransform() out = model.forward_log_det_jacobian(x) place = ( paddle.CUDAPlace(0) if paddle.core.is_compiled_with_cuda() else paddle.CPUPlace() ) exe = paddle.static.Executor(place) (result,) = exe.run(feed={'X': input}, fetch_list=[out]) np.testing.assert_allclose( result, expected, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_forward_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_inverse_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.param_func([(np.array(1.0), np.array(1.0))]) def test_zerodim(self, input, expected): shape = () if paddle.framework.in_pir_mode(): shape = [] with paddle.static.program_guard(paddle.static.Program()): x = paddle.static.data('X', input.shape, 'float32') model = transform.SigmoidTransform() self.assertEqual(model.forward(x).shape, shape) self.assertEqual(model.inverse(x).shape, shape) self.assertEqual(model.forward_log_det_jacobian(x).shape, shape) self.assertEqual(model.inverse_log_det_jacobian(x).shape, shape) self.assertEqual(model.forward_shape(x.shape), shape) self.assertEqual(model.inverse_shape(x.shape), shape) class TestStickBreakingTransform(unittest.TestCase): def setUp(self): self._t = transform.StickBreakingTransform() def test_is_injective(self): self.assertTrue(self._t._is_injective()) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Independent)) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Variable)) @param.param_func(((np.random.random(10),),)) def test_forward(self, input): with paddle.static.program_guard(paddle.static.Program()): x = paddle.static.data('X', input.shape, input.dtype) model = transform.StickBreakingTransform() fwd = model.forward(x) out = model.inverse(fwd) place = ( paddle.CUDAPlace(0) if paddle.core.is_compiled_with_cuda() else paddle.CPUPlace() ) exe = paddle.static.Executor(place) (result,) = exe.run(feed={'X': input}, fetch_list=[out]) np.testing.assert_allclose( result, input, rtol=config.RTOL.get(str(input.dtype)), atol=config.ATOL.get(str(input.dtype)), ) @param.param_func([((2, 3, 5), (2, 3, 6))]) def test_forward_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.param_func([((2, 3, 5), (2, 3, 4))]) def test_inverse_shape(self, shape, expected_shape): self.assertEqual(self._t.inverse_shape(shape), expected_shape) @param.param_func(((np.random.random(10),),)) def test_forward_log_det_jacobian(self, input): with paddle.static.program_guard(paddle.static.Program()): x = paddle.static.data('X', input.shape, input.dtype) model = transform.StickBreakingTransform() out = model.forward_log_det_jacobian(x) place = ( paddle.CUDAPlace(0) if paddle.core.is_compiled_with_cuda() else paddle.CPUPlace() ) exe = paddle.static.Executor(place) (result,) = exe.run(feed={'X': input}, fetch_list=[out]) self.assertEqual(result.shape, ()) @param.place(config.DEVICES) @param.param_cls( (param.TEST_CASE_NAME, 'transforms', 'axis'), [ ('simple_one_transform', [transform.ExpTransform()], 0), ], ) class TestStackTransform(unittest.TestCase): def setUp(self): self._t = transform.StackTransform(self.transforms, self.axis) def test_is_injective(self): self.assertTrue(self._t._is_injective()) def test_domain(self): self.assertTrue(isinstance(self._t._domain, variable.Stack)) def test_codomain(self): self.assertTrue(isinstance(self._t._codomain, variable.Stack)) @param.param_func( [ (np.array([[0.0, 1.0, 2.0, 3.0]]),), (np.array([[-5.0, 6.0, 7.0, 8.0]]),), ] ) def test_forward(self, input): with paddle.static.program_guard(paddle.static.Program()): x = paddle.static.data('X', input.shape, input.dtype) model = transform.StackTransform(self.transforms, self.axis) out = model.forward(x) place = ( paddle.CUDAPlace(0) if paddle.core.is_compiled_with_cuda() else paddle.CPUPlace() ) exe = paddle.static.Executor(place) (result,) = exe.run(feed={'X': input}, fetch_list=[out]) self.assertEqual(tuple(result.shape), input.shape) @param.param_func( [ (np.array([[1.0, 2.0, 3.0]]),), ( np.array( [[6.0, 7.0, 8.0]], ), ), ] ) def test_inverse(self, input): with paddle.static.program_guard(paddle.static.Program()): x = paddle.static.data('X', input.shape, input.dtype) model = transform.StackTransform(self.transforms, self.axis) out = model.inverse(x) place = ( paddle.CUDAPlace(0) if paddle.core.is_compiled_with_cuda() else paddle.CPUPlace() ) exe = paddle.static.Executor(place) (result,) = exe.run(feed={'X': input}, fetch_list=[out]) self.assertEqual(tuple(result.shape), input.shape) @param.param_func( [(np.array([[1.0, 2.0, 3.0]]),), (np.array([[6.0, 7.0, 8.0]]),)] ) def test_forward_log_det_jacobian(self, input): with paddle.static.program_guard(paddle.static.Program()): x = paddle.static.data('X', input.shape, input.dtype) model = transform.StackTransform(self.transforms, self.axis) out = model.forward_log_det_jacobian(x) place = ( paddle.CUDAPlace(0) if paddle.core.is_compiled_with_cuda() else paddle.CPUPlace() ) exe = paddle.static.Executor(place) (result,) = exe.run(feed={'X': input}, fetch_list=[out]) self.assertEqual(tuple(result.shape), input.shape) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_forward_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) @param.param_func([((), ()), ((2, 3, 5), (2, 3, 5))]) def test_inverse_shape(self, shape, expected_shape): self.assertEqual(self._t.forward_shape(shape), expected_shape) def test_axis(self): self.assertEqual(self._t.axis, self.axis) @param.param_func( [ (0, 0, TypeError), ([0], 0, TypeError), ([paddle.distribution.ExpTransform()], 'axis', TypeError), ] ) def test_init_exception(self, transforms, axis, exc): with self.assertRaises(exc): paddle.distribution.StackTransform(transforms, axis) def test_transforms(self): self.assertIsInstance((self._t.transforms), typing.Sequence) if __name__ == '__main__': unittest.main()