/
test_corr.py
122 lines (97 loc) · 3.82 KB
/
test_corr.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
# 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 paddle.fluid as fluid
import unittest
import numpy as np
import six
import paddle
import warnings
def numpy_corr(np_arr, rowvar=True, dtype='float64'):
return np.corrcoef(np_arr, rowvar=rowvar, dtype=dtype)
class Corr_Test(unittest.TestCase):
def setUp(self):
self.shape = [4, 5]
def test_tensor_corr_default(self):
typelist = ['float64', 'float32']
places = [fluid.CPUPlace()]
if fluid.core.is_compiled_with_cuda():
places.append(fluid.CUDAPlace(0))
for idx, p in enumerate(places):
if idx == 0:
paddle.set_device('cpu')
else:
paddle.set_device('gpu')
for dtype in typelist:
np_arr = np.random.rand(*self.shape).astype(dtype)
tensor = paddle.to_tensor(np_arr, place=p)
corr = paddle.linalg.corrcoef(tensor)
np_corr = numpy_corr(np_arr, rowvar=True, dtype=dtype)
if dtype == 'float32':
self.assertTrue(
np.allclose(
np_corr, corr.numpy(), atol=1.e-5))
else:
self.assertTrue(np.allclose(np_corr, corr.numpy()))
def test_tensor_corr_rowvar(self):
typelist = ['float64', 'float32']
places = [fluid.CPUPlace()]
if fluid.core.is_compiled_with_cuda():
places.append(fluid.CUDAPlace(0))
for idx, p in enumerate(places):
if idx == 0:
paddle.set_device('cpu')
else:
paddle.set_device('gpu')
for dtype in typelist:
np_arr = np.random.rand(*self.shape).astype(dtype)
tensor = paddle.to_tensor(np_arr, place=p)
corr = paddle.linalg.corrcoef(tensor, rowvar=False)
np_corr = numpy_corr(np_arr, rowvar=False, dtype=dtype)
if dtype == 'float32':
self.assertTrue(
np.allclose(
np_corr, corr.numpy(), atol=1.e-5))
else:
self.assertTrue(np.allclose(np_corr, corr.numpy()))
# Input(x) only support N-D (1<=N<=2) tensor
class Corr_Test2(Corr_Test):
def setUp(self):
self.shape = [10]
class Corr_Test3(Corr_Test):
def setUp(self):
self.shape = [4, 5]
# Input(x) only support N-D (1<=N<=2) tensor
class Corr_Test4(unittest.TestCase):
def setUp(self):
self.shape = [2, 5, 2]
def test_errors(self):
def test_err():
np_arr = np.random.rand(*self.shape).astype('float64')
tensor = paddle.to_tensor(np_arr)
covrr = paddle.linalg.corrcoef(tensor)
self.assertRaises(ValueError, test_err)
# test unsupported complex input
class Corr_Comeplex_Test(unittest.TestCase):
def setUp(self):
self.dtype = 'complex128'
def test_errors(self):
paddle.enable_static()
x1 = fluid.data(name=self.dtype, shape=[2], dtype=self.dtype)
self.assertRaises(TypeError, paddle.linalg.corrcoef, x=x1)
paddle.disable_static()
class Corr_Test5(Corr_Comeplex_Test):
def setUp(self):
self.dtype = 'complex64'
if __name__ == '__main__':
unittest.main()